import { homedir } from "node:os";
import { fireAndForget } from "./async-guard";
import { existsSync, readdirSync } from "node:fs";
import { createServer } from "node:net";
import { join } from "node:path";
import { getManifest, type ProviderQuotaPollDescriptor } from "agent-relay-providers";
import {
  DEFAULT_PROVIDER_QUOTA_CONFIG,
  QUOTA_FAILURE_LOG_INTERVAL_MS,
  QUOTA_FAST_RETRY_MS,
  QuotaCollectionError,
  codexAutoFireDecision,
  codexRateLimitsResetCredits,
  codexResetConsumeBurnedCredit,
  collectCodexQuotaSample,
  mapCodexResetConsumeOutcome,
  normalizeProviderQuotaConfig,
  providerQuotaErrorFromCollectorError,
  quotaRetryAfterMs,
  resolveStableCodexQuotaIdentityFromHome,
  type ProviderQuotaIdentity,
  type ProviderQuotaSample,
} from "agent-relay-sdk/provider-quota";
import type { CodexResetConsumeOutcome, CodexResetConsumeResult } from "agent-relay-sdk";
import type { ProviderQuotaConfig, ProviderQuotaConfigMap, ProviderQuotaLeaseAcquireInput, ProviderQuotaUpdateInput } from "agent-relay-sdk";
import { errMessage } from "agent-relay-sdk";
import { providerCommandFromEnv, providerHomeRootFromEnv, type OrchestratorConfig } from "./config";

const QUOTA_LEASE_TTL_MS = 90_000;
const QUOTA_LEASE_RENEW_MS = 30_000;
const QUOTA_RETRY_BACKOFF_MAX_MS = 15 * 60_000;
const QUOTA_RETRY_BACKOFF_MIN_MS = 1_000;
const CODEX_APP_SERVER_CONNECT_ATTEMPTS = 40;
const CODEX_APP_SERVER_CONNECT_RETRY_MS = 250;
// #1425 finding 3 — hard ceilings so no websocket RPC can hang forever. The consume ceiling bounds
// the consume round-trip at the poller boundary (covers any injected transport); the per-request
// ceiling bounds each individual live JSON-RPC call; the connect ceiling bounds a single websocket
// connect attempt (an unbounded connect could otherwise stall the poll tick); the read ceiling
// bounds the whole quota-READ round-trip (connect + initialize + read) so a hung read can't stall
// the tick either.
const CODEX_CONSUME_TIMEOUT_MS = 20_000;
const CODEX_RPC_REQUEST_TIMEOUT_MS = 15_000;
const CODEX_CONNECT_TIMEOUT_MS = 10_000;
const CODEX_QUOTA_READ_TIMEOUT_MS = 45_000;

// A provider that is configured but has no usable quota credential on this host
// is uncollectable by design. Rather than silently omitting it (so the row
// vanishes from the dashboard), we report a marker record carrying the reason,
// so the widget can show a muted "unavailable" row.
// A stable synthetic account key keeps it out of the host:/home: unstable-key prune.
const PROVIDER_QUOTA_UNAVAILABLE_ACCOUNT_KEY = "unavailable";
const PROVIDER_QUOTA_UNAVAILABLE_ERROR_TYPE = "unavailable";

type QuotaRelay = {
  acquireProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput): Promise<{
    acquired: boolean;
    lease?: { leaseToken: string; expiresAt: number };
    retryAfterMs?: number;
  }>;
  releaseProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput & { leaseToken: string }): Promise<unknown>;
  reportProviderQuota(input: ProviderQuotaUpdateInput): Promise<unknown>;
  getProviderQuotaConfig(): Promise<ProviderQuotaConfigMap>;
  connected: boolean;
};

type QuotaCandidate = ProviderQuotaIdentity & {
  accessToken?: string;
  appServerUrl?: string;
  codexHome?: string;
};

type ProviderSkip = { provider: string; reason: string };
type QuotaDiscovery = { candidates: QuotaCandidate[]; skips: ProviderSkip[] };

type QuotaPollState = {
  leaseToken?: string;
  leaseExpiresAt?: number;
  nextPollAt?: number;
  lastAttemptAt?: number;
  consecutiveFailures?: number;
  lastLog?: { key: string; at: number };
};

// #1425 — the raw inputs/outputs of the consume transport. The transport is responsible
// only for moving bytes over the app-server websocket; the poller does the outcome/count
// mapping via the SDK's pure helpers so mapping stays uniform and unit-tested.
export type CodexResetConsumeTransportInput = {
  provider: string;
  idempotencyKey: string;
  appServerUrl?: string;
  codexHome?: string;
};
export type CodexResetConsumeTransportResult = {
  /** Raw `account/rateLimitResetCredit/consume` response. */
  consume: unknown;
  /** Optional fresh `account/rateLimits/read` response for the post-consume banked count. */
  rateLimits?: unknown;
};

// #1425 — emitted whenever a reset is actually consumed (auto or manual), for observability.
export type CodexResetFiredEvent = {
  provider: string;
  accountKey: string;
  trigger: "auto" | "manual";
  outcome: CodexResetConsumeOutcome;
  burned: boolean;
  idempotencyKey: string;
  availableCount?: number;
  reason?: string;
};

// #1425 — the codex 7d (weekly) window is the "secondary" window in the manifest; the 5h
// window is "primary" (see provider-quota-burn quotaWindowRole). Auto-fire keys off the 7d
// window's utilization + reset time.
const CODEX_SEVEN_DAY_WINDOW_NAME = "secondary";

