import { spawn } from "node:child_process";
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { FileAuthStorageBackend, getAgentDir } from "@earendil-works/pi-coding-agent";
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import {
  createAssistantMessageEventStream,
  type AssistantMessage,
  type AssistantMessageEventStream,
  type Context,
  type Model,
  type SimpleStreamOptions,
} from "@earendil-works/pi-ai";
import { streamSimpleAnthropic, streamSimpleOpenAICodexResponses } from "@earendil-works/pi-ai/compat";
import {
  loginAnthropic,
  loginOpenAICodex,
  refreshAnthropicToken,
  refreshOpenAICodexToken,
} from "@earendil-works/pi-ai/oauth";
import {
  ANTHROPIC_PROVIDER_ID,
  OPENAI_CODEX_PROVIDER_ID,
  formatStatusLine,
  getLadderProviderId,
  getProviderLabel,
  isManagedProvider,
  addSubscription,
  commitDefaultIdentity,
  markSubscriptionAuthenticationFailed,
  markSubscriptionRateLimited,
  markSubscriptionSuccess,
  parseRetryAfterHeader,
  promoteSubscription,
  readRotationConfig,
  readSubscriptionPool,
  removeManagedAccount,
  resolveFallbackTarget,
  resolveProviderCandidates,
  restorePreferredModelIfPossible,
  switchToFallbackIfPossible,
  upsertSubscription,
  upsertSubscriptionWithUniqueAccount,
  type FallbackTarget,
  type ManagedProviderId,
  type SubscriptionCredential,
} from "../../../core/subscription-state.js";
import {
  ManagedProviderCoolingError,
  readManagedProviderRotationPolicy,
} from "../../../core/runtime/managed-provider-cooling.js";

// A response without Retry-After proves only that THIS attempt was throttled; it
// does not authorize crouter to invent a five-minute provider-wide outage. Keep
// a short turn-local probe backoff to avoid an immediate retry loop, then let a
// fresh turn prove whether the credential is usable again. Real
// server-provided Retry-After deadlines are still honored exactly.
const DEFAULT_RATE_LIMIT_BACKOFF_MS = 15 * 1000;
const INVALID_REFRESH_TOKEN_BACKOFF_MS = 30 * 24 * 60 * 60 * 1000;
// Short in-place retry budget for a transient network/5xx blip on the SAME credential
// (#122): a couple of quick retries, no cooldown. Once this budget is exhausted we no
// longer hard-fail the turn -- we surface the error and rotate (see TRANSIENT_ROTATE below).
const TRANSIENT_RETRY_BACKOFFS_MS = [250, 750];
// After the in-place transient retry budget is exhausted, rotate instead of failing: the
// credential itself is probably healthy (this is a provider-side blip like "overloaded"),
// so cool it down only briefly -- long enough to rotate to another credential / fallback
// provider, short enough that it comes back quickly if it was the only option.
const TRANSIENT_ROTATE_COOLDOWN_MS = 60 * 1000;
// A managed turn can legitimately revisit a pool after a concurrent success or
// a short cooldown. Bound the whole continuation chain, not just provider
// switching, so repeated A/B clears cannot retain an unbounded async stack.
const MAX_MANAGED_PROVIDER_CONTINUATIONS = 32;
// The SDK's synchronous lock helper already retries for roughly 180ms. A few
// short outer retries absorb a normal multi-broker collision; if the metadata
// file remains busy, the provider call still proceeds. Subscription bookkeeping
// must never turn a healthy model response into a fatal generation fault.
const SUBSCRIPTION_LOCK_RETRY_BACKOFFS_MS = [25, 75, 150];

let runtimeContext: ExtensionContext | undefined;
let extensionAPI: ExtensionAPI | undefined;

type AttemptResult = {
  result: "success" | "rate_limited" | "transient_exhausted" | "fatal";
  retryAfterMs?: number;
  attemptAt: number;
  attempts?: number;
  classification?: ProviderErrorClassification;
};

// #122/#124: a genuine rate limit (HTTP 429, or 503 WITH a real retry-after header) must
// cool the credential down and rotate to the next one; a transient network/5xx blip
// (connection reset, socket errors, timeouts, plain 500/502/504, or 503 with NO
// retry-after) is not evidence the credential itself is bad -- it gets a short in-place
// retry on the same credential, never a cooldown. Everything else is fatal. Bare
// `usage`/`quota` substrings stay out of the message match: they are false-positive
// magnets (e.g. "usage" appears in normal assistant text), while the precise
// terminal usage-limit signals above still cool the credential down and rotate to the
// next one.
type ProviderErrorClassification = {
  kind: "rate_limit" | "transient" | "fatal";
  status?: number;
  reason: string;
  hadRetryAfterHeader: boolean;
};

const TRANSIENT_MESSAGE_PATTERN =
  /connection|network|fetch failed|socket|econnreset|econnrefused|enotfound|etimedout|timed? out|timeout|overloaded|service.?unavailable|temporarily unavailable/i;
const RATE_LIMIT_MESSAGE_PATTERN = /\b429\b|rate.?limit|too many requests/i;
// Precise terminal usage-limit signals (as opposed to the bare `usage`/`quota` substrings
// dropped above): pi-ai's Codex adapter surfaces `usage_limit_reached` / `usage_not_included`
// as the friendly "You have hit your ChatGPT usage limit..." message, often on a non-429
// status, so the status-code check above misses it. These are specific enough not to false-
// positive on ordinary assistant text mentioning "usage".
const USAGE_LIMIT_MESSAGE_PATTERN = /usage[ _]limit|usage_not_included|usagelimiterror|insufficient_quota|quota exceeded/i;

function classifyProviderError(error: unknown, observed: { status?: number; retryAfterMs?: number }): ProviderErrorClassification {
  const value = error as { status?: unknown; errorMessage?: unknown; message?: unknown } | undefined;
  const errorStatus = typeof value?.status === "number" ? value.status : undefined;
  const status = observed.status ?? errorStatus;
  const message = [value?.errorMessage, value?.message, error instanceof Error ? error.message : ""]
    .filter((part): part is string => typeof part === "string" && part.length > 0)
    .join(" ");
  const hadRetryAfterHeader = observed.retryAfterMs !== undefined;

  if (status === 429) return { kind: "rate_limit", status, reason: "http 429", hadRetryAfterHeader };
  if (status === 503 && hadRetryAfterHeader) return { kind: "rate_limit", status, reason: "http 503 with retry-after", hadRetryAfterHeader };
  if (status === 500 || status === 502 || status === 503 || status === 504) {
    return { kind: "transient", status, reason: `http ${status}`, hadRetryAfterHeader };
  }
  const transientMatch = message.match(TRANSIENT_MESSAGE_PATTERN);
  if (transientMatch) return { kind: "transient", status, reason: transientMatch[0].toLowerCase(), hadRetryAfterHeader };
  const rateLimitMatch = message.match(RATE_LIMIT_MESSAGE_PATTERN);
  if (rateLimitMatch) return { kind: "rate_limit", status, reason: rateLimitMatch[0].toLowerCase(), hadRetryAfterHeader };
  const usageLimitMatch = message.match(USAGE_LIMIT_MESSAGE_PATTERN);
  if (usageLimitMatch) return { kind: "rate_limit", status, reason: usageLimitMatch[0].toLowerCase(), hadRetryAfterHeader };
  return { kind: "fatal", status, reason: "unclassified", hadRetryAfterHeader };
}

async function defaultSleep(ms: number): Promise<void> {
  await new Promise<void>((resolve) => setTimeout(resolve, ms));
}

// Indirection so tests can inject deterministic waits without real timers.
let activeSleep: typeof defaultSleep = defaultSleep;

/** Test-only seam: override the sleep implementation used for in-place/wait-out retries. Pass undefined to restore the default. */
export function __setSleepForTest(fn: typeof defaultSleep | undefined): void {
  activeSleep = fn ?? defaultSleep;
}

