《互联网产品的 AI 架构·模型接入与能力边界》3.1 模型网关的职责边界

// file: src/gateway/pipeline.ts
// tier: L1
// verify: node --test test/gateway/pipeline.test.ts

import { AiError, normalizeError } from "../core/errors.ts";
import { systemClock } from "../core/types.ts";
import type { Clock } from "../core/types.ts";

/**
 * 阶段只有两种结局：继续，或者到此为止。
 * 缓存命中、模板兜底这类“已经拿到结果”的路径用 stop，后续阶段不再执行，
 * 但收尾钩子仍然会跑，审计不会因为短路而丢失。
 */
export interface StageResult<C> {
  readonly done: boolean;
  readonly context: C;
  readonly reason?: string;
}

export function proceed<C>(context: C): StageResult<C> {
  return { done: false, context };
}

export function stop<C>(context: C, reason: string): StageResult<C> {
  return { done: true, context, reason };
}

export interface Stage<C> {
  readonly name: string;
  run(context: C): StageResult<C> | Promise<StageResult<C>>;
}

export type SpanStatus = "ok" | "halted" | "failed";

/** 每个阶段一条耗时记录，名称固定，便于做指标聚合。 */
export interface StageSpan {
  name: string;
  durationMs: number;
  status: SpanStatus;
  reason?: string;
}

export interface PipelineOutcome<C> {
  /** 最后一个成功阶段产出的上下文；失败时也保留，供审计使用。 */
  context: C;
  error?: AiError;
}

export type Finalizer<C> = (outcome: PipelineOutcome<C>) => void | Promise<void>;

export interface PipelineRun<C> {
  context: C;
  spans: StageSpan[];
  haltedBy?: string;
}

export interface PipelineOptions<C> {
  clock?: Clock;
  /** 返回距离截止时刻还剩多少毫秒；undefined 表示本次不做截止检查。 */
  remainingMs?: (context: C) => number | undefined;
}

export interface Pipeline<C> {
  readonly name: string;
  use(stage: Stage<C>): Pipeline<C>;
  onFinish(finalizer: Finalizer<C>): Pipeline<C>;
  stageNames(): string[];
  run(initial: C): Promise<PipelineRun<C>>;
}

/**
 * 通用阶段组合器：只负责顺序、短路、耗时与错误归一，
 * 不关心认证、配额、路由的具体实现。网关之外的调用链也能复用。
 */
export function createPipeline<C>(
  name: string,
  options: PipelineOptions<C> = {},
): Pipeline<C> {
  if (name.trim() === "") {
    throw new Error("pipeline: name 不能为空");
  }

  const clock = options.clock ?? systemClock;
  const stages: Stage<C>[] = [];
  const finalizers: Finalizer<C>[] = [];
  const names = new Set<string>();

  return {
    name,

    use(stage: Stage<C>): Pipeline<C> {
      if (stage.name.trim() === "") {
        throw new Error("pipeline: stage.name 不能为空");
      }
      if (names.has(stage.name)) {
        throw new Error(`pipeline ${name}: 阶段名重复 ${stage.name}`);
      }
      names.add(stage.name);
      stages.push(stage);
      return this;
    },

    onFinish(finalizer: Finalizer<C>): Pipeline<C> {
      finalizers.push(finalizer);
      return this;
    },

    stageNames(): string[] {
      return [...names];
    },

    async run(initial: C): Promise<PipelineRun<C>> {
      let context = initial;
      const spans: StageSpan[] = [];
      let haltedBy: string | undefined;
      let failure: AiError | undefined;

      for (const stage of stages) {
        const remaining = options.remainingMs?.(context);
        if (remaining !== undefined && remaining <= 0) {
          failure = new AiError(
            "timeout",
            `pipeline ${name}: 进入阶段 ${stage.name} 前已超出截止时间`,
          );
          break;
        }

        const startedAt = clock.now();
        try {
          const result = await stage.run(context);
          context = result.context;
          const span: StageSpan = {
            name: stage.name,
            durationMs: Math.max(0, clock.now() - startedAt),
            status: result.done ? "halted" : "ok",
          };
          if (result.reason !== undefined) {
            span.reason = result.reason;
          }
          spans.push(span);
          if (result.done) {
            haltedBy = stage.name;
            break;
          }
        } catch (error) {
          spans.push({
            name: stage.name,
            durationMs: Math.max(0, clock.now() - startedAt),
            status: "failed",
          });
          failure = normalizeError(error);
          break;
        }
      }

      const finalizerErrors: AiError[] = [];
      for (const finalizer of [...finalizers].reverse()) {
        try {
          await finalizer({ context, ...(failure ? { error: failure } : {}) });
        } catch (error) {
          finalizerErrors.push(normalizeError(error));
        }
      }

      // 主流程的错误优先返回；收尾钩子的问题不能把真正的失败原因盖掉。
      if (failure) {
        throw failure;
      }
      if (finalizerErrors.length > 0) {
        throw finalizerErrors[0];
      }

      return haltedBy === undefined
        ? { context, spans }
        : { context, spans, haltedBy };
    },
  };
}

node --test test/gateway/pipeline.test.ts

✔ 阶段按注册顺序执行并传递上下文 (1.15125ms)
✔ 同名阶段在注册时被拒绝 (0.267167ms)
✔ stop 跳过后继阶段，但收尾钩子仍然执行 (0.154458ms)
✔ 阶段异常被归一为 AiError，并记录失败阶段 (0.230167ms)
✔ 截止时间耗尽时，下一阶段不再执行并抛 timeout (0.144917ms)
✔ 注入时钟记录每个阶段的耗时 (0.132833ms)
✔ 失败时收尾钩子拿到最后一个成功上下文与原始错误 (0.150125ms)
✔ 收尾钩子抛错不能掩盖主流程错误 (0.134083ms)
✔ 收尾钩子自身的错误在没有主错误时会抛出 (0.153083ms)
ℹ tests 9
ℹ suites 0
ℹ pass 9
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 161.745584