feat(rollout): gate each ramp step on the new version's calls and errors

docs/guides/promotion.md

@@ -234,6 +234,23 @@ default, otherwise left for you. The 6 hours assume the promotion step's
 330-minute limit in CI; a promotion run on your machine must finish within
 them.
 
+Optionally, gate each step on the new version's calls:
+
+```yaml
+    rollout:
+      healthGate:
+        minCalls: 20           # calls on the new version before the next step
+        maxErrorRateDelta: 2   # abort if its error rate is this many points higher
+```
+
+Before each step, the ticker reads the calls each version took since the
+current step (the same metrics as the assistant's traffic tab). It waits
+until the new version has `minCalls`, then aborts the ramp, with the reason
+in the allocation's note, if the new version's share of calls that ended in
+an error is more than `maxErrorRateDelta` percentage points above the old
+version's. `healthGate: {}` takes those defaults. It runs after, never
+instead of, a `check:` gate.
+
 Keep in mind:
 
 - GitHub runs scheduled workflows from the default branch only, may start

src/promotion.ts

@@ -34,6 +34,15 @@ export interface PromotionRollout {
   steps: number[];
   // Minimum minutes between steps. The rollout workflow ticks every 15.
   intervalMinutes: number;
+  healthGate?: PromotionRolloutHealthGate;
+}
+
+// Checked before each step: the ramp waits until the new version has taken
+// `minCalls` calls since the last step, and aborts if its error rate is more
+// than `maxErrorRateDelta` percentage points above the old version's.
+export interface PromotionRolloutHealthGate {
+  minCalls: number;
+  maxErrorRateDelta: number;
 }
 
 export interface PromotionOrg {
@@ -162,7 +171,37 @@ export const ROLLOUT_STEPS_DEFAULT = [1, 5, 10, 25, 50, 100];
 export const ROLLOUT_INTERVAL_MINUTES_DEFAULT = 30;
 // The rollout workflow's cron period; a shorter interval could not be honored.
 export const ROLLOUT_INTERVAL_MINUTES_MIN = 15;
-const ROLLOUT_KEYS = ["steps", "intervalMinutes"];
+const ROLLOUT_KEYS = ["steps", "intervalMinutes", "healthGate"];
+const HEALTH_GATE_KEYS = ["minCalls", "maxErrorRateDelta"];
+export const HEALTH_GATE_MIN_CALLS_DEFAULT = 20;
+export const HEALTH_GATE_MAX_ERROR_RATE_DELTA_DEFAULT = 2;
+
+function healthGateParse(
+  value: unknown,
+  slug: string,
+): PromotionRolloutHealthGate {
+  const raw = object(value ?? {}, `org ${slug}.rollout.healthGate`);
+  for (const key of Object.keys(raw))
+    if (!HEALTH_GATE_KEYS.includes(key))
+      throw new Error(
+        `org ${slug}.rollout.healthGate has unknown key "${key}" (allowed: ${HEALTH_GATE_KEYS.join(", ")})`,
+      );
+  const minCalls = raw.minCalls ?? HEALTH_GATE_MIN_CALLS_DEFAULT;
+  if (typeof minCalls !== "number" || !Number.isInteger(minCalls) || minCalls < 1)
+    throw new Error(`org ${slug}.rollout.healthGate.minCalls must be a whole number of at least 1`);
+  const maxErrorRateDelta =
+    raw.maxErrorRateDelta ?? HEALTH_GATE_MAX_ERROR_RATE_DELTA_DEFAULT;
+  if (
+    typeof maxErrorRateDelta !== "number" ||
+    !Number.isFinite(maxErrorRateDelta) ||
+    maxErrorRateDelta < 0 ||
+    maxErrorRateDelta > 100
+  )
+    throw new Error(
+      `org ${slug}.rollout.healthGate.maxErrorRateDelta must be percentage points between 0 and 100`,
+    );
+  return { minCalls, maxErrorRateDelta };
+}
 
 // The API stores shares as integer thousandths of a percent.
 export function millipercentage(percent: number): number {
@@ -214,7 +253,13 @@ export function promotionRolloutParse(
     throw new Error(
       `org ${slug}.rollout.intervalMinutes must be a whole number of at least ${ROLLOUT_INTERVAL_MINUTES_MIN}`,
     );
-  return { steps: [...steps], intervalMinutes };
+  return {
+    steps: [...steps],
+    intervalMinutes,
+    ...("healthGate" in raw
+      ? { healthGate: healthGateParse(raw.healthGate, slug) }
+      : {}),
+  };
 }
 
 export function promotionConfigParse(content: string): PromotionConfig {

src/rollout-api.ts

@@ -42,6 +42,15 @@ export interface AllocationWrite {
   expectedCurrentAllocationId?: string;
 }
 
+// One version's calls in a metrics window, as the health gate reads them.
+export interface RolloutVersionHealth {
+  // The version label; null for calls with no version.
+  version: string | null;
+  calls: number;
+  // Calls whose ended reason is an error.
+  errors: number;
+}
+
 export interface RolloutAssistant {
   // The newest published version label; null when versioning is off.
   latestVersion: string | null;
@@ -66,8 +75,18 @@ export interface RolloutApi {
   ): Promise<Allocation>;
   // false when traffic splitting is not enabled for the org (a beta flag).
   allocationsProbe(): Promise<boolean>;
+  // Per-version calls and errors between `from` and `to`.
+  healthGet(
+    assistantId: string,
+    from: Date,
+    to: Date,
+  ): Promise<RolloutVersionHealth[]>;
 }
 
+// GET /traffic-allocations/metrics allows at most 2000 timeseries buckets;
+// hourly buckets make that 83 days, far longer than any step.
+export const METRICS_WINDOW_MAX_MS = 2000 * 3_600_000;
+
 // The platform's test for an assistant managed by Vapi for CX: a UUID in
 // metadata.vapiAgentId. Anything else there isn't a lock.
 const VAPI_AGENT_ID_PATTERN =
@@ -82,6 +101,19 @@ interface AllocationLatestResponse {
   allocation?: Allocation;
 }
 
+interface MetricsResponse {
+  versions?: Array<{
+    assistantVersionLabel: string | null;
+    calls: number;
+    endedReasons?: Array<{ reason: string; count: number }>;
+  }>;
+}
+
+// The dashboard's rule for an error ended reason: the reason names one.
+function endedReasonIsError(reason: string): boolean {
+  return reason.includes("error");
+}
+
 function notFound(error: unknown): boolean {
   return error instanceof VapiApiError && error.statusCode === 404;
 }
@@ -152,6 +184,29 @@ export function rolloutApiCreate(connection: OrgConnection): RolloutApi {
         { retry: "rate-limit-only" },
       );
     },
+    async healthGet(assistantId, from, to) {
+      const start = new Date(
+        Math.max(from.getTime(), to.getTime() - METRICS_WINDOW_MAX_MS),
+      );
+      const query = new URLSearchParams({
+        assistantId,
+        from: start.toISOString(),
+        to: to.toISOString(),
+        bucket: "hour",
+      });
+      const metrics = await vapiFetchJson<MetricsResponse>(
+        reads,
+        "GET",
+        `/traffic-allocations/metrics?${query}`,
+      );
+      return (metrics.versions ?? []).map((version) => ({
+        version: version.assistantVersionLabel,
+        calls: version.calls,
+        errors: (version.endedReasons ?? [])
+          .filter((ended) => endedReasonIsError(ended.reason))
+          .reduce((sum, ended) => sum + ended.count, 0),
+      }));
+    },
     async allocationsProbe() {
       try {
         await vapiFetchJson<unknown>(

src/rollout-cmd.ts

@@ -11,9 +11,13 @@ import { resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import type { OrgConnection } from "./org-connection.ts";
 import { connectionLoad, tokensParse } from "./org-connection.ts";
-import type { PromotionConfig } from "./promotion.ts";
+import type { PromotionConfig, PromotionRollout } from "./promotion.ts";
 import { promotionConfigParse, promotionStateParse } from "./promotion.ts";
-import type { Allocation, RolloutApi } from "./rollout-api.ts";
+import type {
+  Allocation,
+  RolloutApi,
+  RolloutVersionHealth,
+} from "./rollout-api.ts";
 import { allocationConflictGet, rolloutApiCreate } from "./rollout-api.ts";
 import { rolloutMarkerParse } from "./rollout-marker.ts";
 import type { RolloutTickWrite } from "./rollout-tick.ts";
@@ -152,13 +156,15 @@ async function orgRun(
         continue;
       }
       const assistant = await api.assistantGet(uuid);
+      const now = deps.now();
       const decision = rolloutTickDecide({
         assistantId: uuid,
         allocation,
         latestVersion: assistant?.latestVersion ?? null,
         rollout,
-        now: deps.now(),
+        now,
         runId,
+        health: await healthRead(api, rollout, allocation, now),
       });
       if (args.action === "advance" && decision.write)
         await tickWrite(api, id, decision.write);
@@ -173,6 +179,22 @@ async function orgRun(
   return failures;
 }
 
+// Per-version calls since the current step began, read only when a gated
+// step is due, since each read is a metrics query.
+async function healthRead(
+  api: RolloutApi,
+  rollout: PromotionRollout,
+  allocation: Allocation | null,
+  now: number,
+): Promise<RolloutVersionHealth[] | undefined> {
+  if (!rollout.healthGate || !allocation) return undefined;
+  if (rolloutMarkerParse(allocation.description)?.kind !== "step")
+    return undefined;
+  const since = Date.parse(allocation.createdAt);
+  if (now - since < rollout.intervalMinutes * 60_000) return undefined;
+  return api.healthGet(allocation.assistantId, new Date(since), new Date(now));
+}
+
 // An abort is idempotent in intent, so a lost race is retried once against
 // the newest row.
 async function abortWrite(

src/rollout-marker.ts

@@ -4,7 +4,8 @@
 // nothing about a ramp is committed to git.
 //
 //   gitops-rollout <kind> [from=vN] [to=vM] [step=i/K] [latest=vX]
-//                         [prev=follow-latest|marked|manual] [hold=vK] [run=<id>]
+//                         [prev=follow-latest|marked|manual] [hold=vK]
+//                         [reason=<token>] [run=<id>]
 
 export type RolloutMarkerKind =
   | "pin"
@@ -33,6 +34,8 @@ export interface RolloutMarker {
   prev?: RolloutPrevious;
   // A newer, un-ramped version that kept a complete ramp from releasing.
   hold?: string;
+  // Why the health gate aborted, as one token (e.g. error-rate+4.1pp).
+  reason?: string;
   run?: string;
 }
 
@@ -56,6 +59,7 @@ export function rolloutMarkerFormat(marker: RolloutMarker): string {
   if (marker.latest) fields.push(`latest=${marker.latest}`);
   if (marker.prev) fields.push(`prev=${marker.prev}`);
   if (marker.hold) fields.push(`hold=${marker.hold}`);
+  if (marker.reason) fields.push(`reason=${marker.reason}`);
   if (marker.run) fields.push(`run=${marker.run}`);
   return fields.join(" ");
 }
@@ -77,6 +81,7 @@ export function rolloutMarkerParse(
     if (key === "to") marker.to = value;
     if (key === "latest") marker.latest = value;
     if (key === "hold") marker.hold = value;
+    if (key === "reason") marker.reason = value;
     if (key === "run") marker.run = value;
     if (key === "prev" && PREVIOUS.includes(value as RolloutPrevious))
       marker.prev = value as RolloutPrevious;

src/rollout-tick.ts

@@ -3,7 +3,11 @@
 // is tested without an API.
 
 import type { PromotionRollout } from "./promotion.ts";
-import type { Allocation, AllocationWrite } from "./rollout-api.ts";
+import type {
+  Allocation,
+  AllocationWrite,
+  RolloutVersionHealth,
+} from "./rollout-api.ts";
 import type { RolloutMarker } from "./rollout-marker.ts";
 import { rolloutMarkerFormat, rolloutMarkerParse } from "./rollout-marker.ts";
 import { rolloutStepTargets } from "./rollout-promote.ts";
@@ -23,6 +27,8 @@ export interface RolloutTickInput {
   rollout: PromotionRollout;
   now: number;
   runId?: string;
+  // Calls since the current step, per version; only with a health gate.
+  health?: RolloutVersionHealth[];
 }
 
 export interface RolloutTickWrite {
@@ -84,6 +90,8 @@ function stepNext(
     return {
       note: `step ${marker.step}/${steps.length} at ${marker.to}; next in ${Math.ceil(intervalMinutes - age)} min`,
     };
+  const gated = healthGateCheck(input, marker);
+  if (gated) return gated;
   // The last step (100%) is written as `complete`, not as a 0/100 split.
   if (marker.step + 1 < steps.length)
     return { write: stepWrite(input, marker.from, marker.to, marker.step + 1) };
@@ -175,6 +183,45 @@ function pinNext(
   };
 }
 
+function errorRate(health: RolloutVersionHealth | undefined): number {
+  if (!health || health.calls === 0) return 0;
+  return (health.errors / health.calls) * 100;
+}
+
+// Before a step moves on: wait for enough calls on the new version, then
+// abort if its error rate is too far above the old version's.
+function healthGateCheck(
+  input: RolloutTickInput,
+  marker: RolloutMarker,
+): RolloutTickDecision | null {
+  const gate = input.rollout.healthGate;
+  if (!gate) return null;
+  const health = input.health ?? [];
+  const to = health.find((version) => version.version === marker.to);
+  const from = health.find((version) => version.version === marker.from);
+  const calls = to?.calls ?? 0;
+  if (calls < gate.minCalls)
+    return {
+      note: `step ${marker.step} at ${marker.to}: ${calls} of ${gate.minCalls} calls before the next step`,
+    };
+  const delta = errorRate(to) - errorRate(from);
+  if (delta <= gate.maxErrorRateDelta) return null;
+  const reason = `error-rate+${delta.toFixed(1)}pp`;
+  const abort = rolloutAbortWrite(
+    input.assistantId,
+    input.allocation,
+    input.runId,
+    reason,
+  );
+  if (!abort) return null;
+  return {
+    write: {
+      ...abort,
+      summary: `aborted by the health gate: ${marker.to}'s error rate is ${delta.toFixed(1)} points above ${marker.from}'s`,
+    },
+  };
+}
+
 export function rolloutTickDecide(input: RolloutTickInput): RolloutTickDecision {
   const marker = rolloutMarkerParse(input.allocation?.description);
   if (!input.allocation || !marker) return {};
@@ -198,6 +245,7 @@ export function rolloutAbortWrite(
   assistantId: string,
   allocation: Allocation | null,
   runId?: string,
+  reason?: string,
 ): RolloutTickWrite | null {
   const marker = rolloutMarkerParse(allocation?.description);
   if (!allocation || !marker) return null;
@@ -219,6 +267,7 @@ export function rolloutAbortWrite(
         to: marker.to,
         step: marker.step,
         steps: marker.steps,
+        reason,
         run: runId,
       }),
       expectedCurrentAllocationId: allocation.id,