function isSubscriptionLockContention(error: unknown): boolean {
  const value = error as { code?: unknown; message?: unknown } | undefined;
  return value?.code === "ELOCKED" || (typeof value?.message === "string" && /lock file is already being held/i.test(value.message));
}

async function recordSubscriptionMutation(description: string, mutate: () => void): Promise<boolean> {
  for (let attempt = 0; ; attempt++) {
    try {
      mutate();
      return true;
    } catch (error) {
      if (!isSubscriptionLockContention(error)) throw error;
      const backoffMs = SUBSCRIPTION_LOCK_RETRY_BACKOFFS_MS[attempt];
      if (backoffMs === undefined) {
        logRotation(`subscription metadata lock remained busy while ${description}; continuing without failing the provider turn`);
        return false;
      }
      await activeSleep(backoffMs);
    }
  }
}

/** Test-only seam for the bounded, non-fatal subscription bookkeeping path. */
export async function __recordSubscriptionMutationForTest(description: string, mutate: () => void): Promise<boolean> {
  return recordSubscriptionMutation(description, mutate);
}

// Parse `status=400` from pi-ai's plain Error message so genuine invalid_grant reaches
// cooldown / reauth / fallback; the typed status field is still honored for tests and other
// callers that provide one.
function isInvalidRefreshTokenError(error: unknown): boolean {
  const value = error as { status?: unknown; errorMessage?: unknown; message?: unknown } | undefined;
  const typedStatus = typeof value?.status === "number" ? value.status : undefined;
  const message = [value?.errorMessage, value?.message, error instanceof Error ? error.message : ""]
    .filter((part): part is string => typeof part === "string" && part.length > 0)
    .join(" ");
  const messageStatusMatch = message.match(/\bstatus[=:]\s*(\d+)/i);
  const status = typedStatus ?? (messageStatusMatch ? Number(messageStatusMatch[1]) : Number.NaN);
  return status === 400 && /invalid_grant|refresh token .*invalid|refresh token .*not found/i.test(message);
}

// A user/host abort (ESC, a critical-tier interrupt, a torn-down window) is NOT a provider
// failure — it must never cool down a credential or rotate providers. The upstream pi-ai
// providers set `reason: "aborted"` / `stopReason: "aborted"` (and throw "Request was aborted")
// ONLY when the request's AbortSignal fired, so those markers alone are a reliable abort
// signal; we do NOT require the signal to be threaded back through our own options (it may not
// be), which is what made an ESC drain both provider pools.
function isUserAbortEvent(event: Parameters<AssistantMessageEventStream["push"]>[0], signal: AbortSignal | undefined): boolean {
  if (signal?.aborted) return true;
  if (event.type !== "error") return false;
  return event.reason === "aborted" || event.error.stopReason === "aborted" || /request was aborted/i.test(event.error.errorMessage ?? "");
}

function isUserAbortError(error: unknown, signal: AbortSignal | undefined): boolean {
  if (signal?.aborted) return true;
  const value = error as { name?: unknown; errorMessage?: unknown; message?: unknown } | undefined;
  const message = [value?.errorMessage, value?.message, error instanceof Error ? error.message : String(error ?? "")]
    .filter((part): part is string => typeof part === "string" && part.length > 0)
    .join(" ");
  return (error instanceof Error && /abort/i.test(error.name)) || /request was aborted|operation was aborted|the user aborted/i.test(message);
}

function setStatus(text: string): void {
  runtimeContext?.ui.setStatus("provider-rotation", text);
}

function logRotation(message: string): void {
  process.stderr.write(`[provider-rotation] ${message}\n`);
}

function getCurrentModelRef(model?: Model<any>): { providerId: ManagedProviderId; modelId: string } | undefined {
  return isManagedProvider(model?.provider) ? { providerId: model.provider, modelId: model.id } : undefined;
}

function toAssistantError(model: Model<any>, errorMessage: string): AssistantMessage {
  return {
    role: "assistant",
    content: [],
    api: model.api,
    provider: model.provider,
    model: model.id,
    usage: {
      input: 0,
      output: 0,
      cacheRead: 0,
      cacheWrite: 0,
      totalTokens: 0,
      cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
    },
    stopReason: "error",
    errorMessage,
    timestamp: Date.now(),
  };
}

function retryAfterMs(headers: Record<string, string>): number | undefined {
  const retryAfterMsHeader = headers["retry-after-ms"];
  if (retryAfterMsHeader) {
    const millis = Number(retryAfterMsHeader);
    if (Number.isFinite(millis)) return Math.max(0, millis);
  }
  return parseRetryAfterHeader(headers["retry-after"]);
}

type RefreshedCredential = { refresh: string; access: string; expires: number };
// Codex login carries the ChatGPT `accountId` (JWT `chatgpt_account_id`); Anthropic login does
// not (its OAuth token has no account claim). Optional so both providers and the test seam fit.
type LoginCredential = { refresh: string; access: string; expires: number; accountId?: string };

// Codex login returns the account identity; extract it as a trimmed string (login results type
// it loosely through the OAuthCredentials index signature). Absent for Anthropic.
function loginAccountId(cred: { accountId?: unknown }): string | undefined {
  return typeof cred.accountId === "string" && cred.accountId.trim() ? cred.accountId.trim() : undefined;
}

// Short, display-only rendering of a Codex account id for `/provider-sub list`, so distinct
// accounts are observable without dumping the full opaque id.
function shortAccountId(accountId: string): string {
  return accountId.length > 12 ? `${accountId.slice(0, 8)}\u2026${accountId.slice(-4)}` : accountId;
}

// The default slot's live value only ever exists inside auth.json (see
// `refreshDefaultSlotCredential`) -- once a credential has been through refresh/reauth for
// this turn, `.access` is guaranteed populated (whether it came from the pool's own stored
// value for a non-default account, or was just read/refreshed out of auth.json for the
// default slot). Downstream stream-attempt code relies on `.access` being present rather
// than re-deriving "is this the default slot" logic at every call site.
type ResolvedCredential = SubscriptionCredential & { access: string };

async function defaultRefreshForProvider(providerId: ManagedProviderId, refreshToken: string): Promise<RefreshedCredential> {
  return providerId === ANTHROPIC_PROVIDER_ID ? refreshAnthropicToken(refreshToken) : refreshOpenAICodexToken(refreshToken);
}

// Indirection so tests can inject refresh failures without real network calls.
let activeRefreshForProvider: typeof defaultRefreshForProvider = defaultRefreshForProvider;

/** Test-only seam: override the per-provider refresh function. Pass undefined to restore the default. */
export function __setRefreshForProviderForTest(fn: typeof defaultRefreshForProvider | undefined): void {
  activeRefreshForProvider = fn ?? defaultRefreshForProvider;
}

function openBrowser(url: string): void {
  const [cmd, args] =
    process.platform === "darwin"
      ? ["open", [url]]
      : process.platform === "win32"
        ? ["cmd", ["/c", "start", "", url]]
        : ["xdg-open", [url]];
  try {
    const child = spawn(cmd, args, { stdio: "ignore", detached: true });
    child.on("error", () => {
      /* best effort -- the manual-code dialog's title carries the URL too */
    });
    child.unref();
  } catch {
    /* best effort */
  }
}

type AuthCredentialRecord = { type?: string; refresh?: string; access?: string; expires?: number; [key: string]: unknown };
type AuthFileData = Record<string, AuthCredentialRecord>;

