《互联网产品的 AI 架构·多能力编排与任务闭环》2.3 编排层的输入输出与版本

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

import { AiError } from "../core/errors.ts";
import { checkVersionCompatibility, parseVersion } from "./handoff.ts";

export type RunStatus = "succeeded" | "partial" | "failed" | "cancelled";

export type OrchestrationErrorCode =
  | "invalid-envelope"
  | "contract-mismatch"
  | "node-failed"
  | "partial-result"
  | "budget-exceeded"
  | "deadline-exceeded"
  | "cancelled"
  | "internal";

export interface ContextReference {
  id: string;
  revision: string;
}

/** 编排层对外唯一入口：业务方只描述意图、预算与上下文引用。 */
export interface OrchestrationRequest {
  contractVersion: string;
  runId: string;
  tenantId: string;
  intent: string;
  input: Record<string, unknown>;
  contextRefs: readonly ContextReference[];
  deadlineMs: number;
  maxCostMicros: number;
  mode: "copy" | "pass" | "share";
  idempotencyKey: string;
}

export interface NodeOutcome {
  nodeId: string;
  capability: string;
  status: "ok" | "degraded" | "failed" | "skipped";
  costMicros: number;
  latencyMs: number;
  errorCode?: OrchestrationErrorCode;
}

export interface OrchestrationResponse {
  contractVersion: string;
  runId: string;
  status: RunStatus;
  output: Record<string, unknown>;
  nodes: readonly NodeOutcome[];
  usage: { inputTokens: number; outputTokens: number; costMicros: number };
  cursor?: { step: string; offset: number };
  error?: { code: OrchestrationErrorCode; retryable: boolean; message: string };
  idempotencyKey: string;
}

export interface ErrorPolicy {
  httpStatus: number;
  retryable: boolean;
  /** 该错误是否允许携带部分结果返回。 */
  partialAllowed: boolean;
}

export const ERROR_POLICY: Record<OrchestrationErrorCode, ErrorPolicy> = {
  "invalid-envelope": { httpStatus: 400, retryable: false, partialAllowed: false },
  "contract-mismatch": { httpStatus: 409, retryable: false, partialAllowed: false },
  "node-failed": { httpStatus: 502, retryable: true, partialAllowed: true },
  "partial-result": { httpStatus: 206, retryable: false, partialAllowed: true },
  "budget-exceeded": { httpStatus: 402, retryable: false, partialAllowed: true },
  "deadline-exceeded": { httpStatus: 504, retryable: true, partialAllowed: true },
  cancelled: { httpStatus: 499, retryable: false, partialAllowed: true },
  internal: { httpStatus: 500, retryable: true, partialAllowed: false },
};

export type EnvelopeIssueCode = "missing" | "invalid" | "duplicate" | "version" | "status-inconsistent" | "mismatch";

export interface EnvelopeIssue {
  path: string;
  code: EnvelopeIssueCode;
  expected: string;
  actual: string;
}

const ERROR_CODES = new Set<string>(Object.keys(ERROR_POLICY));
const STATUSES = new Set<string>(["succeeded", "partial", "failed", "cancelled"]);
const MODES = new Set<string>(["copy", "pass", "share"]);

function push(issues: EnvelopeIssue[], path: string, code: EnvelopeIssueCode, expected: string, actual: unknown): void {
  issues.push({ path, code, expected, actual: String(actual) });
}

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

function nonEmpty(value: string): boolean {
  return value.trim() !== "";
}

function sortIssues(issues: EnvelopeIssue[]): EnvelopeIssue[] {
  return issues.sort((a, b) => (a.path < b.path ? -1 : a.path > b.path ? 1 : 0));
}

export function errorPolicy(code: OrchestrationErrorCode): ErrorPolicy {
  return ERROR_POLICY[code];
}

export function httpStatusFor(code: OrchestrationErrorCode): number {
  return ERROR_POLICY[code].httpStatus;
}

