《互联网产品的 AI 架构·多能力编排与任务闭环》10.1 编排链路的最低观测字段

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

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

export const TRACE_SCHEMA_VERSION = 1;
export type TracePhase = "start" | "end";
export type NodeKind = "workflow" | "capability" | "human" | "action";
export type TraceStatus = "ok" | "degraded" | "error" | "skipped";

/**
 * 一次编排 trace 的最小事件。start 与 end 成对出现：
 * 身份与版本在 start 采集，结果与代价在 end 采集。
 */
export interface WorkflowTraceEvent {
  schemaVersion: number;
  phase: TracePhase;
  workflowId: string;
  traceId: string;
  spanId: string;
  parentSpanId: string | null;
  nodeId: string;
  nodeKind: NodeKind;
  attempt: number;
  capability: string;
  modelId: string | null;
  promptVersion: string | null;
  /** 上游交接令牌；根节点为 null，用于把下游结果与上游产物版本对齐。 */
  handoffToken: string | null;
  /** 本次交接携带的 token 数；根节点为 0，三能力样本为 820。 */
  handoffTokens: number;
  /** 相对 workflow 起点的开始时间，毫秒。 */
  startOffsetMs: number;
  /** 仅 end：节点自身耗时，毫秒。 */
  durationMs: number;
  /** 仅 end：节点计费，整数微元；start 固定为 0。 */
  costMicros: number;
  status: TraceStatus;
  errorCode: AiErrorKind | null;
  degraded: boolean;
  degradedReason: string | null;
}

/** 这些字段无法事后补采，必须由节点边界上的埋点直接写出。 */
export const BOUNDARY_FIELDS: Record<TracePhase, readonly (keyof WorkflowTraceEvent)[]> = {
  start: ["schemaVersion", "workflowId", "traceId", "spanId", "parentSpanId", "nodeId",
    "nodeKind", "attempt", "capability", "modelId", "promptVersion",
    "handoffToken", "handoffTokens", "startOffsetMs"],
  end: ["spanId", "durationMs", "costMicros", "status", "errorCode", "degraded", "degradedReason"],
};

export interface TraceIssue {
  path: string;
  code: string;
  message: string;
}

const KINDS: readonly NodeKind[] = ["workflow", "capability", "human", "action"];
const STATUSES: readonly TraceStatus[] = ["ok", "degraded", "error", "skipped"];
const ID_FIELDS = ["workflowId", "traceId", "spanId", "nodeId", "capability"] as const;

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function text(value: unknown): boolean {
  return typeof value === "string" && value.trim() !== "";
}

function count(value: unknown, min = 0): boolean {
  return typeof value === "number" && Number.isInteger(value) && value >= min;
}

function push(issues: TraceIssue[], path: string, code: string, message: string): void {
  issues.push({ path, code, message });
}