// The default slot (`label === providerId`) reads its live value from pi's auth.json
// (design §1/§3). A fresh access token takes a read-only fast path; an expired token
// delegates the whole read -> check-expiry -> refresh -> write transaction to pi's
// `FileAuthStorageBackend.withLockAsync`, which holds a real cross-process file lock,
// re-reads `current` under the lock on every acquisition, and releases in a `finally`
// even if the callback throws. A second refresher that reaches the lock after a winner
// sees the rotated `access`/`expires` and skips its own network refresh. We do NOT route
// through pi's public `getApiKey` because it wraps pi-ai's `getOAuthApiKey`, which
// collapses the refresh failure into a generic message; calling our own
// `activeRefreshForProvider` inside the callback keeps the real error intact so
// `isInvalidRefreshTokenError` can classify the failure (cooldown -> reauth ->
// cross-provider-fallback) exactly as it does for non-default accounts.
async function refreshDefaultSlotCredential(providerId: ManagedProviderId, credential: SubscriptionCredential): Promise<ResolvedCredential> {
  // Healthy requests vastly outnumber refreshes. Reading a still-fresh access
  // token is side-effect free, so it does not need to join the cross-process
  // refresh lock. If this snapshot races a write and cannot be parsed, fall
  // through to the locked transaction below. Expired tokens still use the real
  // lock, preserving the single-refresh-winner guarantee.
  try {
    const snapshot = JSON.parse(readFileSync(join(getAgentDir(), "auth.json"), "utf8")) as AuthFileData;
    const current = snapshot[providerId];
    if (current?.type === "oauth" && current.access && typeof current.expires === "number" && Date.now() < current.expires) {
      return { ...credential, access: current.access };
    }
  } catch {
    // Missing/partial/corrupt snapshots are handled authoritatively under lock.
  }

  const backend = new FileAuthStorageBackend();
  const access = await backend.withLockAsync(async (current) => {
    const data: AuthFileData = current ? JSON.parse(current) : {};
    const cred = data[providerId];
    // Only an OAuth-typed record is the default slot's refreshable credential: an
    // api_key record (or no record at all) has no refresh token to refresh, so it is never
    // treated as fresh or refreshable here.
    if (cred?.type === "oauth" && cred.access && typeof cred.expires === "number" && Date.now() < cred.expires) {
      return { result: cred.access };
    }
    if (cred?.type !== "oauth" || !cred.refresh) {
      throw new Error(`${getProviderLabel(providerId)} is not authenticated — run \`crtr sys setup\` (or /provider-sub ${providerId} add)`);
    }
    const refreshed = await activeRefreshForProvider(providerId, cred.refresh);
    data[providerId] = { ...cred, type: "oauth", refresh: refreshed.refresh, access: refreshed.access, expires: refreshed.expires };
    return { result: refreshed.access, next: `${JSON.stringify(data, null, 2)}\n` };
  });
  return { ...credential, access };
}

async function refreshCredentialIfNeeded(providerId: ManagedProviderId, credential: SubscriptionCredential): Promise<ResolvedCredential> {
  if (credential.label === providerId) return refreshDefaultSlotCredential(providerId, credential);
  if (!credential.refresh || !credential.access || credential.expires === undefined) {
    throw new Error(`${getProviderLabel(providerId)} subscription "${credential.label}" is missing its credential value`);
  }
  if (credential.expires > Date.now()) return credential as ResolvedCredential;
  const refreshed = await activeRefreshForProvider(providerId, credential.refresh);
  const next: SubscriptionCredential = { ...credential, refresh: refreshed.refresh, access: refreshed.access, expires: refreshed.expires };
  return (upsertSubscription(providerId, next).find((entry) => entry.label === next.label) ?? next) as ResolvedCredential;
}

/** Test-only seam: exercises the real `refreshCredentialIfNeeded` (including the default-slot
 * `withLockAsync` path) without duplicating its logic. */
export async function __refreshCredentialIfNeededForTest(providerId: ManagedProviderId, credential: SubscriptionCredential): Promise<ResolvedCredential> {
  return refreshCredentialIfNeeded(providerId, credential);
}

async function defaultLoginForProvider(providerId: ManagedProviderId, ctx: ExtensionContext): Promise<LoginCredential> {
  return loginProvider(providerId, ctx);
}

let activeLoginForProvider: typeof defaultLoginForProvider = defaultLoginForProvider;

/** Test-only seam: override the per-provider login function. Pass undefined to restore the default. */
export function __setLoginForProviderForTest(fn: typeof defaultLoginForProvider | undefined): void {
  activeLoginForProvider = fn ?? defaultLoginForProvider;
}

async function reauthenticateCredential(providerId: ManagedProviderId, credential: SubscriptionCredential): Promise<ResolvedCredential | undefined> {
  const ctx = runtimeContext;
  if (!ctx?.hasUI) return undefined;
  ctx.ui.notify(`${getProviderLabel(providerId)} subscription "${credential.label}" needs re-authentication. Opening OAuth login...`, "warn");
  const authenticated = await activeLoginForProvider(providerId, ctx);
  if (credential.label === providerId) {
    // Atomically commit the default (re)login: validate account-uniqueness (excluding the
    // default slot), THEN persist auth.json, THEN causally reconcile the metadata-only pool entry
    // under the pool lock (pool -> auth order), so a collision rejects BEFORE auth.json switches
    // and a newer same-account 429 survives. Re-read the healed pool for returned metadata.
    commitDefaultIdentity(providerId, {
      refresh: authenticated.refresh,
      access: authenticated.access,
      expires: authenticated.expires,
      accountId: loginAccountId(authenticated),
      lastRateLimitedAt: credential.lastRateLimitedAt,
    });
    const meta = readSubscriptionPool(providerId).find((entry) => entry.label === credential.label) ?? credential;
    return { ...meta, access: authenticated.access };
  }
  const reauthedAccountId = loginAccountId(authenticated) ?? credential.accountId;
  const { authFailure: _authFailure, ...healthyCredential } = credential;
  const next: SubscriptionCredential = {
    ...healthyCredential,
    refresh: authenticated.refresh,
    access: authenticated.access,
    expires: authenticated.expires,
    ...(reauthedAccountId ? { accountId: reauthedAccountId } : {}),
    rateLimitedUntil: 0,
    lastAttemptAt: Date.now(),
    lastRateLimitedAt: credential.lastRateLimitedAt,
  };
  // Identity-checked upsert: reauthenticating this explicit slot into an account already held by
  // ANOTHER slot throws DuplicateSubscriptionError rather than creating a duplicate quota entry.
  return (upsertSubscriptionWithUniqueAccount(providerId, next).find((entry) => entry.label === next.label) ?? next) as ResolvedCredential;
}

function defaultStreamForProvider(model: Model<any>, context: Context, options: SimpleStreamOptions | undefined): AssistantMessageEventStream {
  if (model.provider === ANTHROPIC_PROVIDER_ID) {
    return streamSimpleAnthropic(model as Model<"anthropic-messages">, context, options);
  }
  return streamSimpleOpenAICodexResponses(model as Model<"openai-codex-responses">, context, options);
}

// Indirection so tests can inject a fake per-provider stream without real network calls.
let activeStreamForProvider: typeof defaultStreamForProvider = defaultStreamForProvider;

function streamForProvider(model: Model<any>, context: Context, options: SimpleStreamOptions | undefined): AssistantMessageEventStream {
  return activeStreamForProvider(model, context, options);
}

/** Test-only seam: override the per-provider stream factory. Pass undefined to restore the default. */
export function __setStreamForProviderForTest(fn: typeof defaultStreamForProvider | undefined): void {
  activeStreamForProvider = fn ?? defaultStreamForProvider;
}

/** Test-only seam: exercise `fallbackRejectionHint` for one rejection discriminant
 *  (with an optional resolved-but-rejected candidate) without staging a full turn. */
export function __fallbackRejectionHintForTest(reason: FallbackRejectionReason, candidate?: FallbackTarget): string {
  return fallbackRejectionHint({ reason, candidate });
}