export class OrchestratorQuotaPoller {
  private timer?: Timer;
  private active = false;
  private inFlight = false;
  private readonly states = new Map<string, QuotaPollState>();
  private readonly logStates = new Map<string, { key: string; at: number }>();
  // Per-provider quota config (#605), refreshed each tick. Empty until the first
  // successful fetch — configFor() falls back to defaults so a fresh/empty config
  // (or an older relay without the endpoint) preserves today's behavior.
  private quotaConfig: ProviderQuotaConfigMap = {};

  constructor(
    private readonly config: OrchestratorConfig,
    private readonly relay: QuotaRelay,
    private readonly options: {
      intervalMs?: number;
      fetchImpl?: typeof fetch;
      now?: () => number;
      log?: (message: string) => void;
      codexRateLimitsRead?: (appServerUrl: string) => Promise<unknown>;
      // #1425 — injectable consume transport. MUST be mocked in every test: the real RPC
      // irreversibly burns one of the owner's scarce banked resets. Returns the raw consume
      // payload plus (optionally) a fresh rate-limits read for the post-consume count.
      codexResetCreditConsume?: (input: CodexResetConsumeTransportInput) => Promise<CodexResetConsumeTransportResult>;
      // #1425 — surfaced observably alongside the log line so the owner sees a reset was
      // consumed and why. Best-effort; a throw here never blocks the poll.
      onResetFired?: (event: CodexResetFiredEvent) => void | Promise<void>;
      // #1425 finding 3 — bound the consume RPC so a hung transport (websocket that never
      // answers) can never stall the poll tick. Overridable for tests; defaults below.
      rpcTimeoutMs?: number;
    } = {},
  ) {}

  // #1425 — in-memory at-most-once-per-window guard. The idempotency key is stable per
  // window (account + window id), so this survives rapid re-polls; a process restart is
  // backstopped by the server-side consume idempotency (a redelivered key returns
  // `alreadyRedeemed`, never a second burn).
  private readonly firedAutoResetKeys = new Set<string>();

  // #1425 finding 5 — set true whenever the per-tick config refresh FAILS. Collection keeps
  // running on the last-known config (a transient relay blip must not stop polling), but
  // auto-fire fails CLOSED while stale: we never burn a scarce reset off a config we could
  // not confirm still says "enabled".
  private quotaConfigStale = false;

  start(): void {
    if (this.active) return;
    this.active = true;
    this.schedule(1_000);
  }

  stop(): void {
    this.active = false;
    if (this.timer) clearTimeout(this.timer);
    this.timer = undefined;
    for (const [key, state] of this.states) {
      if (!state.leaseToken) continue;
      const [provider, accountKey] = splitStateKey(key);
      void this.relay.releaseProviderQuotaLease(this.config.id, { provider, accountKey, leaseToken: state.leaseToken }).catch(() => {});
    }
    this.states.clear();
    this.logStates.clear();
  }

  async tick(): Promise<void> {
    if (this.inFlight || !this.active || !this.relay.connected) return;
    this.inFlight = true;
    try {
      await this.refreshQuotaConfig();
      const { candidates, skips } = await this.discoverCandidates();
      await this.releaseRemovedCandidates(candidates);
      for (const candidate of candidates) {
        try {
          await this.processCandidate(candidate);
        } catch (error) {
          await this.handleCandidateFailure(candidate, error);
        }
      }
      for (const skip of skips) {
        await this.reportSkip(skip);
      }
    } finally {
      this.inFlight = false;
      this.schedule(this.nextScheduleDelay(this.options.intervalMs ?? QUOTA_LEASE_RENEW_MS));
    }
  }

  private schedule(delayMs: number): void {
    if (this.timer) clearTimeout(this.timer);
    this.timer = undefined;
    if (!this.active) return;
    this.timer = setTimeout(() => {
      this.timer = undefined;
      void fireAndForget("Provider quota poll", () => this.tick());
    }, Math.max(1_000, delayMs));
  }

  private nextScheduleDelay(defaultDelayMs: number): number {
    const now = this.now();
    let delayMs = defaultDelayMs;
    for (const state of this.states.values()) {
      if (state.nextPollAt !== undefined) delayMs = Math.min(delayMs, state.nextPollAt - now);
      if (state.leaseToken && state.leaseExpiresAt !== undefined) {
        delayMs = Math.min(delayMs, state.leaseExpiresAt - now - QUOTA_LEASE_RENEW_MS);
      }
    }
    return Math.max(QUOTA_RETRY_BACKOFF_MIN_MS, delayMs);
  }

  // Refresh per-provider quota config (#605). Best-effort for COLLECTION: on failure we keep
  // the last known config (defaults for any unset provider), so a transient relay blip never
  // silently stops collection. But for AUTO-FIRE we fail CLOSED (#1425 finding 5): a refresh
  // failure marks the config stale, and maybeAutoFireReset refuses to fire while stale — so a
  // transient fetch failure can never let a since-disabled auto-fire keep burning resets off a
  // stale `enabled: true`.
  private async refreshQuotaConfig(): Promise<void> {
    try {
      this.quotaConfig = await this.relay.getProviderQuotaConfig();
      this.quotaConfigStale = false;
    } catch {
      // keep prior config for collection cadence; configFor() defaults any missing provider.
      this.quotaConfigStale = true;
    }
  }