/** 边界校验返回问题列表而不抛异常：坏事件可以同时进入告警与拒收统计。 */
export function validateTraceEvent(value: unknown): TraceIssue[] {
  if (!isRecord(value)) return [{ path: "$", code: "not-object", message: "event 必须是对象" }];
  const issues: TraceIssue[] = [];
  if (value.schemaVersion !== TRACE_SCHEMA_VERSION) push(issues, "schemaVersion", "bad-version", `必须等于 ${TRACE_SCHEMA_VERSION}`);
  if (value.phase !== "start" && value.phase !== "end") push(issues, "phase", "bad-phase", "必须是 start 或 end");
  for (const field of ID_FIELDS) if (!text(value[field])) push(issues, field, "empty", "必须是非空字符串");
  if (value.parentSpanId !== null && !text(value.parentSpanId)) push(issues, "parentSpanId", "bad-parent", "必须是 null 或非空字符串");
  if (!KINDS.includes(value.nodeKind as NodeKind)) push(issues, "nodeKind", "bad-kind", "未知节点类型");
  if (!count(value.attempt, 1)) push(issues, "attempt", "bad-attempt", "必须是 >= 1 的整数");
  if (!count(value.startOffsetMs)) push(issues, "startOffsetMs", "bad-offset", "必须是非负整数");
  if (!count(value.handoffTokens)) push(issues, "handoffTokens", "bad-handoff", "必须是非负整数");
  if (value.parentSpanId === null) {
    if (value.handoffTokens !== 0) push(issues, "handoffTokens", "root-handoff", "根节点不得携带交接 token");
    if (value.handoffToken !== null) push(issues, "handoffToken", "root-handoff", "根节点交接令牌必须为 null");
  } else {
    if (!text(value.handoffToken)) push(issues, "handoffToken", "missing-handoff", "非根节点必须有交接令牌");
    if (!count(value.handoffTokens, 1)) push(issues, "handoffTokens", "missing-handoff", "非根节点交接 token 必须 >= 1");
  }
  if (value.nodeKind === "capability") {
    if (!text(value.modelId)) push(issues, "modelId", "missing-model", "能力节点必须记录模型版本");
    if (!text(value.promptVersion)) push(issues, "promptVersion", "missing-prompt", "能力节点必须记录提示版本");
  } else if (value.nodeKind === "human" && (value.modelId !== null || value.promptVersion !== null)) {
    push(issues, "modelId", "human-model", "人工节点不应携带模型或提示版本");
  }
  if (!count(value.durationMs)) push(issues, "durationMs", "bad-duration", "必须是非负整数");
  if (!count(value.costMicros)) push(issues, "costMicros", "bad-cost", "必须是非负整数");
  if (!STATUSES.includes(value.status as TraceStatus)) push(issues, "status", "bad-status", "未知终态");
  if (value.status === "error" && value.errorCode === null) push(issues, "errorCode", "missing-error", "error 必须带错误码");
  if (value.status !== "error" && value.errorCode !== null) push(issues, "errorCode", "unexpected-error", "非 error 不得带错误码");
  if (value.degraded !== (value.status === "degraded")) push(issues, "degraded", "degrade-mismatch", "degraded 必须与 status 一致");
  if (value.status === "degraded" && !text(value.degradedReason)) push(issues, "degradedReason", "missing-degrade", "降级必须写明原因");
  if (value.status !== "degraded" && value.degradedReason !== null) push(issues, "degradedReason", "unexpected-degrade", "非降级不得带降级原因");
  if (value.phase === "start") {
    if (value.durationMs !== 0 || value.costMicros !== 0) push(issues, "$", "start-has-result", "start 事件不得包含耗时或成本");
    if (value.status !== "ok" || value.errorCode !== null || value.degraded) push(issues, "$", "start-has-outcome", "start 事件不得包含终态");
  }
  return issues;
}

export interface TraceNode extends WorkflowTraceEvent { finishOffsetMs: number; }
export interface CriticalStep { spanId: string; nodeId: string; attempt: number; finishOffsetMs: number; }
export interface TraceSummary {
  workflowId: string; traceId: string; spans: number; attempts: number; retries: number;
  failures: number; degraded: number; totalCostMicros: number; endOffsetMs: number;
  rootSpanId: string; nodes: TraceNode[]; children: Record<string, string[]>;
  criticalPath: CriticalStep[]; criticalPathMs: number; busyMs: number; handoffMs: number;
}

/** 值为 null 也算已采集；缺键或 undefined 才算缺失。 */
export function missingBoundaryFields(event: WorkflowTraceEvent): string[] {
  return BOUNDARY_FIELDS[event.phase].filter(
    (field) => !(field in event) || event[field] === undefined,
  );
}

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