// Drives ONE credential, retrying transient (non-rate-limit) failures in place on the
// SAME credential per #122, and reports back a classified result for #124's logging.
// A genuine rate limit ends the attempt immediately (the caller cools the credential down
// and rotates); a transient blip gets a short in-place retry budget before giving up.
async function streamProviderAttempt(
  providerId: ManagedProviderId,
  model: Model<any>,
  context: Context,
  options: SimpleStreamOptions | undefined,
  credential: ResolvedCredential,
  outer: AssistantMessageEventStream,
): Promise<AttemptResult> {
  for (let attempt = 1; ; attempt++) {
    const attemptAt = Date.now();
    let observedStatus: number | undefined;
    let requestedRetryAfterMs: number | undefined;
    const inner = streamForProvider(model, context, {
      ...options,
      apiKey: credential.access,
      onResponse: async (response, responseModel) => {
        observedStatus = response.status;
        if (response.status === 429 || response.status === 503) {
          requestedRetryAfterMs = retryAfterMs(response.headers);
        }
        await options?.onResponse?.(response, responseModel);
      },
    });

    const bufferedEvents: Parameters<AssistantMessageEventStream["push"]>[0][] = [];
    let sawContent = false;
    let committed = false;
    let classification: ProviderErrorClassification | undefined;

    // Events are buffered ONLY until this attempt commits to the provider -- i.e. until the
    // first real content event (thinking/text/toolcall) arrives, after which rotating away is
    // no longer possible. Before commit the buffer lets a clean pre-content rate-limit rotate
    // to another credential with nothing yet shown. Once committed we flush and forward every
    // subsequent event LIVE, so viewers stream token-by-token instead of receiving the whole
    // message at once when the turn finally completes.
    const emit = (event: Parameters<AssistantMessageEventStream["push"]>[0]): void => {
      if (committed) outer.push(event);
      else bufferedEvents.push(event);
    };
    const flush = (): void => {
      for (const event of bufferedEvents) outer.push(event);
      bufferedEvents.length = 0;
      committed = true;
    };

    try {
      for await (const event of inner) {
        if (event.type === "error") {
          if (isUserAbortEvent(event, options?.signal)) {
            emit(event);
            flush();
            return { result: "fatal", attemptAt };
          }
          classification = classifyProviderError(event.error, { status: observedStatus, retryAfterMs: requestedRetryAfterMs });
          if (classification.kind === "fatal" || sawContent) {
            emit(event);
            flush();
            return { result: "fatal", attemptAt, classification };
          }
          break;
        }
        if (event.type !== "start" && event.type !== "done") sawContent = true;
        emit(event);
        // First content event: commit to this provider -- flush the buffer and switch to live
        // forwarding so the rest of the turn streams incrementally.
        if (sawContent && !committed) flush();
      }
    } catch (error) {
      if (isUserAbortError(error, options?.signal)) {
        emit({
          type: "error",
          reason: "aborted",
          error: { ...toAssistantError(model, error instanceof Error ? error.message : String(error)), stopReason: "aborted" },
        });
        flush();
        return { result: "fatal", attemptAt };
      }
      classification = classifyProviderError(error, { status: observedStatus, retryAfterMs: requestedRetryAfterMs });
      if (classification.kind === "fatal" || sawContent) {
        emit({
          type: "error",
          reason: "error",
          error: toAssistantError(model, error instanceof Error ? error.message : String(error)),
        });
        flush();
        return { result: "fatal", attemptAt, classification };
      }
    }

    if (!classification && requestedRetryAfterMs !== undefined && !sawContent) {
      // Some providers signal a 429/503 purely via the response header without ever raising
      // a stream "error" event; treat that the same as an explicit rate-limit classification.
      classification = { kind: "rate_limit", status: observedStatus, reason: "retry-after header on success path", hadRetryAfterHeader: true };
    }

    if (!classification) {
      flush();
      return { result: "success", attemptAt };
    }

    if (classification.kind === "rate_limit") {
      return { result: "rate_limited", retryAfterMs: requestedRetryAfterMs, attemptAt, classification };
    }

    // Transient: retry the SAME credential/provider in place first -- no cooldown for the
    // quick in-place budget.
    const backoffMs = TRANSIENT_RETRY_BACKOFFS_MS[attempt - 1];
    if (backoffMs === undefined) {
      // In-place retry budget exhausted. Rather than hard-failing the turn, hand back a
      // transient-exhausted result so the caller surfaces the error to the user AND rotates
      // to another credential / fallback provider. Nothing is emitted into the stream here:
      // the buffered (pre-content) events are discarded exactly like the rate-limit path, so
      // the rotation target streams a clean turn.
      return { result: "transient_exhausted", attemptAt, classification, attempts: attempt };
    }
    logRotation(
      `${getProviderLabel(providerId)} subscription "${credential.label}" hit a transient error (${classification.reason}, status=${classification.status ?? "n/a"}); retrying same credential in ${backoffMs}ms (attempt ${attempt + 1})`,
    );
    await activeSleep(backoffMs);
  }
}

function emitTerminalError(stream: AssistantMessageEventStream, model: Model<any>, message: string): void {
  stream.push({ type: "error", reason: "error", error: toAssistantError(model, message) });
  stream.end();
}

// Why a candidate fallback was rejected -- each reason gets its OWN accurate hint
// (`fallbackRejectionHint`) so the terminal error never gives advice that can't fix the
// actual blocker. Only `no_credentials` is a genuine "go authenticate" case; the others
// are config/exhaustion states where telling the user to authenticate would be misleading.
type FallbackRejectionReason =
  | "no_ladder_entry"
  | "already_attempted"
  | "no_credentials"
  | "model_unregistered"
  | "cooling"
  | "provider_pinned";

// On rejection we carry the resolved-but-rejected `candidate` FallbackTarget whenever one
// exists (every reason except `no_ladder_entry`, where no target resolved at all), so the
// hint can name the exact provider/model and build an executable repair command from the
// candidate's config ladder key + strength.
type FallbackResolution =
  | { target: FallbackTarget }
  | { target?: undefined; reason: FallbackRejectionReason; candidate?: FallbackTarget };

// #123: gates cross-provider fallback on the target actually being usable right now --
// authenticated (its subscription pool has at least one credential, even if currently
// cooling down) AND registered in the model registry -- rather than blindly switching the
// session model to a provider nobody logged into. Returns the rejection reason (config has
// no ladder rung, the target was already tried this turn, the target has zero credentials,
// or its resolved model isn't in the registry) rather than just `undefined`, so callers can
// surface an accurate terminal error instead of always defaulting to "authenticate a
// fallback provider" -- which is actively wrong when the real blocker is a stale ladder.
function resolveViableFallback(
  providerId: ManagedProviderId,
  modelId: string | undefined,
  attempted: Set<ManagedProviderId>,
): FallbackResolution {
  const target = resolveFallbackTarget(providerId, modelId, readRotationConfig(), extensionAPI?.getThinkingLevel?.());
  if (!target) return { reason: "no_ladder_entry" };
  if (attempted.has(target.providerId)) return { reason: "already_attempted", candidate: target };
  const pool = readSubscriptionPool(target.providerId);
  const resolution = resolveProviderCandidates([{
    providerId: target.providerId,
    modelId: target.modelId,
    thinkingLevel: target.thinkingLevel,
    registered: Boolean(runtimeContext?.modelRegistry.find(target.providerId, target.modelId)),
    authenticated: pool.length > 0,
    coolingUntil: pool.length > 0 && pool.every((entry) => (entry.rateLimitedUntil || 0) > Date.now())
      ? Math.min(...pool.map((entry) => entry.rateLimitedUntil || 0))
      : undefined,
  }]);
  if (resolution.candidates.length > 0) return { target };
  switch (resolution.rejected[0]?.reason) {
    case "unauthenticated": return { reason: "no_credentials", candidate: target };
    case "unregistered": return { reason: "model_unregistered", candidate: target };
    // Do not churn the session model merely to wait on an unavailable target.
    // The caller includes this candidate's pool in the typed cooling deadline,
    // so the broker can re-drive when either provider becomes ready.
    case "cooling": return { reason: "cooling", candidate: target };
    default: return { reason: "no_credentials", candidate: target };
  }
}