export function isRetryableCode(code: OrchestrationErrorCode): boolean {
  return ERROR_POLICY[code].retryable;
}

/** 请求门禁：版本、身份、预算、上下文引用与幂等键。 */
export function validateRequest(request: OrchestrationRequest): EnvelopeIssue[] {
  const issues: EnvelopeIssue[] = [];
  if (!nonEmpty(request.contractVersion) || parseVersion(request.contractVersion) === null) {
    push(issues, "$.contractVersion", "invalid", "major.minor.patch", request.contractVersion);
  }
  if (!nonEmpty(request.runId)) push(issues, "$.runId", "missing", "non-empty", request.runId);
  if (!nonEmpty(request.tenantId)) push(issues, "$.tenantId", "missing", "non-empty", request.tenantId);
  if (!nonEmpty(request.intent)) push(issues, "$.intent", "missing", "non-empty", request.intent);
  if (!isRecord(request.input)) push(issues, "$.input", "invalid", "object", typeof request.input);
  if (!Number.isFinite(request.deadlineMs) || request.deadlineMs <= 0) {
    push(issues, "$.deadlineMs", "invalid", "finite > 0", request.deadlineMs);
  }
  if (!Number.isInteger(request.maxCostMicros) || request.maxCostMicros <= 0) {
    push(issues, "$.maxCostMicros", "invalid", "integer > 0", request.maxCostMicros);
  }
  if (!MODES.has(request.mode)) push(issues, "$.mode", "invalid", "copy|pass|share", request.mode);
  if (!nonEmpty(request.idempotencyKey)) push(issues, "$.idempotencyKey", "missing", "non-empty", request.idempotencyKey);

  const seen = new Set<string>();
  request.contextRefs.forEach((reference, index) => {
    const path = `$.contextRefs[${index}]`;
    if (!nonEmpty(reference.id) || !nonEmpty(reference.revision)) {
      push(issues, path, "invalid", "id and revision", reference.id);
      return;
    }
    if (seen.has(reference.id)) push(issues, path, "duplicate", "unique id", reference.id);
    seen.add(reference.id);
  });
  return sortIssues(issues);
}