  private configFor(provider: string): ProviderQuotaConfig {
    const stored = this.quotaConfig[provider];
    return stored ? normalizeProviderQuotaConfig(stored) : { ...DEFAULT_PROVIDER_QUOTA_CONFIG };
  }

  private async discoverCandidates(): Promise<QuotaDiscovery> {
    const candidates: QuotaCandidate[] = [];
    const skips: ProviderSkip[] = [];
    // A disabled provider (#605) is collected from at all: no discovery → no
    // polling/API calls, leases released by releaseRemovedCandidates, and no
    // skip-marker row (disabled is intentional, not a credential failure).
    for (const provider of this.config.providers) {
      if (!this.configFor(provider).enabled) continue;
      const found = await this.discoverProviderCandidates(provider);
      candidates.push(...found.candidates);
      if (found.skipReason) skips.push({ provider, reason: found.skipReason });
    }
    const deduped = new Map<string, QuotaCandidate>();
    for (const candidate of candidates) {
      deduped.set(candidateStateKey(candidate), candidate);
    }
    return { candidates: [...deduped.values()], skips };
  }

  private async discoverProviderCandidates(provider: string): Promise<{ candidates: QuotaCandidate[]; skipReason?: string }> {
    const manifest = getManifest(provider);
    const quotaPoll = manifest?.quotaPoll;
    if (!quotaPoll || quotaPoll.strategy === "none") return { candidates: [] };
    if (quotaPoll.strategy === "codex-app-server") return this.discoverCodexCandidates(provider, quotaPoll);
    return { candidates: [] };
  }

  private async discoverCodexCandidates(provider: string, _quotaPoll: ProviderQuotaPollDescriptor): Promise<{ candidates: QuotaCandidate[]; skipReason?: string }> {
    const manifest = getManifest(provider);
    const providerLabel = manifest?.label ?? provider;
    const markerFile = manifest?.home?.quotaCredentialFile ?? "auth.json";
    const homes = [
      this.providerHomeDir(provider),
      ...providerHomeConfigDirs(provider, markerFile),
    ];
    const envPrefix = manifest?.home?.envPrefix ?? provider.toUpperCase().replace(/[^A-Z0-9]+/g, "_");
    const appServerUrl = this.config.env[`${envPrefix}_APP_SERVER_URL`] || process.env[`${envPrefix}_APP_SERVER_URL`];
    const candidates: QuotaCandidate[] = [];
    for (const codexHome of homes) {
      const identity = await resolveStableCodexQuotaIdentityFromHome({ codexHome });
      if (identity) candidates.push({ ...identity, codexHome, ...(appServerUrl ? { appServerUrl } : {}) });
    }
    if (candidates.length === 0) {
      const reason = `no ${providerLabel} account id found in ${markerFile}`;
      this.logOnce(`${provider}:no-stable-auth`, `quota refresh skipped for ${provider}: ${reason}`);
      return { candidates, skipReason: reason };
    }
    return { candidates };
  }

  // Publish a marker record for a provider that is configured but uncollectable on
  // this host, so the dashboard shows a muted "unavailable" row with the reason
  // instead of dropping the provider entirely. Re-sent each tick to keep the row
  // fresh (the day-old prune is keyed on updated_at).
  private async reportSkip(skip: ProviderSkip): Promise<void> {
    await this.relay.reportProviderQuota({
      provider: skip.provider,
      accountKey: PROVIDER_QUOTA_UNAVAILABLE_ACCOUNT_KEY,
      lastAttemptAt: this.now(),
      lastError: { type: PROVIDER_QUOTA_UNAVAILABLE_ERROR_TYPE, message: skip.reason },
      sourceAgentId: this.sourceAgentId(),
    }).catch((publishError) => this.log(`quota skip publish failed for ${skip.provider}: ${errMessage(publishError)}`));
  }

  private async releaseRemovedCandidates(candidates: QuotaCandidate[]): Promise<void> {
    const live = new Set(candidates.map(candidateStateKey));
    for (const [key, state] of this.states) {
      if (live.has(key)) continue;
      this.states.delete(key);
      if (!state.leaseToken) continue;
      const [provider, accountKey] = splitStateKey(key);
      await this.relay.releaseProviderQuotaLease(this.config.id, { provider, accountKey, leaseToken: state.leaseToken }).catch(() => {});
    }
  }