// Actionable hint appended to a terminal error when no fallback was viable. Each rejection
// reason gets advice that can actually resolve IT -- only `no_credentials` says
// "authenticate", since that's the sole case where a login is the missing piece. Telling a
// user to authenticate when both providers are already exhausted, or when the ladder points
// at an unregistered model, is actively wrong and was the original bug.
function fallbackRejectionHint(resolution: FallbackResolution): string {
  if (resolution.target) return "try again later";
  const candidate = resolution.candidate;
  switch (resolution.reason) {
    case "no_credentials": {
      const who = candidate ? ` (${getProviderLabel(candidate.providerId)}, e.g. /provider-sub ${candidate.providerId} add)` : " with /provider-sub";
      return `authenticate a fallback provider${who}, or try again later`;
    }
    case "model_unregistered": {
      // Repair command must be EXECUTABLE as printed: `--value`, and the CONFIG ladder
      // provider key (`openai`, not the runtime `openai-codex`) plus the candidate's own
      // strength -- so the user pastes it, swaps in an installed model id, and it works.
      const ladderKey = candidate ? getLadderProviderId(candidate.providerId) : "<provider>";
      const strength = candidate?.strength ?? "<strength>";
      const named = candidate ? candidate.label : "the configured fallback";
      return `configured fallback ${named} is not in this build's model registry -- the modelLadders config is out of sync with the installed model catalog. Point it at an installed model with \`crtr sys config set modelLadders.${ladderKey}.${strength} --value <installed-model-id>\`, or upgrade/downgrade to a build whose ladder matches`;
    }
    case "already_attempted":
      return "every managed provider was already tried this turn and is exhausted or cooling down -- try again later";
    case "cooling":
      return "the configured fallback provider is also cooling down -- retrying when the first provider becomes ready";
    case "provider_pinned":
      return "the selected provider is explicitly pinned -- retrying it when its first credential becomes ready";
    case "no_ladder_entry":
      return "no cross-provider fallback is configured for this model -- set one with `crtr sys config set modelLadders.<provider>.<strength> --value <model-id>`, or try again later";
  }
}

// Switches the session model to an already-viability-checked fallback target. On the rare
// race where the switch itself still fails (e.g. `setModel` rejects it), emits a graceful,
// actionable terminal error on `stream` instead of throwing a generic "failed to switch".
async function switchToFallbackModel(
  stream: AssistantMessageEventStream,
  model: Model<any>,
  originalModel: { providerId: ManagedProviderId; modelId: string },
): Promise<FallbackTarget | undefined> {
  const ctx = runtimeContext;
  if (!extensionAPI || !ctx) {
    emitTerminalError(stream, model, `Provider fallback unavailable: runtime not initialized for ${originalModel.providerId}/${originalModel.modelId}`);
    return undefined;
  }
  const target = await switchToFallbackIfPossible(extensionAPI, ctx, originalModel);
  if (!target) {
    emitTerminalError(stream, model, `Provider fallback unavailable: failed to switch models for ${getProviderLabel(originalModel.providerId)}`);
    return undefined;
  }
  return target;
}

