refactor(rc-8): wire circuit-breaker half-open gate into request pipeline (#123)

Adds a non-throwing canAttempt() gate to CircuitBreaker and consults it from the main fetch dispatcher before every upstream call. Previously the breaker state machine existed but was never consulted, so OPEN-state short-circuiting and HALF-OPEN probe serialization were dead code.

- lib/circuit-breaker.ts: add canAttempt() returning CanAttemptResult ({ allowed, state, reason }); preserve existing canExecute() throwing surface unchanged for backward compatibility.

- lib/errors.ts: extend CircuitOpenError with optional breakerKey, state, and reason so callers can classify the short-circuit without parsing message strings; zero-arg constructor preserved.

- index.ts: compute per-(account, family) breaker key inside the retry loop, call canAttempt() before fetch, short-circuit to rotation on denial (break + refund + circuit-open metric), and feed recordFailure() on 5xx/network errors plus recordSuccess() on successful responses.

- test/circuit-breaker-wiring.test.ts: 10 new tests covering CLOSED->OPEN, OPEN->HALF_OPEN after cooldown, HALF_OPEN->CLOSED on probe success, HALF_OPEN->OPEN on probe failure (with cooldown reset), concurrent-probe rejection, customized halfOpenMaxAttempts, and CircuitOpenError payload.

All 2131 tests pass (2121 baseline + 10 new). typecheck, lint, and build clean.

Author: ndycode

index.ts

@@ -155,6 +155,7 @@ import {
 	resetRateLimitBackoff,
 } from "./lib/request/rate-limit-backoff.js";
 import { isEmptyResponse } from "./lib/request/response-handler.js";
+import { getCircuitBreaker } from "./lib/circuit-breaker.js";
 import {
 	RetryBudgetTracker,
 	resolveRetryBudgetLimits,
@@ -1866,10 +1867,36 @@ while (attempted.size < Math.max(1, accountCount)) {
 									break;
 								}
 
+							// RC-8: per-(account, family) circuit-breaker key. The breaker gates
+							// upstream calls so that repeated failures short-circuit to the
+							// rotation path instead of hammering a degraded endpoint.
+							const circuitBreakerKey = `${accountId}:${modelFamily}`;
+							const circuitBreaker = getCircuitBreaker(circuitBreakerKey);
+
 							while (true) {
 								let response: Response;
 								const fetchStart = performance.now();
 
+								// RC-8: consult the breaker BEFORE firing upstream. When the gate is
+								// closed every call passes through unchanged. When the gate denies
+								// (open within cooldown, or half-open with a probe already in
+								// flight) we short-circuit to the rotation path instead of
+								// retrying here. We classify the short-circuit as `circuit-open`
+								// so observability traces and the runtime metrics agree with the
+								// `CircuitOpenError` type exported from `lib/errors.ts`.
+								const breakerCheck = circuitBreaker.canAttempt();
+								if (!breakerCheck.allowed) {
+									const shortCircuitMessage = `Circuit ${breakerCheck.state} for ${circuitBreakerKey}`;
+									logWarn(
+										`[circuit-breaker] ${shortCircuitMessage} (reason=${breakerCheck.reason ?? "denied"}). Rotating account.`,
+									);
+									accountManager.refundToken(account, modelFamily, model);
+									runtimeMetrics.accountRotations++;
+									runtimeMetrics.lastError = shortCircuitMessage;
+									runtimeMetrics.lastErrorCategory = "circuit-open";
+									break;
+								}
+
 								// Merge user AbortSignal with timeout (Node 18 compatible - no AbortSignal.any)
 								const fetchController = new AbortController();
 								const requestTimeoutMs = fetchTimeoutMs;
@@ -1939,6 +1966,10 @@ while (attempted.size < Math.max(1, accountCount)) {
 								runtimeMetrics.lastErrorCategory = "network";
 								accountManager.refundToken(account, modelFamily, model);
 								accountManager.recordFailure(account, modelFamily, model);
+								// RC-8: network failures feed the breaker so a degraded upstream
+								// trips the gate for this (account, family) key after N hits
+								// inside the failure window.
+								circuitBreaker.recordFailure();
 								break;
 							} finally {
 								clearTimeout(fetchTimeoutId);
@@ -2195,6 +2226,11 @@ while (attempted.size < Math.max(1, accountCount)) {
 						runtimeMetrics.lastErrorCategory = "server";
 						accountManager.refundToken(account, modelFamily, model);
 						accountManager.recordFailure(account, modelFamily, model);
+						// RC-8: 5xx responses are treated the same as network failures by
+						// the breaker — they indicate an upstream fault rather than a
+						// client-side classifier decision (401/403/404/429 are handled
+						// upstream and do not feed the breaker).
+						circuitBreaker.recordFailure();
 						if (
 							!consumeRetryBudget(
 								"server",
@@ -2325,6 +2361,9 @@ while (attempted.size < Math.max(1, accountCount)) {
 					}
 
 					accountManager.recordSuccess(account, modelFamily, model);
+					// RC-8: closes a half-open gate or prunes the failure window so a
+					// sequence of successes keeps the breaker healthy.
+					circuitBreaker.recordSuccess();
 					runtimeMetrics.successfulRequests++;
 					runtimeMetrics.lastError = null;
 					runtimeMetrics.lastErrorCategory = null;

lib/circuit-breaker.ts

@@ -18,6 +18,23 @@ export const DEFAULT_CIRCUIT_BREAKER_CONFIG: CircuitBreakerConfig = {
 
 export type CircuitState = "closed" | "open" | "half-open";
 
+/**
+ * Non-throwing gate check result returned by {@link CircuitBreaker.canAttempt}.
+ *
+ * Unlike {@link CircuitBreaker.canExecute} (which throws `CircuitOpenError`
+ * when the gate denies the call), `canAttempt` returns this result so the
+ * request pipeline can decide whether to short-circuit, rotate accounts,
+ * or emit typed errors without a try/catch around every upstream call.
+ */
+export interface CanAttemptResult {
+	/** True when the caller may proceed to the protected dependency. */
+	allowed: boolean;
+	/** Current breaker state at the time of the check. */
+	state: CircuitState;
+	/** Populated when `allowed` is false. Stable machine-readable reason. */
+	reason?: "open" | "probe-in-flight";
+}
+
 export class CircuitBreaker {
 	private state: CircuitState = "closed";
 	private failures: number[] = [];
@@ -29,6 +46,49 @@ export class CircuitBreaker {
 		this.config = { ...DEFAULT_CIRCUIT_BREAKER_CONFIG, ...config };
 	}
 
+	/**
+	 * Non-throwing gate check used by the request pipeline.
+	 *
+	 * Semantics parallel {@link canExecute} but the caller receives a
+	 * {@link CanAttemptResult} instead of an exception:
+	 * - `closed` → `{ allowed: true, state: "closed" }`
+	 * - `open` past cooldown → auto-transitions to `half-open` then admits a
+	 *   single probe; returns `{ allowed: true, state: "half-open" }`
+	 * - `open` within cooldown → `{ allowed: false, state: "open", reason: "open" }`
+	 * - `half-open` with in-flight probe → `{ allowed: false, state: "half-open", reason: "probe-in-flight" }`
+	 *
+	 * HALF-OPEN probe serialization: the first caller increments
+	 * `halfOpenAttempts` and proceeds; subsequent callers see the budget
+	 * exhausted and are denied with `probe-in-flight`. The probe slot is
+	 * released on the next {@link recordSuccess} (closes) or
+	 * {@link recordFailure} (reopens), which both reset `halfOpenAttempts`.
+	 */
+	canAttempt(): CanAttemptResult {
+		const now = Date.now();
+
+		if (this.state === "open") {
+			if (now - this.lastStateChange >= this.config.resetTimeoutMs) {
+				this.transitionToHalfOpen(now);
+			} else {
+				return { allowed: false, state: "open", reason: "open" };
+			}
+		}
+
+		if (this.state === "half-open") {
+			if (this.halfOpenAttempts >= this.config.halfOpenMaxAttempts) {
+				return {
+					allowed: false,
+					state: "half-open",
+					reason: "probe-in-flight",
+				};
+			}
+			this.halfOpenAttempts += 1;
+			return { allowed: true, state: "half-open" };
+		}
+
+		return { allowed: true, state: "closed" };
+	}
+
 	canExecute(): boolean {
 		const now = Date.now();
 

lib/errors.ts

@@ -209,16 +209,40 @@ export class StorageError extends CodexError {
 	}
 }
 
+/**
+ * Options carried by {@link CircuitOpenError} when raised from the request
+ * pipeline so callers can classify the short-circuit without parsing the
+ * message string.
+ */
+export interface CircuitOpenErrorOptions {
+	/** The breaker key that denied the call, e.g. `account:modelFamily`. */
+	breakerKey?: string;
+	/** Snapshot of the breaker state at denial time (`open` | `half-open`). */
+	state?: "open" | "half-open";
+	/** Machine-readable denial reason from `CanAttemptResult`. */
+	reason?: "open" | "probe-in-flight";
+}
+
 /**
  * Error thrown when a circuit breaker is open (or half-open past its attempt
  * budget) and further calls must short-circuit instead of hitting the
  * protected dependency.
+ *
+ * When constructed from the request pipeline's gate check, {@link breakerKey},
+ * {@link state}, and {@link reason} carry the metadata needed by the rotation
+ * path to pick a different account/family without re-querying the breaker.
  */
 export class CircuitOpenError extends CodexError {
 	override readonly name = "CircuitOpenError";
+	readonly breakerKey?: string;
+	readonly state?: "open" | "half-open";
+	readonly reason?: "open" | "probe-in-flight";
 
-	constructor(message = "Circuit is open") {
+	constructor(message = "Circuit is open", options?: CircuitOpenErrorOptions) {
 		super(message, { code: ErrorCode.CIRCUIT_OPEN });
+		this.breakerKey = options?.breakerKey;
+		this.state = options?.state;
+		this.reason = options?.reason;
 	}
 }
 

test/circuit-breaker-wiring.test.ts

@@ -0,0 +1,178 @@
+/**
+ * RC-8 — wiring tests for the non-throwing `canAttempt` gate.
+ *
+ * These tests exercise the request-pipeline contract: `canAttempt` returns a
+ * `CanAttemptResult` without throwing, so the dispatcher in `index.ts` can
+ * short-circuit to the rotation path. State transitions (CLOSED → OPEN,
+ * OPEN → HALF_OPEN after cooldown, HALF_OPEN → CLOSED on probe success,
+ * HALF_OPEN → OPEN on probe failure, concurrent-probe rejection) are verified
+ * here — the existing `test/circuit-breaker.test.ts` keeps coverage for the
+ * legacy throwing `canExecute` surface.
+ */
+
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import {
+	CircuitBreaker,
+	CircuitOpenError,
+	DEFAULT_CIRCUIT_BREAKER_CONFIG,
+} from "../lib/circuit-breaker.js";
+
+const { failureThreshold, resetTimeoutMs } = DEFAULT_CIRCUIT_BREAKER_CONFIG;
+
+describe("circuit-breaker: wired gate (canAttempt)", () => {
+	beforeEach(() => {
+		vi.useFakeTimers();
+		vi.setSystemTime(new Date(0));
+	});
+
+	afterEach(() => {
+		vi.useRealTimers();
+	});
+
+	it("CLOSED: allows the call and reports state=closed", () => {
+		const breaker = new CircuitBreaker();
+		const result = breaker.canAttempt();
+		expect(result.allowed).toBe(true);
+		expect(result.state).toBe("closed");
+		expect(result.reason).toBeUndefined();
+	});
+
+	it("CLOSED → OPEN: denies the call once failureThreshold reached within the window", () => {
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		expect(breaker.getState()).toBe("open");
+
+		const result = breaker.canAttempt();
+		expect(result.allowed).toBe(false);
+		expect(result.state).toBe("open");
+		expect(result.reason).toBe("open");
+	});
+
+	it("OPEN → HALF_OPEN: auto-transitions after cooldown and admits a probe", () => {
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		expect(breaker.getState()).toBe("open");
+
+		// Advance past the cooldown window.
+		vi.setSystemTime(new Date(resetTimeoutMs + 1));
+
+		const first = breaker.canAttempt();
+		expect(first.allowed).toBe(true);
+		expect(first.state).toBe("half-open");
+		expect(breaker.getState()).toBe("half-open");
+	});
+
+	it("HALF_OPEN: first probe allowed, concurrent probe rejected with probe-in-flight", () => {
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		vi.setSystemTime(new Date(resetTimeoutMs + 1));
+
+		const first = breaker.canAttempt();
+		expect(first.allowed).toBe(true);
+		expect(first.state).toBe("half-open");
+
+		// A second concurrent caller (no recordSuccess / recordFailure yet) sees
+		// the probe slot already consumed and is denied without throwing.
+		const second = breaker.canAttempt();
+		expect(second.allowed).toBe(false);
+		expect(second.state).toBe("half-open");
+		expect(second.reason).toBe("probe-in-flight");
+	});
+
+	it("HALF_OPEN → CLOSED: probe success closes the gate and allows the next caller", () => {
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		vi.setSystemTime(new Date(resetTimeoutMs + 1));
+
+		expect(breaker.canAttempt().allowed).toBe(true);
+		breaker.recordSuccess();
+		expect(breaker.getState()).toBe("closed");
+
+		const next = breaker.canAttempt();
+		expect(next.allowed).toBe(true);
+		expect(next.state).toBe("closed");
+	});
+
+	it("HALF_OPEN → OPEN: probe failure reopens and resets the cooldown", () => {
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		vi.setSystemTime(new Date(resetTimeoutMs + 1));
+
+		// Admit the probe…
+		expect(breaker.canAttempt().allowed).toBe(true);
+
+		// …and fail it. The breaker must reopen with a fresh cooldown window
+		// so subsequent callers are denied until the timer elapses again.
+		breaker.recordFailure();
+		expect(breaker.getState()).toBe("open");
+
+		const immediate = breaker.canAttempt();
+		expect(immediate.allowed).toBe(false);
+		expect(immediate.state).toBe("open");
+		expect(immediate.reason).toBe("open");
+
+		// Cooldown advances relative to the reopen timestamp, not the original
+		// open transition.
+		vi.setSystemTime(new Date(resetTimeoutMs + 2 + resetTimeoutMs + 1));
+		const recovered = breaker.canAttempt();
+		expect(recovered.allowed).toBe(true);
+		expect(recovered.state).toBe("half-open");
+	});
+
+	it("respects a custom halfOpenMaxAttempts > 1 before denying with probe-in-flight", () => {
+		const breaker = new CircuitBreaker({ halfOpenMaxAttempts: 2 });
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		vi.setSystemTime(new Date(resetTimeoutMs + 1));
+
+		expect(breaker.canAttempt().allowed).toBe(true);
+		expect(breaker.canAttempt().allowed).toBe(true);
+		const third = breaker.canAttempt();
+		expect(third.allowed).toBe(false);
+		expect(third.reason).toBe("probe-in-flight");
+	});
+
+	it("canAttempt does not regress existing canExecute behavior", () => {
+		// canExecute still throws on the denial paths so the legacy surface
+		// used by other call sites is preserved.
+		const breaker = new CircuitBreaker();
+		for (let i = 0; i < failureThreshold; i++) {
+			breaker.recordFailure();
+		}
+		expect(() => breaker.canExecute()).toThrow(CircuitOpenError);
+	});
+});
+
+describe("CircuitOpenError (typed short-circuit payload)", () => {
+	it("carries breakerKey, state, and reason when constructed from the pipeline", () => {
+		const error = new CircuitOpenError("Circuit open for acct:gpt-5", {
+			breakerKey: "acct:gpt-5",
+			state: "open",
+			reason: "open",
+		});
+		expect(error).toBeInstanceOf(CircuitOpenError);
+		expect(error.code).toBe("CODEX_CIRCUIT_OPEN");
+		expect(error.breakerKey).toBe("acct:gpt-5");
+		expect(error.state).toBe("open");
+		expect(error.reason).toBe("open");
+	});
+
+	it("preserves the zero-argument constructor for backward compatibility", () => {
+		const error = new CircuitOpenError();
+		expect(error.message).toBe("Circuit is open");
+		expect(error.breakerKey).toBeUndefined();
+		expect(error.state).toBeUndefined();
+		expect(error.reason).toBeUndefined();
+	});
+});