  private async processCandidate(candidate: QuotaCandidate): Promise<void> {
    const state = this.stateFor(candidate);
    const now = this.now();
    if (!await this.ensureLease(candidate, state, now)) return;
    if (state.nextPollAt !== undefined && state.nextPollAt > now) return;

    // Per-provider cadence (#605): the configured interval governs the gap between
    // successful polls and the post-failure retry once a first attempt has landed.
    const pollIntervalMs = this.configFor(candidate.provider).pollIntervalMs;
    const lastAttemptAt = now;
    try {
      const sample = await this.collect(candidate);
      const quota = sample.quota;
      if (!quota) throw new QuotaCollectionError("creds_not_ready", `${candidate.provider} quota source is not ready`);
      const update: ProviderQuotaUpdateInput = {
        provider: candidate.provider,
        accountKey: sample.accountKey ?? candidate.accountKey,
        quota,
        ...(sample.unavailableWindows?.length ? { unavailableWindows: sample.unavailableWindows } : {}),
        lastAttemptAt: quota.updatedAt,
        sourceAgentId: this.sourceAgentId(),
      };
      await this.relay.reportProviderQuota(update);
      state.lastAttemptAt = update.lastAttemptAt;
      state.nextPollAt = now + pollIntervalMs;
      state.consecutiveFailures = 0;
      // #1425 — auto-fire evaluation runs AFTER a successful poll+report, using the sample we
      // just collected. It is a no-op unless the owner has opted in (default OFF), and any
      // failure inside it is swallowed so it can never disturb the poll cadence.
      await this.maybeAutoFireReset(candidate, sample, now).catch((autoFireError) =>
        this.log(`codex reset auto-fire error for ${candidate.provider}/${candidate.accountKey}: ${errMessage(autoFireError)}`));
    } catch (error) {
      const retryAfterMs = quotaRetryAfterMs(error);
      const lastError = providerQuotaErrorFromCollectorError(error, retryAfterMs);
      const retryDelayMs = this.retryDelayMs(state, retryAfterMs);
      state.lastAttemptAt = lastAttemptAt;
      state.nextPollAt = now + retryDelayMs;
      state.consecutiveFailures = (state.consecutiveFailures ?? 0) + 1;
      await this.relay.reportProviderQuota({
        provider: candidate.provider,
        accountKey: candidate.accountKey,
        lastAttemptAt,
        lastError,
        sourceAgentId: this.sourceAgentId(),
      }).catch((publishError) => this.log(`quota status publish failed: ${errMessage(publishError)}`));
      this.logFailure(candidate, error, retryAfterMs, retryDelayMs);
    }
  }

  // #1425 — evaluate the auto-fire policy for a freshly-collected codex sample and, if every
  // guard passes, consume a banked reset. DEFAULT OFF: the very first check is the opt-in flag,
  // so with auto-fire disabled this returns before touching the sample or the consume RPC.
  private async maybeAutoFireReset(candidate: QuotaCandidate, sample: ProviderQuotaSample, now: number): Promise<void> {
    if (getManifest(candidate.provider)?.quotaPoll?.strategy !== "codex-app-server") return;
    // #1425 finding 5 — fail closed: never auto-fire off a config we could not freshly confirm.
    if (this.quotaConfigStale) return;
    const config = this.configFor(candidate.provider);
    if (!config.autoFireResetEnabled) return; // opt-in gate — zero behavior change while off.

    const sevenDay = sample.quota?.windows?.find((window) => window.name === CODEX_SEVEN_DAY_WINDOW_NAME);
    const resetsAt = typeof sevenDay?.resetsAt === "number" ? sevenDay.resetsAt : null;
    const decision = codexAutoFireDecision({
      config: {
        enabled: config.autoFireResetEnabled,
        threshold: config.autoFireResetThreshold ?? DEFAULT_PROVIDER_QUOTA_CONFIG.autoFireResetThreshold ?? 0.99,
      },
      // #1425 finding 4 — key the idempotency identity off the STABLE credential-derived
      // candidate identity (canonical `account:<id>`) EXCLUSIVELY, never the raw upstream
      // `sample.accountKey`. If the upstream string representation drifts between polls or
      // restarts, the per-window key must not drift with it (that would re-fire in-window).
      accountKey: candidate.accountKey,
      utilization: sevenDay?.utilization ?? null,
      availableCount: sample.resetCredits ?? null,
      timeToResetMs: resetsAt !== null ? resetsAt - now : null,
      windowId: resetsAt !== null ? String(resetsAt) : null,
      alreadyFired: (key) => this.firedAutoResetKeys.has(key),
    });
    if (!decision.fire) return;

    // Optimistic mark BEFORE the await: this is the strongest guard against a double-fire — a
    // re-entrant/rapid poll observing the same window sees the key already present. The
    // server-side idempotency key is the second line of defense across a process restart.
    this.firedAutoResetKeys.add(decision.idempotencyKey);
    await this.consumeResetCredit(candidate, decision.idempotencyKey, "auto", candidate.accountKey, now);
  }