/** 响应门禁：状态与错误必须自洽，节点结果唯一，用量非负。 */
export function validateResponse(response: OrchestrationResponse, request?: OrchestrationRequest): EnvelopeIssue[] {
  const issues: EnvelopeIssue[] = [];
  if (!nonEmpty(response.contractVersion) || parseVersion(response.contractVersion) === null) {
    push(issues, "$.contractVersion", "invalid", "major.minor.patch", response.contractVersion);
  }
  if (!nonEmpty(response.runId)) push(issues, "$.runId", "missing", "non-empty", response.runId);
  if (!STATUSES.has(response.status)) push(issues, "$.status", "invalid", "succeeded|partial|failed|cancelled", response.status);
  if (!isRecord(response.output)) push(issues, "$.output", "invalid", "object", typeof response.output);
  if (!nonEmpty(response.idempotencyKey)) push(issues, "$.idempotencyKey", "missing", "non-empty", response.idempotencyKey);

  if (request !== undefined) {
    if (response.runId !== request.runId) push(issues, "$.runId", "mismatch", request.runId, response.runId);
    if (response.idempotencyKey !== request.idempotencyKey) {
      push(issues, "$.idempotencyKey", "mismatch", request.idempotencyKey, response.idempotencyKey);
    }
    if (nonEmpty(response.contractVersion) && parseVersion(response.contractVersion) !== null) {
      if (!checkVersionCompatibility(response.contractVersion, request.contractVersion).ok) {
        push(issues, "$.contractVersion", "version", request.contractVersion, response.contractVersion);
      }
    }
  }

  for (const key of ["inputTokens", "outputTokens", "costMicros"] as const) {
    const value = response.usage[key];
    if (!Number.isInteger(value) || value < 0) push(issues, `$.usage.${key}`, "invalid", "integer >= 0", value);
  }

  const seen = new Set<string>();
  response.nodes.forEach((node, index) => {
    const path = `$.nodes[${index}]`;
    if (!nonEmpty(node.nodeId) || !nonEmpty(node.capability)) {
      push(issues, path, "invalid", "nodeId and capability", node.nodeId);
      return;
    }
    if (seen.has(node.nodeId)) push(issues, `${path}.nodeId`, "duplicate", "unique", node.nodeId);
    seen.add(node.nodeId);
    if (!Number.isInteger(node.costMicros) || node.costMicros < 0) {
      push(issues, `${path}.costMicros`, "invalid", "integer >= 0", node.costMicros);
    }
    if (!Number.isFinite(node.latencyMs) || node.latencyMs < 0) {
      push(issues, `${path}.latencyMs`, "invalid", "finite >= 0", node.latencyMs);
    }
    if (node.errorCode !== undefined && !ERROR_CODES.has(node.errorCode)) {
      push(issues, `${path}.errorCode`, "invalid", "known code", node.errorCode);
    }
  });

  if (response.error !== undefined) {
    if (!ERROR_CODES.has(response.error.code)) {
      push(issues, "$.error.code", "invalid", "known code", response.error.code);
    } else if (response.error.retryable !== ERROR_POLICY[response.error.code].retryable) {
      push(issues, "$.error.retryable", "status-inconsistent", String(ERROR_POLICY[response.error.code].retryable), response.error.retryable);
    }
  }

  const failedNodes = response.nodes.filter((node) => node.status === "failed").length;
  if (response.status === "succeeded") {
    if (response.error !== undefined) push(issues, "$.error", "status-inconsistent", "absent", response.error.code);
    if (failedNodes > 0) push(issues, "$.status", "status-inconsistent", "no failed node", `${failedNodes} failed`);
  } else if (response.status === "failed") {
    if (response.error === undefined) push(issues, "$.error", "status-inconsistent", "present", "missing");
  } else if (response.status === "partial") {
    if (response.error === undefined) push(issues, "$.error", "status-inconsistent", "present", "missing");
    else if (!ERROR_POLICY[response.error.code].partialAllowed) {
      push(issues, "$.error.code", "status-inconsistent", "partial-allowed code", response.error.code);
    }
  } else if (response.status === "cancelled") {
    if (response.error === undefined || response.error.code !== "cancelled") {
      push(issues, "$.error.code", "status-inconsistent", "cancelled", response.error?.code ?? "missing");
    }
  }
  return sortIssues(issues);
}

export function assertRequest(request: OrchestrationRequest): void {
  const issues = validateRequest(request);
  if (issues.length > 0) throw new AiError("invalid-request", `envelope: ${describeEnvelopeIssues(issues)}`);
}

export function assertResponse(response: OrchestrationResponse, request?: OrchestrationRequest): void {
  const issues = validateResponse(response, request);
  if (issues.length > 0) throw new AiError("invalid-output", `envelope: ${describeEnvelopeIssues(issues)}`);
}

export function describeEnvelopeIssues(issues: readonly EnvelopeIssue[]): string {
  return issues.map((issue) => `${issue.path} ${issue.code} expected=${issue.expected} actual=${issue.actual}`).join("; ");
}

/** 有可用输出时，失败或降级降为 partial；没有可用输出时只能是 failed。 */
export function deriveStatus(nodes: readonly NodeOutcome[], hasUsableOutput: boolean): RunStatus {
  const degraded = nodes.some(
    (node) => node.status === "failed" || node.status === "degraded" || node.status === "skipped",
  );
  if (!degraded) return "succeeded";
  return hasUsableOutput ? "partial" : "failed";
}

/** 把内部错误码翻译成对外的 HTTP 语义。 */
export function toPublicError(code: OrchestrationErrorCode, message: string): {
  status: number;
  body: { code: OrchestrationErrorCode; retryable: boolean; message: string };
} {
  const policy = ERROR_POLICY[code];
  return { status: policy.httpStatus, body: { code, retryable: policy.retryable, message } };
}