// Drives one managed provider's pool into `stream`; on exhaustion it switches the
// session model to the cross-provider fallback and CONTINUES the same turn on it,
// rather than failing the turn with an error. `attempted` guards anthropic<->codex
// ping-pong: once both pools are dry (or unauthenticated) we emit a single terminal error.
// If neither a fallback nor more credentials are available, emit a typed cooling
// terminal with the earliest deadline. The broker schedules the re-drive; never
// hold the active turn open in a silent in-stream sleep.
async function runManagedProvider(
  model: Model<any>,
  context: Context,
  options: SimpleStreamOptions | undefined,
  stream: AssistantMessageEventStream,
  attempted: Set<ManagedProviderId>,
  continuations = 0,
  authFailures = new Set<string>(),
  localCoolingDeadlines = new Map<ManagedProviderId, number>(),
): Promise<void> {
  if (continuations >= MAX_MANAGED_PROVIDER_CONTINUATIONS) {
    emitTerminalError(stream, model, "Managed provider recovery exceeded its bounded continuation budget; retry after checking provider status");
    return;
  }
  const providerId = model.provider as ManagedProviderId;
  attempted.add(providerId);
  const rawPool = readSubscriptionPool(providerId);
  for (const entry of rawPool) {
    if (entry.authFailure === "invalid_grant") authFailures.add(`${providerId}/${entry.label}`);
  }
  const available = rawPool.filter((entry) => !entry.rateLimitedUntil || entry.rateLimitedUntil <= Date.now());
  // Labels ready at the START of this attempt. If a credential becomes ready
  // AFTER this snapshot (a concurrent turn's cooldown expired or a causally valid
  // success cleared it), we re-enter to consume it rather than declaring
  // exhaustion from a stale selection (see the fresh-availability re-entry below).
  const availableLabels = new Set(available.map((entry) => entry.label));

  for (const credential of available) {
    let freshCredential: ResolvedCredential;
    try {
      freshCredential = await refreshCredentialIfNeeded(providerId, credential);
    } catch (error) {
      const attemptAt = Date.now();
      if (isInvalidRefreshTokenError(error)) {
        let reauthenticated: ResolvedCredential | undefined;
        try {
          reauthenticated = await reauthenticateCredential(providerId, credential);
        } catch (loginError) {
          runtimeContext?.ui.notify(
            `${getProviderLabel(providerId)} re-authentication failed: ${loginError instanceof Error ? loginError.message : String(loginError)}`,
            "error",
          );
        }
        if (reauthenticated) {
          freshCredential = reauthenticated;
        } else {
          // Keep the unusable credential out of ordinary rotation, but retain
          // its auth cause for this turn. A 30-day exclusion is not a transient
          // rate limit and must never author an auto rate-limit recovery fault.
          await recordSubscriptionMutation(
            `recording ${getProviderLabel(providerId)} authentication failure`,
            () => { markSubscriptionAuthenticationFailed(providerId, credential.label, INVALID_REFRESH_TOKEN_BACKOFF_MS, attemptAt); },
          );
          authFailures.add(`${providerId}/${credential.label}`);
          setStatus(formatStatusLine(providerId));
          continue;
        }
      } else {
        throw error;
      }
    }
    const result = await streamProviderAttempt(providerId, model, context, options, freshCredential, stream);

    if (result.result === "success") {
      // The common healthy state already has nothing to clear. Avoid two global
      // file locks just to advance display-only lastAttemptAt telemetry.
      if (freshCredential.rateLimitedUntil || freshCredential.authFailure) {
        await recordSubscriptionMutation(
          `recording ${getProviderLabel(providerId)} success`,
          () => { markSubscriptionSuccess(providerId, freshCredential.label, result.attemptAt); },
        );
      }
      logRotation(`${getProviderLabel(providerId)} subscription "${freshCredential.label}" succeeded; cooldown cleared`);
      setStatus(formatStatusLine(providerId));
      stream.end();
      return;
    }

    if (result.result === "rate_limited") {
      const cooldownMs = result.retryAfterMs ?? DEFAULT_RATE_LIMIT_BACKOFF_MS;
      if (result.retryAfterMs !== undefined) {
        // A server-declared deadline is authoritative for every broker sharing
        // this credential, so persist it in the cross-process pool.
        const persisted = await recordSubscriptionMutation(
          `recording ${getProviderLabel(providerId)} server cooldown`,
          () => { markSubscriptionRateLimited(providerId, freshCredential.label, cooldownMs, result.attemptAt); },
        );
        if (!persisted) {
          localCoolingDeadlines.set(
            providerId,
            Math.max(localCoolingDeadlines.get(providerId) ?? 0, Date.now() + cooldownMs),
          );
        }
      } else {
        // A headerless 429/usage-limit response proves only that this request
        // failed. Persisting an invented deadline globally let one busy node
        // gate unrelated nodes that were succeeding concurrently. Keep the
        // short probe deadline turn-local instead.
        localCoolingDeadlines.set(
          providerId,
          Math.max(localCoolingDeadlines.get(providerId) ?? 0, Date.now() + cooldownMs),
        );
      }
      // #124: make a misclassification obvious at a glance -- status, matched reason, and
      // whether a real retry-after header drove the cooldown vs. our default backoff.
      const retryAfterPart = result.retryAfterMs !== undefined ? `retry-after=${Math.ceil(result.retryAfterMs / 1000)}s` : "no retry-after header (default backoff)";
      logRotation(
        `${getProviderLabel(providerId)} subscription "${freshCredential.label}" rate-limited (status=${result.classification?.status ?? "n/a"}, reason=${result.classification?.reason ?? "unclassified"}, ${retryAfterPart}); ${result.retryAfterMs !== undefined ? "shared cooldown" : "turn-local retry backoff"} ${Math.ceil(cooldownMs / 1000)}s, trying next credential`,
      );
      // Surface the rotation to the user too (parity with the transient-exhausted path), so a
      // rate-limit-driven credential/provider switch is visible rather than silent.
      runtimeContext?.ui.notify?.(
        `${getProviderLabel(providerId)} "${freshCredential.label}" rate-limited (${result.retryAfterMs !== undefined ? "cooling down" : "retrying this turn in"} ${Math.ceil(cooldownMs / 1000)}s); rotating to another credential/provider.`,
        "warn",
      );
      continue;
    }

    if (result.result === "transient_exhausted") {
      // The credential is likely healthy -- this was a provider-side transient blip (e.g.
      // "overloaded") that survived the in-place retries. Surface it to the user and rotate:
      // cool the credential down briefly so this loop (and the pool-exhaustion logic below)
      // moves on to the next credential / fallback provider instead of failing the turn.
      const reason = result.classification?.reason ?? "transient error";
      const persisted = await recordSubscriptionMutation(
        `recording ${getProviderLabel(providerId)} transient cooldown`,
        () => { markSubscriptionRateLimited(providerId, freshCredential.label, TRANSIENT_ROTATE_COOLDOWN_MS, result.attemptAt); },
      );
      if (!persisted) {
        localCoolingDeadlines.set(
          providerId,
          Math.max(localCoolingDeadlines.get(providerId) ?? 0, Date.now() + TRANSIENT_ROTATE_COOLDOWN_MS),
        );
      }
      runtimeContext?.ui.notify?.(
        `${getProviderLabel(providerId)} "${freshCredential.label}" hit a transient error after ${result.attempts ?? 1} attempts (${reason}); rotating to another credential/provider.`,
        "warn",
      );
      logRotation(
        `${getProviderLabel(providerId)} subscription "${freshCredential.label}" transient error exhausted (reason=${reason}, status=${result.classification?.status ?? "n/a"}); cooling down ${Math.ceil(TRANSIENT_ROTATE_COOLDOWN_MS / 1000)}s and rotating`,
      );
      setStatus(formatStatusLine(providerId));
      continue;
    }

    // Fatal (or partial content already streamed): cannot safely switch providers mid-turn.
    if (result.classification) {
      logRotation(
        `${getProviderLabel(providerId)} subscription "${freshCredential.label}" failed fatally (status=${result.classification.status ?? "n/a"}, reason=${result.classification.reason})`,
      );
    }
    stream.end();
    return;
  }

  // Pool exhausted (or this provider was never authenticated at all) — figure out what to
  // do next: switch to a viable cross-provider fallback, wait out a sole/authed provider's
  // cooldown, or surface an actionable terminal error. Re-read the pool here rather than
  // reusing the pre-loop `rawPool` snapshot: the credential loop above mutates cooldowns
  // in place via `markSubscriptionRateLimited` (e.g. a sole credential that just got a
  // fresh 429), and the stale snapshot would still show it as available, making the
  // wait-out math below think there's nothing to wait for and wrongly declare exhaustion.
  const currentPool = readSubscriptionPool(providerId);

  // Fresh-availability re-entry: if any current credential is ready-now AND its
  // label was absent from the initial `available` snapshot, another turn freed it
  // up mid-attempt. Re-enter the SAME provider (same `attempted` set) so it
  // consumes that new readiness instead of falling through to fallback/wait/
  // terminal off a stale snapshot. The turn-global continuation budget bounds
  // repeated concurrent state changes across every re-entry path.
  const freshlyReady = currentPool.some(
    (entry) => (!entry.rateLimitedUntil || entry.rateLimitedUntil <= Date.now()) && !availableLabels.has(entry.label),
  );
  if (freshlyReady) {
    await runManagedProvider(model, context, options, stream, attempted, continuations + 1, authFailures, localCoolingDeadlines);
    return;
  }

  const originalModel = getCurrentModelRef(model);
  if (!originalModel) throw new Error(`Provider fallback unavailable: missing current model`);

  const rotationPolicy = readManagedProviderRotationPolicy(runtimeContext?.ui);
  const fallbackResolution: FallbackResolution = rotationPolicy.allowCrossProviderFallback
    ? resolveViableFallback(originalModel.providerId, originalModel.modelId, attempted)
    : { reason: "provider_pinned" };
  if (fallbackResolution.target) {
    const target = await switchToFallbackModel(stream, model, originalModel);
    if (!target) return; // terminal error already emitted
    logRotation(`${getProviderLabel(providerId)} pool exhausted -> ${target.label}`);
    setStatus(`${getProviderLabel(providerId)} -> ${target.label}`);

    const fallbackModel = runtimeContext?.modelRegistry.find(target.providerId, target.modelId);
    if (!fallbackModel) {
      emitTerminalError(stream, model, `Provider fallback unavailable: ${target.label} not in model registry`);
      return;
    }

    await runManagedProvider(fallbackModel, context, options, stream, attempted, continuations + 1, authFailures, localCoolingDeadlines);
    return;
  }

  // `attempted` prevents blind fallback ping-pong, but it must not hide a
  // credential that became ready again in an earlier pool. Re-resolve that
  // provider as a deliberate re-entry; the turn-global continuation bound above
  // remains the guard against repeated concurrent A/B state changes.
  const readyEarlierProvider = [...attempted].find((candidate) =>
    candidate !== providerId &&
    (localCoolingDeadlines.get(candidate) ?? 0) <= Date.now() &&
    readSubscriptionPool(candidate).some((entry) => !entry.rateLimitedUntil || entry.rateLimitedUntil <= Date.now()),
  );
  if (readyEarlierProvider !== undefined) {
    const reentryResolution = resolveViableFallback(originalModel.providerId, originalModel.modelId, new Set<ManagedProviderId>());
    if (reentryResolution.target?.providerId === readyEarlierProvider) {
      const target = await switchToFallbackModel(stream, model, originalModel);
      if (!target) return;
      const reentryModel = runtimeContext?.modelRegistry.find(target.providerId, target.modelId);
      if (!reentryModel) {
        emitTerminalError(stream, model, `Provider re-entry unavailable: ${target.label} not in model registry`);
        return;
      }
      await runManagedProvider(reentryModel, context, options, stream, attempted, continuations + 1, authFailures, localCoolingDeadlines);
      return;
    }
  }

  const label = getProviderLabel(providerId);
  const triedList = [...attempted].join(", ");

  // #123: this provider was never authenticated at all (no subscription pool entries ever
  // existed for it) -- there's nothing to wait out, so this is a config/setup problem, not
  // a transient one.
  if (currentPool.length === 0) {
    emitTerminalError(stream, model, `${label} not authenticated — run \`crtr sys setup\` (or /provider-sub ${providerId} add) (tried ${triedList})`);
    return;
  }

  // An invalid refresh token that could not be reauthenticated is a user-action
  // auth failure, not an ordinary cooldown. Fallbacks have already been given a
  // chance above; do not turn this into a daemon-owned rate-limit retry.
  const authFailure = [...authFailures][0];
  if (authFailure !== undefined) {
    emitTerminalError(stream, model, `Provider authentication failed (invalid_grant) for ${authFailure}; re-authenticate the credential and retry (tried ${triedList})`);
    return;
  }

  const fallbackHint = fallbackRejectionHint(fallbackResolution);

  // Every managed pool tried this turn is cooling with a finite future deadline:
  // surface a TYPED, rate-limit-classifiable terminal error carrying the earliest
  // deadline. The stophook stamps that deadline onto the pi→provider fault's
  // retry.nextAt so the broker's auto-retry re-drives the turn when the soonest
  // pool frees up — instead of the generic "exhausted" text classifying as a
  // fatal `other` fault that strands the node.
  const coolingProviders = new Set(attempted);
  if (fallbackResolution.reason === "cooling" && fallbackResolution.candidate) {
    coolingProviders.add(fallbackResolution.candidate.providerId);
  }
  const coolingDeadline = earliestCoolingDeadline(coolingProviders, localCoolingDeadlines);
  if (coolingDeadline !== undefined) {
    const coolingSecs = Math.ceil((coolingDeadline - Date.now()) / 1000);
    emitCoolingError(
      stream,
      model,
      new ManagedProviderCoolingError(
        `All managed providers are rate limited (cooling down ~${coolingSecs}s); ${fallbackHint} (tried ${triedList})`,
        coolingDeadline,
      ),
    );
    return;
  }

  emitTerminalError(
    stream,
    model,
    `All managed provider pools exhausted; ${fallbackHint} (tried ${triedList})`,
  );
}