  // #1425 — run the consume RPC, map the outcome + refreshed banked count, log it observably,
  // push the fresh count to the relay, and fire the onResetFired hook. Shared by the auto-fire
  // path and the manual (MCP/dashboard) path so both record identically.
  private async consumeResetCredit(
    candidate: QuotaCandidate,
    idempotencyKey: string,
    trigger: "auto" | "manual",
    accountKey: string,
    now = this.now(),
  ): Promise<CodexResetConsumeResult> {
    // #1425 finding 1 — the class default is a SAFE STUB that THROWS. The live transport is
    // injectable ONLY at the production entrypoint (orchestrator/src/index.ts). So a test/dev
    // path that forgets to mock — or a dev orchestrator that somehow reaches this with auto-fire
    // opted in — cannot touch the destructive live RPC by construction; it errors instead.
    const transport = this.options.codexResetCreditConsume ?? throwingConsumeTransport;
    // #1425 finding 3 — bound the CONSUME round-trip (connect + initialize + consume) so a hung
    // transport can never stall the poll tick (auto-fire) or the manual HTTP response. On timeout
    // this rejects; the in-flight fire is already recorded (auto: firedAutoResetKeys; manual: relay
    // dedupe), and the stable idempotency key means any retry dedupes on the codex side rather than
    // double-burning. The transport does the consume ONLY (no chained refresh under this ceiling), so
    // once it resolves the fire is committed IMMEDIATELY below — the banked count is taken from the
    // consume response itself (falling back to the next natural poll), never a refresh that the outer
    // timeout could discard.
    const raw = await withTimeout(
      transport({
        provider: candidate.provider,
        idempotencyKey,
        ...(candidate.appServerUrl ? { appServerUrl: candidate.appServerUrl } : {}),
        ...(candidate.codexHome ? { codexHome: candidate.codexHome } : {}),
      }),
      this.options.rpcTimeoutMs ?? CODEX_CONSUME_TIMEOUT_MS,
      `codex reset ${trigger}-consume`,
    );
    // A mock transport may still return an inline `rateLimits` refresh (instant); the live transport
    // returns consume only, so the count is read from the consume payload and refreshed by the poll.
    const outcome = mapCodexResetConsumeOutcome(raw.consume);
    const availableCount = codexRateLimitsResetCredits(raw.rateLimits ?? raw.consume);
    const burned = codexResetConsumeBurnedCredit(outcome);
    const result: CodexResetConsumeResult = {
      outcome,
      idempotencyKey,
      ...(availableCount !== null ? { availableCount } : {}),
    };

    // Observable log line so the owner can see a reset was consumed (and why).
    this.log(
      `codex reset ${trigger}-fire → ${outcome}${burned ? " (consumed a banked reset)" : ""}` +
      `${availableCount !== null ? `; ${availableCount} banked remaining` : ""} [${idempotencyKey}]`,
    );

    // Push the refreshed banked count to the relay so the dashboard reflects the consumption.
    if (availableCount !== null) {
      await this.relay.reportProviderQuota({
        provider: candidate.provider,
        accountKey,
        codexResetCredits: { availableCount, updatedAt: now },
        lastAttemptAt: now,
        sourceAgentId: this.sourceAgentId(),
      }).catch((publishError) => this.log(`codex reset count publish failed: ${errMessage(publishError)}`));
    }

    // Best-effort observability hook (deploy-event / notification). Never blocks or throws.
    try {
      await this.options.onResetFired?.({
        provider: candidate.provider,
        accountKey,
        trigger,
        outcome,
        burned,
        idempotencyKey,
        ...(availableCount !== null ? { availableCount } : {}),
      });
    } catch (hookError) {
      this.log(`codex reset onResetFired hook error: ${errMessage(hookError)}`);
    }

    return result;
  }

  // #1425 — manual (owner-driven) trigger entry point, called by the orchestrator API route on
  // behalf of the relay MCP tool / dashboard button. Resolves a codex candidate on this host,
  // fires exactly one consume with the caller-supplied idempotency key, and returns the outcome
  // plus the refreshed banked count.
  async manualConsumeResetCredit(input: { provider?: string; accountKey?: string; idempotencyKey: string }): Promise<CodexResetConsumeResult & { accountKey: string; provider: string }> {
    const candidate = await this.resolveCodexCandidate(input.accountKey, input.provider);
    if (!candidate) throw new Error("no codex quota source available on this host");
    const result = await this.consumeResetCredit(candidate, input.idempotencyKey, "manual", candidate.accountKey);
    return { ...result, accountKey: candidate.accountKey, provider: candidate.provider };
  }

  // #1425 finding 4 — resolve the consume candidate. When the caller names a provider, resolve
  // ONLY that provider's candidate (so the relay's declared reset-credit provider is honored and a
  // request can never resolve a different provider's account). Any resolvable provider must still
  // actually speak the codex-app-server transport (the only surface exposing the consume RPC).
  private async resolveCodexCandidate(accountKey?: string, provider?: string): Promise<QuotaCandidate | undefined> {
    const candidates: QuotaCandidate[] = [];
    for (const configuredProvider of this.config.providers) {
      if (provider && configuredProvider !== provider) continue;
      if (getManifest(configuredProvider)?.quotaPoll?.strategy !== "codex-app-server") continue;
      const found = await this.discoverProviderCandidates(configuredProvider);
      candidates.push(...found.candidates);
    }
    if (accountKey) return candidates.find((candidate) => candidate.accountKey === accountKey);
    return candidates[0];
  }

  private async handleCandidateFailure(candidate: QuotaCandidate, error: unknown): Promise<void> {
    const state = this.stateFor(candidate);
    const now = this.now();
    const retryAfterMs = quotaRetryAfterMs(error);
    const retryDelayMs = this.retryDelayMs(state, retryAfterMs);
    state.lastAttemptAt = now;
    state.nextPollAt = now + retryDelayMs;
    state.consecutiveFailures = (state.consecutiveFailures ?? 0) + 1;
    await this.relay.reportProviderQuota({
      provider: candidate.provider,
      accountKey: candidate.accountKey,
      lastAttemptAt: now,
      lastError: providerQuotaErrorFromCollectorError(error, retryAfterMs),
      sourceAgentId: this.sourceAgentId(),
    }).catch((publishError) => this.log(`quota status publish failed: ${errMessage(publishError)}`));
    this.logFailure(candidate, error, retryAfterMs, retryDelayMs);
  }

  private retryDelayMs(state: QuotaPollState, retryAfterMs: number | undefined): number {
    const baseDelayMs = retryAfterMs ?? QUOTA_FAST_RETRY_MS;
    const multiplier = 2 ** Math.min(state.consecutiveFailures ?? 0, 10);
    return Math.min(
      QUOTA_RETRY_BACKOFF_MAX_MS,
      Math.max(QUOTA_RETRY_BACKOFF_MIN_MS, Math.round(baseDelayMs * multiplier)),
    );
  }