function assertTrace(events: readonly WorkflowTraceEvent[]): void {
  if (events.length === 0) fail("trace 至少需要一条事件");
  const issues = events.flatMap((event, index) =>
    validateTraceEvent(event).map((issue) => ({ ...issue, path: `events[${index}].${issue.path}` })),
  );
  if (issues.length > 0) fail(issues.map((issue) => `${issue.path}: ${issue.message}`).join("; "));

  const starts = new Map<string, WorkflowTraceEvent>();
  const ends = new Map<string, WorkflowTraceEvent>();
  for (const event of events) {
    const target = event.phase === "start" ? starts : ends;
    if (target.has(event.spanId)) fail(`span ${event.spanId} 的 ${event.phase} 事件重复`);
    target.set(event.spanId, event);
  }
  for (const spanId of starts.keys()) {
    if (!ends.has(spanId)) fail(`span ${spanId} 缺少 end 事件`);
  }
  if (starts.size !== ends.size) fail(`start 与 end 数量不一致：${starts.size} vs ${ends.size}`);
  const roots = [...starts.values()].filter((event) => event.parentSpanId === null);
  if (roots.length !== 1) fail(`trace 必须恰好有一个根 span，实际 ${roots.length}`);
  if (roots[0].nodeKind !== "workflow") fail("根 span 必须是 workflow 节点");

  for (const event of starts.values()) {
    if (event.parentSpanId === null) continue;
    if (event.parentSpanId === event.spanId) fail(`span ${event.spanId} 指向自己`);
    const parent = starts.get(event.parentSpanId);
    if (!parent) fail(`span ${event.spanId} 的父 span ${event.parentSpanId} 不存在`);
    if (event.startOffsetMs < parent.startOffsetMs) fail(`span ${event.spanId} 早于父 span 开始`);
    const chain = new Set<string>([event.spanId]);
    let cursor: WorkflowTraceEvent | undefined = event;
    while (cursor.parentSpanId !== null) {
      if (chain.has(cursor.parentSpanId)) fail(`span ${event.spanId} 的父链存在环`);
      chain.add(cursor.parentSpanId);
      cursor = starts.get(cursor.parentSpanId);
    }
  }
}

/**
 * 从事件重建节点树与关键路径。关键路径沿“父链上最后结束的孩子”下探，
 * 返回的是墙钟路径，包含重试空档与交接延迟。
 */
export function rebuildTrace(events: readonly WorkflowTraceEvent[]): TraceSummary {
  assertTrace(events);
  const starts = new Map<string, WorkflowTraceEvent>();
  const ends = new Map<string, WorkflowTraceEvent>();
  for (const event of events) {
    (event.phase === "start" ? starts : ends).set(event.spanId, event);
  }
  const bySpan = new Map<string, TraceNode>();
  for (const [spanId, start] of starts) {
    const end = ends.get(spanId) as WorkflowTraceEvent;
    bySpan.set(spanId, { ...start, ...end, finishOffsetMs: start.startOffsetMs + end.durationMs });
  }
  const nodes = [...bySpan.values()].sort(
    (left, right) => left.startOffsetMs - right.startOffsetMs || left.spanId.localeCompare(right.spanId),
  );
  const children: Record<string, string[]> = {};
  for (const node of nodes) {
    if (node.parentSpanId !== null) (children[node.parentSpanId] ??= []).push(node.spanId);
  }
  for (const list of Object.values(children)) {
    list.sort((left, right) => (bySpan.get(left) as TraceNode).startOffsetMs - (bySpan.get(right) as TraceNode).startOffsetMs);
  }

  const root = nodes.find((node) => node.parentSpanId === null) as TraceNode;
  const path: TraceNode[] = [];
  let cursor: TraceNode | undefined = root;
  while (cursor) {
    path.push(cursor);
    cursor = (children[cursor.spanId] ?? [])
      .map((spanId) => bySpan.get(spanId) as TraceNode)
      .sort((left, right) => right.finishOffsetMs - left.finishOffsetMs || left.spanId.localeCompare(right.spanId))[0];
  }
  const body = path.filter((node) => node.spanId !== root.spanId);
  const endsList = [...ends.values()];
  const criticalPathMs = (body.at(-1)?.finishOffsetMs ?? root.finishOffsetMs) - root.startOffsetMs;
  const busyMs = body.reduce((total, node) => total + node.durationMs, 0);
  return {
    workflowId: root.workflowId,
    traceId: root.traceId,
    spans: nodes.length,
    attempts: endsList.length,
    retries: endsList.filter((event) => event.attempt > 1).length,
    failures: endsList.filter((event) => event.status === "error").length,
    degraded: endsList.filter((event) => event.status === "degraded").length,
    totalCostMicros: endsList
      .filter((event) => event.nodeKind !== "workflow")
      .reduce((total, event) => total + event.costMicros, 0),
    endOffsetMs: Math.max(...endsList.map((event) => (bySpan.get(event.spanId) as TraceNode).finishOffsetMs)),
    rootSpanId: root.spanId,
    nodes,
    children,
    criticalPath: body.map((node) => ({
      spanId: node.spanId,
      nodeId: node.nodeId,
      attempt: node.attempt,
      finishOffsetMs: node.finishOffsetMs,
    })),
    criticalPathMs,
    busyMs,
    handoffMs: criticalPathMs - busyMs,
  };
}