// The earliest strictly-future cooldown deadline across every managed pool tried
// this turn, or undefined unless EVERY existing credential in those pools is
// cooling (a single ready credential, or an empty attempted set, means this is not
// an all-cooling condition and must not be typed as such). Read fresh under each
// pool's lock-free snapshot, mirroring the wait-out math above.
function earliestCoolingDeadline(
  attempted: Set<ManagedProviderId>,
  localCoolingDeadlines = new Map<ManagedProviderId, number>(),
): number | undefined {
  const now = Date.now();
  let earliest: number | undefined;
  let sawCredential = false;
  for (const attemptedProvider of attempted) {
    const localUntil = localCoolingDeadlines.get(attemptedProvider) ?? 0;
    if (localUntil > now) {
      sawCredential = true;
      if (earliest === undefined || localUntil < earliest) earliest = localUntil;
      continue;
    }
    for (const entry of readSubscriptionPool(attemptedProvider)) {
      sawCredential = true;
      const until = entry.rateLimitedUntil || 0;
      if (until <= now) return undefined; // a ready credential exists — not all cooling
      if (earliest === undefined || until < earliest) earliest = until;
    }
  }
  return sawCredential ? earliest : undefined;
}

// Emit the typed cooling terminal error, attaching its diagnostic (the earliest
// deadline) to the assistant error message so the deadline survives verbatim into
// agent_end.messages for the stophook to read.
function emitCoolingError(stream: AssistantMessageEventStream, model: Model<any>, error: ManagedProviderCoolingError): void {
  const message = toAssistantError(model, error.message);
  message.diagnostics = [error.toDiagnostic()];
  stream.push({ type: "error", reason: "error", error: message });
  stream.end();
}