  private async ensureLease(candidate: QuotaCandidate, state: QuotaPollState, now: number): Promise<boolean> {
    if (state.leaseToken && state.leaseExpiresAt && state.leaseExpiresAt - now > QUOTA_LEASE_RENEW_MS) return true;
    const result = await this.relay.acquireProviderQuotaLease(this.config.id, {
      provider: candidate.provider,
      accountKey: candidate.accountKey,
      ...(state.leaseToken ? { leaseToken: state.leaseToken } : {}),
      ttlMs: QUOTA_LEASE_TTL_MS,
    });
    if (!result.acquired || !result.lease) {
      state.leaseToken = undefined;
      state.leaseExpiresAt = undefined;
      state.nextPollAt = now + Math.min(result.retryAfterMs ?? QUOTA_LEASE_RENEW_MS, QUOTA_LEASE_RENEW_MS);
      return false;
    }
    state.leaseToken = result.lease.leaseToken;
    state.leaseExpiresAt = result.lease.expiresAt;
    return true;
  }

  private async collect(candidate: QuotaCandidate): Promise<ProviderQuotaSample> {
    const strategy = getManifest(candidate.provider)?.quotaPoll?.strategy;
    if (strategy === "codex-app-server") {
      return collectCodexQuotaSample({
        agentId: this.sourceAgentId(),
        now: this.now(),
        // #1425 finding 3 — bound the WHOLE quota-read round-trip (connect + initialize + read) so a
        // hung read can never stall the poll tick. connectWithRetry bounds each connect attempt too;
        // this enclosing ceiling backstops the full sequence.
        rateLimitsRead: () => withTimeout(
          this.options.codexRateLimitsRead
            ? this.options.codexRateLimitsRead(candidate.appServerUrl ?? "")
            : candidate.appServerUrl
              ? codexRateLimitsRead(candidate.appServerUrl)
              : codexRateLimitsReadFromHome(candidate.provider, candidate.codexHome),
          CODEX_QUOTA_READ_TIMEOUT_MS,
          "codex quota read",
        ),
      });
    }
    return {};
  }

  private stateFor(candidate: QuotaCandidate): QuotaPollState {
    const key = candidateStateKey(candidate);
    const existing = this.states.get(key);
    if (existing) return existing;
    const created: QuotaPollState = {};
    this.states.set(key, created);
    return created;
  }

  private sourceAgentId(): string {
    return `orchestrator-${this.config.id}`;
  }

  private now(): number {
    return this.options.now?.() ?? Date.now();
  }

  private log(message: string): void {
    (this.options.log ?? ((line) => console.error(`[orchestrator] ${line}`)))(message);
  }

  private logFailure(candidate: QuotaCandidate, error: unknown, retryAfterMs: number | undefined, retryDelayMs: number): void {
    const state = this.stateFor(candidate);
    const key = retryAfterMs !== undefined ? `retry-after:${retryAfterMs}:delay:${retryDelayMs}` : `${errMessage(error)}:delay:${retryDelayMs}`;
    const now = this.now();
    if (state.lastLog?.key === key && now - state.lastLog.at < QUOTA_FAILURE_LOG_INTERVAL_MS) return;
    state.lastLog = { key, at: now };
    const suffix = `; retrying in ${Math.round(retryDelayMs / 1000)}s`;
    this.log(`quota refresh failed for ${candidate.provider}/${candidate.accountKey}${suffix}: ${errMessage(error)}`);
  }

  private logOnce(key: string, message: string): void {
    const lastLog = this.logStates.get(key);
    const now = this.now();
    if (lastLog && now - lastLog.at < QUOTA_FAILURE_LOG_INTERVAL_MS) return;
    this.logStates.set(key, { key, at: now });
    this.log(message);
  }

  private providerHomeDir(provider: string): string {
    const manifest = getManifest(provider);
    const envPrefix = manifest?.home?.envPrefix ?? provider.toUpperCase().replace(/[^A-Z0-9]+/g, "_");
    return this.config.env[`${envPrefix}_HOME`] || process.env[`${envPrefix}_HOME`] || join(homedir(), manifest?.home?.configDir ?? `.${provider}`);
  }
}

// #1425 finding 3 — reject with a clear, labelled error if `promise` does not settle within
// `ms`. Used at the poller boundary (bounding any injected transport) and around each live
// JSON-RPC request. The timer is cleared on settle so a resolved promise leaves nothing pending.
export function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
  return new Promise<T>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(`${label} timed out after ${ms}ms`)), ms);
    promise.then(
      (value) => { clearTimeout(timer); resolve(value); },
      (error) => { clearTimeout(timer); reject(error); },
    );
  });
}

// #1425 finding 1 — the class-default consume transport. It NEVER reaches the live RPC: it
// throws, so tests/dev that forget to inject a mock (or a dev orchestrator that reaches the
// consume path with auto-fire opted in) fail loudly instead of burning a scarce banked reset.
// The production entrypoint injects `defaultCodexResetCreditConsume` explicitly.
async function throwingConsumeTransport(): Promise<never> {
  throw new Error("no codex reset consume transport configured — refusing to reach the live banked-reset RPC");
}

async function codexRateLimitsRead(appServerUrl: string, attempts = 1): Promise<unknown> {
  const client = new JsonRpcWebSocket(appServerUrl);
  try {
    await connectWithRetry(client, attempts);
    await withTimeout(client.request("initialize", {
      clientInfo: { name: "agent-relay-orchestrator", title: "Agent Relay Orchestrator", version: "0.1.0" },
      capabilities: { experimentalApi: true },
    }), CODEX_RPC_REQUEST_TIMEOUT_MS, "codex initialize");
    client.notify("initialized");
    return await withTimeout(client.request("account/rateLimits/read"), CODEX_RPC_REQUEST_TIMEOUT_MS, "codex rateLimits/read");
  } finally {
    client.close();
  }
}

// #1425 — the real consume transport (mocked in every test via the codexResetCreditConsume
// option). Reuses the exact connect/initialize handshake OrchestratorQuotaPoller already uses
// for account/rateLimits/read; adds the consume call + a post-consume re-read for the fresh
// banked count. NEVER exercised by tests — it burns a scarce, irreversible banked reset.
// #1425 finding 1 — the REAL consume transport. Exported so the production entrypoint can inject
// it EXPLICITLY (the class default throws instead). Still never exercised by tests — it burns a
// scarce, irreversible banked reset.
export async function defaultCodexResetCreditConsume(input: CodexResetConsumeTransportInput): Promise<CodexResetConsumeTransportResult> {
  if (input.appServerUrl) return codexResetCreditConsumeOverWs(input.appServerUrl, input.idempotencyKey, 1);
  return codexResetCreditConsumeFromHome(input.provider, input.codexHome, input.idempotencyKey);
}

// #1425 finding 1 — the live consume transport is wired ONLY in a genuine production run, and NEVER
// in the dev-service profile / a test / CI. It is an EXPLICIT, off-by-default opt-in:
//   • the dev service sets AGENT_RELAY_DEV_PROFILE=1 in the orchestrator env → NEVER live (hard gate).
//   • production must explicitly set AGENT_RELAY_CODEX_RESET_CONSUME_LIVE=1 to opt in.
// Anything else (default, tests, CI, an unset flag) resolves to `undefined`, so the poller falls back
// to the throwing stub and the destructive RPC is UNREACHABLE by construction. Returns the live
// transport ONLY when both conditions hold. Env is the merged orchestrator/process env.
export function resolveLiveCodexResetConsumeTransport(
  env: Record<string, string | undefined>,
): ((input: CodexResetConsumeTransportInput) => Promise<CodexResetConsumeTransportResult>) | undefined {
  // Hard gate: the dev-service profile can NEVER reach the live RPC, regardless of any opt-in flag.
  if (env.AGENT_RELAY_DEV_PROFILE === "1") return undefined;
  // Explicit, off-by-default production opt-in.
  if (env.AGENT_RELAY_CODEX_RESET_CONSUME_LIVE !== "1") return undefined;
  return defaultCodexResetCreditConsume;
}

async function codexResetCreditConsumeOverWs(appServerUrl: string, idempotencyKey: string, attempts: number): Promise<CodexResetConsumeTransportResult> {
  const client = new JsonRpcWebSocket(appServerUrl);
  try {
    await connectWithRetry(client, attempts);
    // #1425 finding 3 — each live JSON-RPC call is individually bounded so initialize and consume
    // can never hang forever.
    await withTimeout(client.request("initialize", {
      clientInfo: { name: "agent-relay-orchestrator", title: "Agent Relay Orchestrator", version: "0.1.0" },
      capabilities: { experimentalApi: true },
    }), CODEX_RPC_REQUEST_TIMEOUT_MS, "codex initialize");
    client.notify("initialized");
    // #1425 finding 3 — the transport does the CONSUME ONLY. It deliberately does NOT chain a
    // post-consume rateLimits refresh on the same call: nesting a second 15s RPC under the poller's
    // 20s consume ceiling meant a consume that SUCCEEDED but whose refresh was slow could be
    // discarded by the outer timeout, losing the fire record → a later retry double-burns. The
    // banked count is instead taken from the consume response itself (which carries it) and, failing
    // that, refreshed by the next natural quota poll. So a successful consume is committed the
    // instant it returns, never gated behind a refresh.
    const consume = await withTimeout(
      client.request("account/rateLimitResetCredit/consume", { idempotencyKey }),
      CODEX_RPC_REQUEST_TIMEOUT_MS,
      "codex rateLimitResetCredit/consume",
    );
    return { consume };
  } finally {
    client.close();
  }
}

async function codexResetCreditConsumeFromHome(provider: string, codexHome: string | undefined, idempotencyKey: string): Promise<CodexResetConsumeTransportResult> {
  if (!codexHome) throw new QuotaCollectionError("creds_not_ready", `${provider} home is not configured`);
  const appServerUrl = await freeLoopbackWsUrl();
  const command = providerCommandFromEnv(provider) || getManifest(provider)?.probe?.command || provider;
  const proc = Bun.spawn([command, "app-server", "--listen", appServerUrl], {
    env: {
      ...process.env,
      CODEX_HOME: codexHome,
      CODEX_APP_SERVER_URL: appServerUrl,
    },
    stdin: "ignore",
    stdout: "ignore",
    stderr: "ignore",
  });
  try {
    return await codexResetCreditConsumeOverWs(appServerUrl, idempotencyKey, CODEX_APP_SERVER_CONNECT_ATTEMPTS);
  } finally {
    proc.kill();
    await Promise.race([proc.exited.catch(() => undefined), Bun.sleep(2_000)]).catch(() => undefined);
  }
}