function streamManagedProvider(model: Model<any>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream {
  if (!isManagedProvider(model.provider)) {
    return streamForProvider(model, context, options);
  }

  const stream = createAssistantMessageEventStream();

  runManagedProvider(model, context, options, stream, new Set<ManagedProviderId>()).catch((error) => {
    stream.push({
      type: "error",
      reason: "error",
      error: toAssistantError(model, error instanceof Error ? error.message : String(error)),
    });
    stream.end();
  });

  return stream;
}

function parseCommand(args: string): string[] {
  return args
    .trim()
    .split(/\s+/)
    .map((part) => part.trim())
    .filter(Boolean);
}

function parseProvider(value: string | undefined): ManagedProviderId | undefined {
  return isManagedProvider(value) ? value : undefined;
}

function updateRotationStatus(ctx: ExtensionContext): void {
  const current = getCurrentModelRef(ctx.model);
  if (current) setStatus(formatStatusLine(current.providerId));
}

// Manual-code input is a human-in-the-loop OAuth round-trip (open link, log in on
// whatever device, copy the localhost redirect URL back) that routinely exceeds
// DEFAULT_DIALOG_TIMEOUT_MS -- give it a human-scale bound instead of the ordinary
// confirm/select default.
const OAUTH_MANUAL_INPUT_TIMEOUT_MS = 600_000;

/**
 * pi-ai races the manual-code-paste dialog against its own loopback callback
 * server (local attach: the SAME browser can hit both). If the loopback wins,
 * pi-ai abandons the manual-input promise without ever awaiting it again -- so
 * left alone it stays open, on the broker AND onscreen, until
 * OAUTH_MANUAL_INPUT_TIMEOUT_MS elapses (review mrf5doek: local-attach
 * regression). `run` receives two things:
 *  - `readCode`, a `ctx.ui.input` wrapper it uses for the manual-code prompt. It
 *    threads an AbortSignal into the dialog: aborting it makes the broker drop
 *    the abandoned pending entry AND emit an `extension_ui_dismiss` frame keyed
 *    to THIS request's id, so the attach viewer tears down exactly this overlay
 *    (input-controller.ts `dismissDialog`) and nothing else.
 *  - `dismissManual`, which aborts that signal on demand. `run` fires it at the
 *    race-winner boundary -- pi-ai's `onProgress` "exchanging..." step, which
 *    runs the instant it HAS a code and BEFORE the token exchange completes.
 *    That dismisses the losing overlay promptly instead of after the (for
 *    OpenAI, unbounded/timeout-less) token exchange returns (review
 *    mrf68z4h). `onProgress` only ever fires post-code in pi-ai's browser login
 *    (anthropic.js exchange step; OpenAI's browser path emits none, so its
 *    dismissal falls back to `finally` below), so firing it never dismisses a
 *    dialog the user still needs -- a remote paste has already resolved
 *    `readCode` by then, making the abort a no-op.
 * `finally` re-aborts unconditionally as the backstop for any path that reported
 * no progress; AbortController.abort() is idempotent, so a second call is inert.
 *
 * `undefined` from `readCode` can ONLY mean the human cancelled or the dialog's
 * own timeout fired (or our own late abort, which pi-ai's abandoned-promise
 * `.catch` swallows as inert). pi-ai treats a falsy manual result as "no code
 * yet" and falls through to its `onPrompt` paste-only fallback -- a second,
 * URL-less dialog with the ordinary 120s default timeout, breaking the
 * one-dialog contract and making Cancel look inert (review mrf5doek finding
 * 2). `readCode` throws instead, ending the OAuth attempt directly.
 */
export async function loginWithManualInputCleanup<T>(
  ctx: ExtensionContext,
  run: (
    readCode: (title: string, placeholder: string) => Promise<string>,
    dismissManual: () => void,
  ) => Promise<T>,
): Promise<T> {
  const controller = new AbortController();
  const readCode = async (title: string, placeholder: string): Promise<string> => {
    const result = await ctx.ui.input(title, placeholder, { timeout: OAUTH_MANUAL_INPUT_TIMEOUT_MS, signal: controller.signal });
    if (result === undefined) throw new Error("OAuth login cancelled");
    return result;
  };
  try {
    return await run(readCode, () => controller.abort());
  } finally {
    controller.abort();
  }
}

async function loginProvider(providerId: ManagedProviderId, ctx: ExtensionContext) {
  if (providerId === ANTHROPIC_PROVIDER_ID) {
    let authorizeUrl: string | undefined;
    return loginWithManualInputCleanup(ctx, (readCode, dismissManual) =>
      loginAnthropic({
        onAuth: ({ url, instructions }) => {
          authorizeUrl = url;
          // Best-effort for the local attach case where a browser is reachable; on a
          // headless remote box this opens nothing and the URL embedded in the
          // manual-code dialog title below (rendered as a tappable link on web) is
          // the real path.
          openBrowser(url);
          if (instructions) ctx.ui.notify(instructions, "info");
        },
        onPrompt: async (prompt) => (await ctx.ui.input(prompt.message, prompt.placeholder ?? "")) ?? "",
        // Fires only once pi-ai already has a code (its exchange step) -- the
        // race-winner boundary. Dismiss the losing manual dialog NOW, before the
        // token exchange, rather than waiting on it (review mrf68z4h).
        onProgress: (message) => {
          dismissManual();
          ctx.ui.notify(message, "info");
        },
        onManualCodeInput: () =>
          readCode(
            `Open this URL to authenticate Claude:\n${authorizeUrl}\n\nPaste the redirect URL or code`,
            "http://localhost:53692/callback?...",
          ),
      }),
    );
  }

  let authorizeUrl: string | undefined;
  return loginWithManualInputCleanup(ctx, (readCode, dismissManual) =>
    loginOpenAICodex({
      onAuth: ({ url, instructions }) => {
        authorizeUrl = url;
        // Best-effort for the local attach case where a browser is reachable; on a
        // headless remote box this opens nothing and the URL embedded in the
        // manual-code dialog title below (rendered as a tappable link on web) is
        // the real path.
        openBrowser(url);
        if (instructions) ctx.ui.notify(instructions, "info");
      },
      onPrompt: async (prompt) => (await ctx.ui.input(prompt.message, prompt.placeholder ?? "")) ?? "",
      // pi-ai's OpenAI browser path emits no progress between callback-win and
      // exchange, so this dismisses at the exchange step when present; the
      // finally-abort backstop covers the (bounded-by-dialog-timeout) gap.
      onProgress: (message) => {
        dismissManual();
        ctx.ui.notify(message, "info");
      },
      onManualCodeInput: () =>
        readCode(
          `Open this URL to authenticate OpenAI Codex:\n${authorizeUrl}\n\nPaste the redirect URL or code`,
          "http://localhost:1455/auth/callback?...",
        ),
    }),
  );
}

export default async function (pi: ExtensionAPI): Promise<void> {
  extensionAPI = pi;

  pi.registerProvider(ANTHROPIC_PROVIDER_ID, {
    api: "anthropic-messages",
    streamSimple: streamManagedProvider,
  });

  pi.registerProvider(OPENAI_CODEX_PROVIDER_ID, {
    api: "openai-codex-responses",
    streamSimple: streamManagedProvider,
  });

  pi.on("session_start", async (_event, ctx) => {
    runtimeContext = ctx;
    const restored = readManagedProviderRotationPolicy(ctx.ui).allowCrossProviderFallback
      && await restorePreferredModelIfPossible(pi, ctx);
    if (restored && ctx.model) {
      logRotation(`restored preferred ${ctx.model.provider}/${ctx.model.id}`);
      setStatus(formatStatusLine(ctx.model.provider as ManagedProviderId));
    }
    updateRotationStatus(ctx);
  });

  pi.on("before_agent_start", async (_event, ctx) => {
    runtimeContext = ctx;
    const restored = readManagedProviderRotationPolicy(ctx.ui).allowCrossProviderFallback
      && await restorePreferredModelIfPossible(pi, ctx);
    if (restored && ctx.model) {
      logRotation(`restored preferred ${ctx.model.provider}/${ctx.model.id}`);
      setStatus(formatStatusLine(ctx.model.provider as ManagedProviderId));
    }
  });

  pi.on("model_select", async (_event, ctx) => {
    runtimeContext = ctx;
    updateRotationStatus(ctx);
  });

  pi.registerCommand("provider-sub", {
    description: "Manage subscription credential pools for provider rotation",
    handler: async (args, ctx) => {
      const [providerArg, verb, ...rest] = parseCommand(args);
      const providerId = parseProvider(providerArg);
      if (!providerId) {
        ctx.ui.notify("Usage: /provider-sub <anthropic|openai-codex> <list|add|select|rm>", "error");
        return;
      }

      const action = verb === "rm" ? "remove" : verb || "list";

      if (action === "status" || action === "list") {
        const pool = readSubscriptionPool(providerId);
        const lines = [formatStatusLine(providerId), ""];
        for (const [index, entry] of pool.entries()) {
          const cooldown = entry.rateLimitedUntil > Date.now() ? ` cooldown=${Math.ceil((entry.rateLimitedUntil - Date.now()) / 1000)}s` : "";
          const account = entry.accountId ? ` account=${shortAccountId(entry.accountId)}` : "";
          lines.push(`${index + 1}. ${entry.label}${account}${cooldown}`);
        }
        ctx.ui.notify(lines.join("\n"), "info");
        return;
      }

      if (action === "add") {
        if (!ctx.hasUI) {
          ctx.ui.notify(`${getProviderLabel(providerId)} login requires an interactive UI.`, "error");
          return;
        }
        const pool = readSubscriptionPool(providerId);
        const label = (rest.join(" ") || (await ctx.ui.input(`${getProviderLabel(providerId)} subscription label`, providerId)) || "").trim();
        if (!label) return;
        // The default label (`label === providerId`) is the OAuth-login-linked default slot,
        // whose live value lives exclusively in auth.json (design §1/§3). Accepting it bypasses
        // the duplicate-label check because the metadata-only pool entry for it may already exist,
        // and the value never gets written into the pool: `writeSubscriptionPool` strips any
        // `label === providerId` value on write so the login credential stays in auth.json.
        const isDefaultLabel = label.toLowerCase() === providerId.toLowerCase();
        if (!isDefaultLabel && pool.some((entry) => entry.label.toLowerCase() === label.toLowerCase())) {
          ctx.ui.notify(`${getProviderLabel(providerId)} subscription already exists: ${label}`, "error");
          return;
        }
        try {
          const cred = await activeLoginForProvider(providerId, ctx);
          const accountId = loginAccountId(cred);
          if (isDefaultLabel) {
            // Atomically commit the default login: validate account-uniqueness (excluding the
            // default slot), persist auth.json, then reflect identity and causally reconcile the
            // metadata-only cooldown under the pool lock. A collision rejects before auth.json
            // switches, and a newer same-account 429 recorded during OAuth remains authoritative.
            commitDefaultIdentity(providerId, {
              refresh: cred.refresh,
              access: cred.access,
              expires: cred.expires,
              accountId,
              lastRateLimitedAt: pool.find((entry) => entry.label === providerId)?.lastRateLimitedAt ?? 0,
            });
          } else {
            // Enforce label + Codex account uniqueness ATOMICALLY inside the pool lock: a sticky
            // browser session (or two brokers finishing OAuth together) must not land the same
            // quota under a second label. addSubscription throws DuplicateSubscriptionError,
            // surfaced by the catch below. The account check also catches a collision with the
            // default slot. Anthropic has no programmatic account identity, so its duplicate
            // protection stays a validation-procedure obligation, not a code check.
            addSubscription(providerId, {
              label,
              refresh: cred.refresh,
              access: cred.access,
              expires: cred.expires,
              ...(accountId ? { accountId } : {}),
              rateLimitedUntil: 0,
              lastAttemptAt: 0,
              lastRateLimitedAt: 0,
            });
          }
          logRotation(`added ${getProviderLabel(providerId)} subscription "${label}"`);
          ctx.ui.notify(`Added ${getProviderLabel(providerId)} subscription: ${label}`, "info");
          updateRotationStatus(ctx);
        } catch (error) {
          ctx.ui.notify(error instanceof Error ? error.message : String(error), "error");
        }
        return;
      }

      if (action === "select") {
        const ref = rest.join(" ");
        if (!ref) {
          ctx.ui.notify(`Usage: /provider-sub ${providerId} select <label|index>`, "error");
          return;
        }
        try {
          promoteSubscription(providerId, ref);
          logRotation(`selected ${getProviderLabel(providerId)} subscription "${ref}"`);
          ctx.ui.notify(`Preferred ${getProviderLabel(providerId)} subscription: ${ref}`, "info");
          updateRotationStatus(ctx);
        } catch (error) {
          ctx.ui.notify(error instanceof Error ? error.message : String(error), "error");
        }
        return;
      }

      if (action === "remove") {
        const ref = rest.join(" ");
        if (!ref) {
          ctx.ui.notify(`Usage: /provider-sub ${providerId} rm <label|index>`, "error");
          return;
        }
        try {
          removeManagedAccount(providerId, ref);
          logRotation(`removed ${getProviderLabel(providerId)} subscription "${ref}"`);
          ctx.ui.notify(`Removed ${getProviderLabel(providerId)} subscription: ${ref}`, "info");
          updateRotationStatus(ctx);
        } catch (error) {
          ctx.ui.notify(error instanceof Error ? error.message : String(error), "error");
        }
        return;
      }

      ctx.ui.notify(`Unknown /provider-sub verb: ${verb}`, "error");
    },
  });
}