async function codexRateLimitsReadFromHome(provider: string, codexHome: string | undefined): Promise<unknown> {
  if (!codexHome) throw new QuotaCollectionError("creds_not_ready", `${provider} home is not configured`);
  const appServerUrl = await freeLoopbackWsUrl();
  const command = providerCommandFromEnv(provider) || getManifest(provider)?.probe?.command || provider;
  const proc = Bun.spawn([command, "app-server", "--listen", appServerUrl], {
    env: {
      ...process.env,
      CODEX_HOME: codexHome,
      CODEX_APP_SERVER_URL: appServerUrl,
    },
    stdin: "ignore",
    stdout: "ignore",
    stderr: "ignore",
  });
  try {
    return await codexRateLimitsRead(appServerUrl, CODEX_APP_SERVER_CONNECT_ATTEMPTS);
  } finally {
    proc.kill();
    await Promise.race([proc.exited.catch(() => undefined), Bun.sleep(2_000)]).catch(() => undefined);
  }
}

async function connectWithRetry(client: JsonRpcWebSocket, attempts: number): Promise<void> {
  let lastError: unknown;
  for (let attempt = 0; attempt < attempts; attempt += 1) {
    try {
      // #1425 finding 3 — bound each connect attempt. A websocket `connect()` that never resolves
      // its onopen/onerror (a black-holed endpoint) would otherwise hang the poll tick forever.
      await withTimeout(client.connect(), CODEX_CONNECT_TIMEOUT_MS, "codex websocket connect");
      return;
    } catch (error) {
      lastError = error;
      await Bun.sleep(CODEX_APP_SERVER_CONNECT_RETRY_MS);
    }
  }
  throw lastError ?? new Error("websocket connect failed");
}

async function freeLoopbackWsUrl(): Promise<string> {
  const port = await new Promise<number>((resolve, reject) => {
    const server = createServer();
    server.on("error", reject);
    server.listen(0, "127.0.0.1", () => {
      const address = server.address();
      const port = typeof address === "object" && address ? address.port : undefined;
      server.close(() => port ? resolve(port) : reject(new Error("failed to reserve loopback port")));
    });
  });
  return `ws://127.0.0.1:${port}`;
}

function providerHomeRoot(): string {
  return providerHomeRootFromEnv();
}

function providerHomeConfigDirs(provider: string, markerFile: string): string[] {
  const root = join(providerHomeRoot(), provider);
  const dirs: string[] = [];
  for (const profile of safeReadDir(root)) {
    const profileDir = join(root, profile);
    for (const instance of safeReadDir(profileDir)) {
      const dir = join(profileDir, instance);
      if (existsSync(join(dir, markerFile))) dirs.push(dir);
    }
  }
  return dirs;
}

function safeReadDir(path: string): string[] {
  try {
    return readdirSync(path, { withFileTypes: true })
      .filter((entry) => entry.isDirectory())
      .map((entry) => entry.name);
  } catch {
    return [];
  }
}

class JsonRpcWebSocket {
  private ws!: WebSocket;
  private nextId = 1;
  private pending = new Map<number, { resolve: (value: unknown) => void; reject: (error: unknown) => void }>();

  constructor(private readonly url: string) {}

  async connect(): Promise<void> {
    await new Promise<void>((resolve, reject) => {
      const ws = new WebSocket(this.url);
      this.ws = ws;
      ws.onopen = () => resolve();
      ws.onerror = () => reject(new Error("websocket error"));
      ws.onclose = (event) => {
        const error = new Error(`websocket closed code=${event.code} reason=${event.reason || "(none)"}`);
        for (const pending of this.pending.values()) pending.reject(error);
        this.pending.clear();
      };
      ws.onmessage = (event) => this.handleMessage(String(event.data));
    });
  }

  request(method: string, params?: unknown): Promise<unknown> {
    const id = this.nextId++;
    const promise = new Promise<unknown>((resolve, reject) => {
      this.pending.set(id, { resolve, reject });
    });
    this.ws.send(JSON.stringify({ id, method, params }));
    return promise;
  }

  notify(method: string, params?: unknown): void {
    this.ws.send(JSON.stringify(params === undefined ? { method } : { method, params }));
  }

  close(): void {
    this.ws?.close();
  }

  private handleMessage(raw: string): void {
    let message: unknown;
    try {
      message = JSON.parse(raw) as unknown;
    } catch {
      return;
    }
    if (!message || typeof message !== "object" || !("id" in message)) return;
    const id = Number((message as { id: unknown }).id);
    const pending = this.pending.get(id);
    if (!pending) return;
    this.pending.delete(id);
    const response = message as { result?: unknown; error?: { message?: string; code?: number } };
    if (response.error) pending.reject(new Error(`${response.error.message ?? "JSON-RPC error"} (${response.error.code ?? "unknown"})`));
    else pending.resolve(response.result);
  }
}

function candidateStateKey(candidate: Pick<QuotaCandidate, "provider" | "accountKey">): string {
  return `${candidate.provider}\0${candidate.accountKey}`;
}

function splitStateKey(key: string): [string, string] {
  const [provider, accountKey] = key.split("\0");
  return [provider ?? "", accountKey ?? ""];
}