import {
  acquireRateStateViaAppRuntime,
  penalizeRateStateViaAppRuntime,
  releaseRateStateViaAppRuntime,
  type WorkerRuntimeApiContext,
} from '../app-runtime-api';
import {
  noopPacingPermit,
  type PacingPermit,
  type PacingRule,
  type RateStateBackend,
  type RateStateBackendKind,
} from './rate-state-backend';
import {
  ProviderExhaustedError,
  providerNameFromBucketId,
} from '../run-failure';

/**
 * Longest provider pacer hold this substrate will sleep toward before failing
 * the call fast with a typed {@link ProviderExhaustedError}. Duration alone
 * decides — there is no quota-vs-rate classification. A hold at or under this is
 * slept off (today's behavior); a hold above it is skipped with no spend and no
 * provider dispatch.
 *
 * Enforced ONLY on the `app_runtime_postgres` (Absurd/Node DB-authoritative)
 * backend; the in-memory and coordinator backends never construct this class and
 * so keep their sleep-toward-it behavior byte-for-byte.
 */
export const PROVIDER_EXHAUSTED_MAX_WAIT_MS = 60_000;

/**
 * Permits leased per app-runtime round trip for the direct Node substrate.
 *
 * Eight keeps DB/API round trips amortized for batch-heavy maps while limiting
 * stale-block over-issue to 8 x live runner processes per bucket/window.
 */
/**
 * Transport seam for the app-runtime rate-state store. Mirrors the three module
 * calls the backend makes, minus the `context` argument (the default port binds
 * it). Aligns this backend with {@link CoordinatorRateStateBackend}'s injectable
 * port so both are driven — and tested — the same way.
 */
export interface AppRuntimeRatePort {
  acquire(
    input: Parameters<typeof acquireRateStateViaAppRuntime>[1],
  ): ReturnType<typeof acquireRateStateViaAppRuntime>;
  release(
    input: Parameters<typeof releaseRateStateViaAppRuntime>[1],
  ): ReturnType<typeof releaseRateStateViaAppRuntime>;
  penalize(
    input: Parameters<typeof penalizeRateStateViaAppRuntime>[1],
  ): ReturnType<typeof penalizeRateStateViaAppRuntime>;
}

function contextRatePort(context: WorkerRuntimeApiContext): AppRuntimeRatePort {
  return {
    acquire: (input) => acquireRateStateViaAppRuntime(context, input),
    release: (input) => releaseRateStateViaAppRuntime(context, input),
    penalize: (input) => penalizeRateStateViaAppRuntime(context, input),
  };
}

/**
 * Permits leased per app-runtime round trip. Eight keeps DB/API round trips
 * amortized for batch-heavy maps while bounding stale-block over-issue to 8 x
 * live runner processes per bucket/window. Deliberately conservative: absurd
 * runs many concurrent sandboxes against the same org+provider bucket, so a
 * larger block would let one runner reserve a whole window and starve peers.
 */
export const APP_RUNTIME_RATE_STATE_LEASE_BLOCK_SIZE = 8;
const LEASE_BLOCK_MIN_TTL_MS = 250;
const LEASE_BLOCK_MAX_TTL_MS = 5_000;
const MAX_ACQUIRE_SLEEP_MS = 5_000;
const STORE_FAILURE_RETRY_DELAYS_MS = [100, 250] as const;
/**
 * Jitter fraction applied to a server-supplied `waitMs` so a synchronized wave
 * of buckets that all hit granted=0 at the same instant re-probe at slightly
 * different times instead of thundering the store in lockstep. ±10%.
 */
const ACQUIRE_WAIT_JITTER_FRACTION = 0.1;

/** Last-known server cooldown surface for a bucket (consumed by PROVIDER_EXHAUSTED). */
export interface RateStateCooldown {
  /** Absolute epoch-ms until which the pacer's own (internally-capped) cooldown holds. */
  coolUntilMs: number;
  /** Most recent server Retry-After, stored verbatim as an absolute epoch-ms deadline. */
  claimedRetryAtMs: number;
}

interface LeasedBlock {
  permits: Array<[scheduledAtMs: number, leaseId: string | null]>;
  rateScopeToken: string | null;
  expiresAt: number;
  rulesKey: string;
  rules: PacingRule[];
}

interface PendingRelease {
  bucketId: string;
  rateScopeToken: string | null;
  rules: PacingRule[];
  leaseIds: string[];
  flushing: boolean;
}

interface Options {
  schedulerSchema?: string | null;
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
  onStoreFailure?: (info: { bucketId: string; error: string }) => void;
  /** Override the transport (tests inject a counting/faulting port). */
  port?: AppRuntimeRatePort;
  /**
   * Deterministic jitter source in [0,1) for the ±10% wait jitter. Defaults to
   * Math.random; tests pin it to assert the jittered sleep stays in bounds.
   */
  jitter?: () => number;
}

export class AppRuntimeRateStateBackend implements RateStateBackend {
  /**
   * DB-authoritative: the whole token bucket + AIMD runs in Postgres, so the
   * Governor defers pacing to this backend (see `resolveAdaptivePacing` /
   * `reportProviderBackpressure` in governor.ts).
   */
  readonly kind: RateStateBackendKind = 'app_runtime_postgres';
  private readonly port: AppRuntimeRatePort;
  private readonly schedulerSchema: string | null;
  private readonly now: () => number;
  private readonly sleep: (ms: number) => Promise<void>;
  private readonly jitter: () => number;
  private readonly onStoreFailure: (info: {
    bucketId: string;
    error: string;
  }) => void;
  private readonly blocks = new Map<string, LeasedBlock>();
  private readonly refills = new Map<string, Promise<void>>();
  private readonly pendingReleases = new Map<string, PendingRelease>();
  /**
   * Acquirers currently parked on each bucket's refill (empty block, awaiting a
   * grant). Demand-sizing reads this to request a block sized to the live wave
   * instead of a fixed floor, so a 200-row map fans into ~one round trip.
   */
  private readonly waiters = new Map<string, number>();
  /** Last-known server cooldown per bucket (surfaced for PROVIDER_EXHAUSTED). */
  private readonly cooldowns = new Map<string, RateStateCooldown>();
  private readonly effectiveRps = new Map<string, number>();
  private readonly pendingSuccesses = new Map<string, number>();
  private pendingReleaseFailure: Error | null = null;

  constructor(context: WorkerRuntimeApiContext, options: Options = {}) {
    this.port = options.port ?? contextRatePort(context);
    this.schedulerSchema = options.schedulerSchema?.trim() || null;
    this.now = options.now ?? (() => Date.now());
    this.sleep =
      options.sleep ??
      ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)));
    this.jitter = options.jitter ?? Math.random;
    this.onStoreFailure =
      options.onStoreFailure ??
      ((info) => {
        console.error('[app-runtime-rate-state] store failure', info);
      });
  }

  /**
   * Last-known server cooldown for a bucket, or null if the bucket has never
   * reported one. Minimal read surface for the PROVIDER_EXHAUSTED task: it lets
   * a caller learn how long the pacer is holding a provider off without issuing
   * its own acquire.
   */
  cooldownFor(bucketId: string): RateStateCooldown | null {
    return this.cooldowns.get(bucketId) ?? null;
  }

  /**
   * Pre-check the last-known server cooldown for a bucket. If the pacer is
   * already holding this provider past PROVIDER_EXHAUSTED_MAX_WAIT_MS, throw a
   * typed {@link ProviderExhaustedError} before any store round trip. Prefers
   * the server Retry-After (`claimedRetryAtMs`) as the retry deadline when it is
   * later than the internally-capped `coolUntilMs` and the cooldown is active;
   * otherwise reports `coolUntilMs`. No-op when the bucket has no live hold.
   */
  private throwIfPacerExhausted(bucketId: string): void {
    const cooldown = this.cooldowns.get(bucketId);
    if (!cooldown) return;
    const now = this.now();
    const coolActive = cooldown.coolUntilMs - now > 0;
    if (!coolActive) return;
    const coolWait = cooldown.coolUntilMs - now;
    const claimActive = cooldown.claimedRetryAtMs - now > 0;
    // Duration alone decides. Neither hold is over threshold → let it sleep.
    if (
      coolWait <= PROVIDER_EXHAUSTED_MAX_WAIT_MS &&
      !(
        claimActive &&
        cooldown.claimedRetryAtMs - now > PROVIDER_EXHAUSTED_MAX_WAIT_MS
      )
    ) {
      return;
    }
    // Retry deadline: the server Retry-After verbatim when present, else the
    // pacer's own cooldown deadline.
    const retryAtMs =
      cooldown.claimedRetryAtMs > 0
        ? cooldown.claimedRetryAtMs
        : cooldown.coolUntilMs;
    throw new ProviderExhaustedError({
      provider: providerNameFromBucketId(bucketId),
      retryAtMs,
    });
  }

  async acquire(input: {
    bucketId: string;
    rateScopeToken?: string | null;
    rules: readonly PacingRule[];
    signal?: AbortSignal;
  }): Promise<PacingPermit> {
    const { bucketId, rules, signal } = input;
    if (rules.length === 0) return noopPacingPermit();
    this.throwPendingReleaseFailure();
    // Pre-check: zero round trips when the pacer is already holding past the
    // threshold. Fails fast with a typed PROVIDER_EXHAUSTED before touching the
    // store.
    this.throwIfPacerExhausted(bucketId);

    const ordered = [...rules].sort((a, b) => a.ruleId.localeCompare(b.ruleId));
    const rulesKey = rulesSignature(ordered);
    const laneKey = rateLaneKey(bucketId, rulesKey);

    // Drawing a cached permit is a synchronous, single-threaded critical
    // section (`drawFromBlock` has no `await`), so concurrent acquires can never
    // hand out the same permit — no lock is needed for the fast path. The ONLY
    // thing that needs coordination is the block refill, which we single-flight
    // per bucket + rule set: one network round-trip (and any window-exhausted
    // sleep) serves the whole matching wave, then everyone re-draws. This replaces the old
    // per-bucket lock that held across the RPC + up-to-5s sleep and thereby
    // serialized every map row behind one sleeping acquirer.
    while (true) {
      if (signal?.aborted) {
        throw signal.reason instanceof Error
          ? signal.reason
          : new Error('Rate-state acquire aborted.');
      }
      const localPermit = this.drawFromBlock(bucketId, laneKey);
      if (localPermit) {
        const waitMs = localPermit.scheduledAtMs - this.now();
        if (waitMs > 0) await this.sleep(waitMs);
        if (signal?.aborted) {
          localPermit.permit.release();
          throw signal.reason instanceof Error
            ? signal.reason
            : new Error('Rate-state acquire aborted.');
        }
        return localPermit.permit;
      }
      // Count this acquirer as waiting on the refill so the single-flight refill
      // can size its block to the live wave. Decremented once a block lands (or
      // the refill errors) — see the finally in `refillBlock`.
      this.waiters.set(laneKey, (this.waiters.get(laneKey) ?? 0) + 1);
      try {
        await this.refillBlock(
          bucketId,
          input.rateScopeToken?.trim() || null,
          ordered,
          rulesKey,
          laneKey,
          signal,
        );
      } finally {
        const remaining = (this.waiters.get(laneKey) ?? 1) - 1;
        if (remaining > 0) this.waiters.set(laneKey, remaining);
        else this.waiters.delete(laneKey);
      }
    }
  }

  /**
   * Size the next lease from live demand and the provider's real capacity.
   *
   * Pure-rate buckets retain a small prefetch floor. Concurrency-bearing
   * buckets do not: each unused permit pins a durable provider slot, so request
   * only the live waiter count, capped by both the rate horizon and declared
   * max concurrency. This lets a 30-concurrent provider fill in one round trip
   * without reserving a single slot that cannot be used.
   */
  private requestedForBucket(
    laneKey: string,
    rules: readonly PacingRule[],
  ): number {
    const floor = APP_RUNTIME_RATE_STATE_LEASE_BLOCK_SIZE;
    const configuredRps = Math.min(
      ...rules.map((rule) => (rule.requestsPerWindow / rule.windowMs) * 1_000),
    );
    const horizonBudget = Math.max(
      1,
      Math.ceil(this.effectiveRps.get(laneKey) ?? configuredRps),
    );
    const rateBudget = Math.min(
      horizonBudget,
      ...rules
        .map((rule) => rule.requestsPerWindow)
        .filter((value) => Number.isFinite(value) && value > 0),
    );
    if (!Number.isFinite(rateBudget) || rateBudget <= 0) return floor;
    const waiters = this.waiters.get(laneKey) ?? 1;
    const concurrencyLimits = rules.flatMap((rule) =>
      rule.maxConcurrency != null &&
      Number.isFinite(rule.maxConcurrency) &&
      rule.maxConcurrency > 0
        ? [Math.trunc(rule.maxConcurrency)]
        : [],
    );
    if (concurrencyLimits.length > 0) {
      return Math.max(
        1,
        Math.min(waiters, Math.trunc(rateBudget), ...concurrencyLimits),
      );
    }
    return Math.max(floor, Math.min(waiters, Math.trunc(rateBudget)));
  }

  /**
   * Sleep duration for a granted=0 refill: the server's `waitMs`, spread by ±10%
   * jitter and capped by MAX_ACQUIRE_SLEEP_MS. Always at least 1ms.
   */
  private jitteredWaitMs(serverWaitMs: number): number {
    const base = Math.max(1, Math.min(serverWaitMs, MAX_ACQUIRE_SLEEP_MS));
    // jitter() in [0,1) → factor in [1-0.1, 1+0.1).
    const factor = 1 + (this.jitter() * 2 - 1) * ACQUIRE_WAIT_JITTER_FRACTION;
    return Math.max(
      1,
      Math.min(Math.round(base * factor), MAX_ACQUIRE_SLEEP_MS),
    );
  }

  /** Store the server's advisory cooldown for a bucket if the response carried one. */
  private recordCooldown(
    bucketId: string,
    response: { coolUntilMs?: number; claimedRetryAtMs?: number },
  ): void {
    const coolUntilMs = Number(response.coolUntilMs);
    const claimedRetryAtMs = Number(response.claimedRetryAtMs);
    if (!Number.isFinite(coolUntilMs) && !Number.isFinite(claimedRetryAtMs)) {
      return;
    }
    this.cooldowns.set(bucketId, {
      coolUntilMs: Number.isFinite(coolUntilMs) ? Math.max(0, coolUntilMs) : 0,
      claimedRetryAtMs: Number.isFinite(claimedRetryAtMs)
        ? Math.max(0, claimedRetryAtMs)
        : 0,
    });
  }

  /**
   * Build a PROVIDER_EXHAUSTED error for a granted=0 refill response. Retry
   * deadline: the server Retry-After (`claimedRetryAtMs`) verbatim when present,
   * else the pacer cooldown deadline (`coolUntilMs`), else `now + waitMs`.
   */
  private providerExhaustedFromResponse(
    bucketId: string,
    response: {
      waitMs: number;
      coolUntilMs?: number;
      claimedRetryAtMs?: number;
    },
  ): ProviderExhaustedError {
    const claimedRetryAtMs = Number(response.claimedRetryAtMs);
    const coolUntilMs = Number(response.coolUntilMs);
    const retryAtMs =
      Number.isFinite(claimedRetryAtMs) && claimedRetryAtMs > 0
        ? claimedRetryAtMs
        : Number.isFinite(coolUntilMs) && coolUntilMs > 0
          ? coolUntilMs
          : this.now() + Math.max(0, response.waitMs);
    return new ProviderExhaustedError({
      provider: providerNameFromBucketId(bucketId),
      retryAtMs,
    });
  }

  /**
   * Single-flight a block refill for a bucket. Concurrent acquires that find the
   * block empty all await the SAME in-flight refill instead of each issuing its
   * own serialized round-trip, so the RPC and any window-exhausted sleep happen
   * once per wave. Rejection (fail-closed store failure) propagates to every
   * waiter.
   */
  private refillBlock(
    bucketId: string,
    rateScopeToken: string | null,
    ordered: PacingRule[],
    rulesKey: string,
    laneKey: string,
    signal?: AbortSignal,
  ): Promise<void> {
    const existing = this.refills.get(laneKey);
    if (existing) return existing;
    const promise = this.performRefill(
      bucketId,
      rateScopeToken,
      ordered,
      rulesKey,
      laneKey,
      signal,
    ).finally(() => {
      if (this.refills.get(laneKey) === promise) {
        this.refills.delete(laneKey);
      }
    });
    this.refills.set(laneKey, promise);
    return promise;
  }

  private async performRefill(
    bucketId: string,
    rateScopeToken: string | null,
    ordered: PacingRule[],
    rulesKey: string,
    laneKey: string,
    signal?: AbortSignal,
  ): Promise<void> {
    let consecutiveFailures = 0;
    // Yield once before sizing the first request so the whole synchronously-
    // scheduled acquire wave finishes parking (incrementing `waiters`) before
    // `requestedForBucket` reads the count. Without this, the single-flight
    // refill fires while only the first acquirer has registered, and demand
    // sizing collapses to the floor. Single-flight still holds — this is the
    // one and only refill for the bucket; peers just enqueue behind it.
    let firstPass = true;
    while (true) {
      // Abort is enforced per-acquirer in the acquire loop; a shared refill just
      // stops early so waiters re-check their own signal.
      if (signal?.aborted) return;
      if (firstPass) {
        firstPass = false;
        await Promise.resolve();
        if (signal?.aborted) return;
      }
      try {
        const successKey = pendingSuccessKey(bucketId, ordered);
        const observedSuccesses = this.pendingSuccesses.get(successKey) ?? 0;
        const response = await this.port.acquire({
          bucketId,
          rateScopeToken,
          // Stamp adaptiveMaxRps on default-pacing rules so the DB row's ceiling
          // matches the old in-memory AIMD range.
          rules: withAdaptiveMaxRps(ordered),
          requested: this.requestedForBucket(laneKey, ordered),
          observedSuccesses,
          schedulerSchema: this.schedulerSchema,
        });
        if (observedSuccesses > 0) {
          const remaining = Math.max(
            0,
            (this.pendingSuccesses.get(successKey) ?? 0) - observedSuccesses,
          );
          if (remaining > 0) this.pendingSuccesses.set(successKey, remaining);
          else this.pendingSuccesses.delete(successKey);
        }
        this.recordCooldown(bucketId, response);
        if (
          Number.isFinite(response.effectiveRequestsPerSecond) &&
          Number(response.effectiveRequestsPerSecond) > 0
        ) {
          this.effectiveRps.set(
            laneKey,
            Number(response.effectiveRequestsPerSecond),
          );
        }
        if (response.granted > 0) {
          // Merge the whole grant into the block and let acquirers re-draw. No
          // permit is handed out here, so nothing is double-drawn.
          this.mergeGrantedBlock(
            bucketId,
            rateScopeToken,
            ordered,
            rulesKey,
            laneKey,
            response,
          );
          return;
        }
        // Live path: the server told us to wait past the threshold. Fail fast
        // with a typed PROVIDER_EXHAUSTED instead of sleeping toward it. The
        // rejection propagates to every waiter parked on this single-flight
        // refill (they all skip with no spend). Duration alone decides.
        if (response.waitMs > PROVIDER_EXHAUSTED_MAX_WAIT_MS) {
          throw this.providerExhaustedFromResponse(bucketId, response);
        }
        // Window exhausted: sleep the SERVER's waitMs once for the whole waiting
        // wave (not per row, and not a guess), then return so acquirers re-draw
        // and single-flight another refill if still empty. ±10% jitter so a
        // synchronized wave re-probes at spread-out times; capped by
        // MAX_ACQUIRE_SLEEP_MS.
        await this.sleep(this.jitteredWaitMs(response.waitMs));
        return;
      } catch (error) {
        // PROVIDER_EXHAUSTED is a deliberate skip, not a store failure: it must
        // escape this fail-closed retry ladder verbatim so it reaches the
        // acquire caller (and, via step-miss handling, the row).
        if (error instanceof ProviderExhaustedError) throw error;
        // Fail closed. A degraded rate-state store must not grant unlimited
        // provider calls; after this bounded retry ladder the run fails loudly.
        consecutiveFailures += 1;
        const normalized = normalizeError(error);
        this.onStoreFailure({ bucketId, error: normalized.message });
        if (consecutiveFailures > STORE_FAILURE_RETRY_DELAYS_MS.length) {
          throw new Error(
            `Rate-state store unavailable for ${bucketId}; failing closed after ${consecutiveFailures} acquire attempts: ${normalized.message}`,
          );
        }
        await this.sleep(
          STORE_FAILURE_RETRY_DELAYS_MS[consecutiveFailures - 1] ?? 250,
        );
      }
    }
  }

  /**
   * Validate a granted acquire response and merge its permits into the bucket's
   * local block. Concurrency rules carry one lease token per permit (release
   * semantics); pure-rate rules carry untracked window permits the store already
   * decremented.
   */
  private mergeGrantedBlock(
    bucketId: string,
    rateScopeToken: string | null,
    ordered: PacingRule[],
    rulesKey: string,
    laneKey: string,
    response: {
      granted: number;
      leaseIds?: unknown;
      serverNowMs?: unknown;
      scheduledAtMs?: unknown;
    },
  ): void {
    const hasConcurrency = ordered.some((rule) => rule.maxConcurrency != null);
    const leaseIds = Array.isArray(response.leaseIds)
      ? response.leaseIds.filter(
          (leaseId): leaseId is string =>
            typeof leaseId === 'string' && leaseId.trim().length > 0,
        )
      : [];
    if (hasConcurrency && leaseIds.length !== response.granted) {
      throw new Error(
        `Rate-state store granted ${response.granted} concurrency permits for ${bucketId} with ${leaseIds.length} lease tokens.`,
      );
    }
    if (!hasConcurrency && leaseIds.length > 0) {
      throw new Error(
        `Rate-state store returned ${leaseIds.length} lease tokens for pure-rate bucket ${bucketId}.`,
      );
    }
    // Rolling deploy compatibility: an older gateway returns no schedule. It
    // has already debited the block under the legacy token-bucket contract, so
    // keep those permits immediately usable until every gateway serves the new
    // globally reserved timestamps.
    const gatewayScheduledAtMs = Array.isArray(response.scheduledAtMs)
      ? response.scheduledAtMs.map(Number)
      : Array.from({ length: response.granted }, () => this.now());
    const serverNowMs = Number(response.serverNowMs);
    const receivedAtMs = this.now();
    const scheduledAtMs =
      Number.isFinite(serverNowMs) && serverNowMs >= 0
        ? gatewayScheduledAtMs.map(
            (scheduledAtMs) =>
              receivedAtMs + Math.max(0, scheduledAtMs - serverNowMs),
          )
        : gatewayScheduledAtMs;
    if (
      scheduledAtMs.length !== response.granted ||
      scheduledAtMs.some((value) => !Number.isFinite(value) || value < 0)
    ) {
      throw new Error(
        `Rate-state store granted ${response.granted} permits for ${bucketId} with ${scheduledAtMs.length} dispatch timestamps.`,
      );
    }
    this.mergeBlock(
      bucketId,
      rateScopeToken,
      scheduledAtMs.map((scheduledAtMs, index): [number, string | null] => [
        scheduledAtMs,
        hasConcurrency ? (leaseIds[index] ?? null) : null,
      ]),
      rulesKey,
      laneKey,
      ordered,
    );
  }

  async penalize(input: {
    bucketId: string;
    rateScopeToken?: string | null;
    cooldownMs: number;
  }): Promise<void> {
    if (input.cooldownMs <= 0) return;
    this.drainBlocksForBucket(input.bucketId);
    try {
      await this.port.penalize({
        bucketId: input.bucketId,
        rateScopeToken: input.rateScopeToken?.trim() || null,
        cooldownMs: input.cooldownMs,
        schedulerSchema: this.schedulerSchema,
      });
    } catch (error) {
      const normalized = normalizeError(error);
      this.onStoreFailure({
        bucketId: input.bucketId,
        error: normalized.message,
      });
      throw new Error(
        `Rate-state penalize failed for ${input.bucketId}: ${normalized.message}`,
      );
    }
  }

  observeSuccess(input: {
    bucketId: string;
    rules: readonly PacingRule[];
  }): void {
    const key = pendingSuccessKey(input.bucketId, input.rules);
    this.pendingSuccesses.set(
      key,
      Math.min(1_000_000, (this.pendingSuccesses.get(key) ?? 0) + 1),
    );
  }

  private permitFor(
    bucketId: string,
    rateScopeToken: string | null,
    rules: PacingRule[],
    leaseId: string | null,
  ): PacingPermit {
    if (!rules.some((rule) => rule.maxConcurrency != null)) {
      return noopPacingPermit();
    }
    let released = false;
    return {
      release: () => {
        if (released) return;
        released = true;
        this.releaseReserved(
          bucketId,
          rateScopeToken,
          rules,
          leaseId ? [leaseId] : [],
        );
      },
    };
  }

  private mergeBlock(
    bucketId: string,
    rateScopeToken: string | null,
    permits: LeasedBlock['permits'],
    rulesKey: string,
    laneKey: string,
    rules: PacingRule[],
  ): void {
    const lastScheduledAtMs = permits.at(-1)?.[0] ?? this.now();
    const freshExpiresAt =
      Math.max(this.now(), lastScheduledAtMs) + leasedBlockTtlMs(rules);
    const existing = this.blocks.get(laneKey);
    if (
      existing &&
      existing.rulesKey === rulesKey &&
      existing.expiresAt > this.now()
    ) {
      existing.permits.push(...permits);
      existing.permits.sort((a, b) => a[0] - b[0]);
      existing.expiresAt = Math.max(existing.expiresAt, freshExpiresAt);
      return;
    }
    if (existing)
      this.releaseReserved(
        bucketId,
        existing.rateScopeToken,
        existing.rules,
        existing.permits.flatMap(([, leaseId]) => (leaseId ? [leaseId] : [])),
      );
    this.blocks.set(laneKey, {
      permits,
      rateScopeToken,
      expiresAt: freshExpiresAt,
      rulesKey,
      rules,
    });
  }

  private drawFromBlock(
    bucketId: string,
    laneKey: string,
  ): { permit: PacingPermit; scheduledAtMs: number } | null {
    const block = this.blocks.get(laneKey);
    if (!block) return null;
    if (block.expiresAt <= this.now()) {
      this.blocks.delete(laneKey);
      // Unused window permits are forfeited on TTL expiry (the server already
      // decremented the provider window for them, and the rule-scoped TTL is
      // capped so at most one block's worth is stranded per bucket/window).
      // Concurrency leases must be released so the shared slot count drains.
      this.releaseReserved(
        bucketId,
        block.rateScopeToken,
        block.rules,
        block.permits.flatMap(([, leaseId]) => (leaseId ? [leaseId] : [])),
      );
      return null;
    }
    const scheduled = block.permits.shift();
    if (!scheduled) {
      this.blocks.delete(laneKey);
      return null;
    }
    if (block.permits.length === 0) this.blocks.delete(laneKey);
    return {
      permit: this.permitFor(
        bucketId,
        block.rateScopeToken,
        block.rules,
        scheduled[1],
      ),
      scheduledAtMs: scheduled[0],
    };
  }

  private drainBlocksForBucket(bucketId: string): void {
    const prefix = `${bucketId}\0`;
    for (const [laneKey, block] of this.blocks) {
      if (!laneKey.startsWith(prefix)) continue;
      this.blocks.delete(laneKey);
      this.releaseReserved(
        bucketId,
        block.rateScopeToken,
        block.rules,
        block.permits.flatMap(([, leaseId]) => (leaseId ? [leaseId] : [])),
      );
    }
  }

  private releaseReserved(
    bucketId: string,
    rateScopeToken: string | null,
    rules: readonly PacingRule[],
    leaseIds: readonly string[],
  ): void {
    if (
      leaseIds.length === 0 ||
      !rules.some((rule) => rule.maxConcurrency != null)
    )
      return;
    const ordered = [...rules].sort((a, b) => a.ruleId.localeCompare(b.ruleId));
    const key = `${bucketId}\0${rulesSignature(ordered)}`;
    const pending = this.pendingReleases.get(key) ?? {
      bucketId,
      rateScopeToken,
      rules: ordered,
      leaseIds: [],
      flushing: false,
    };
    pending.leaseIds.push(...leaseIds);
    this.pendingReleases.set(key, pending);
    if (pending.flushing) return;
    pending.flushing = true;
    queueMicrotask(() => void this.flushReleases(key, pending));
  }

  private async flushReleases(
    key: string,
    pending: PendingRelease,
  ): Promise<void> {
    while (pending.leaseIds.length > 0) {
      const leaseIds = pending.leaseIds.splice(0);
      try {
        await this.port.release({
          bucketId: pending.bucketId,
          rateScopeToken: pending.rateScopeToken,
          rules: pending.rules,
          leaseIds,
          schedulerSchema: this.schedulerSchema,
        });
      } catch (error) {
        const normalized = normalizeError(error);
        this.pendingReleaseFailure = new Error(
          `Rate-state release failed for ${pending.bucketId}: ${normalized.message}`,
        );
        this.onStoreFailure({
          bucketId: pending.bucketId,
          error: normalized.message,
        });
      }
    }
    pending.flushing = false;
    if (pending.leaseIds.length > 0) {
      pending.flushing = true;
      queueMicrotask(() => void this.flushReleases(key, pending));
      return;
    }
    if (this.pendingReleases.get(key) === pending) {
      this.pendingReleases.delete(key);
    }
  }

  private throwPendingReleaseFailure(): void {
    if (!this.pendingReleaseFailure) return;
    const error = this.pendingReleaseFailure;
    this.pendingReleaseFailure = null;
    throw error;
  }
}

/**
 * Multiplier applied to an undeclared default-pacing rule's
 * `requestsPerWindow` to derive its `adaptiveMaxRps` ceiling. Declared rules
 * carry either their explicit advisory ceiling or their hard base rate.
 */
const DEFAULT_PACING_ADAPTIVE_MAX_RPS_MULTIPLIER = 25;

/**
 * A rule id shaped `default:<toolId>` is the undeclared-tool default-pacing rule
 * minted by `defaultPacingForTool` in governor.ts.
 */
function isDefaultPacingRuleId(ruleId: string): boolean {
  return ruleId.startsWith('default:');
}

/**
 * Preserve explicit advisory ceilings, stamp the legacy base * 25 ceiling onto
 * undeclared default-pacing rules, and stamp declared rules with their base
 * rate. Sending the hard base explicitly also repairs bucket rows widened by a
 * previous runtime version on their next acquire.
 */
function withAdaptiveMaxRps(rules: readonly PacingRule[]): PacingRule[] {
  return rules.map((rule) =>
    rule.adaptiveMaxRps != null
      ? rule
      : isDefaultPacingRuleId(rule.ruleId)
        ? ({
            ...rule,
            adaptiveMaxRps:
              rule.requestsPerWindow *
              DEFAULT_PACING_ADAPTIVE_MAX_RPS_MULTIPLIER,
          } as PacingRule)
        : ({
            ...rule,
            adaptiveMaxRps: (rule.requestsPerWindow / rule.windowMs) * 1_000,
          } as PacingRule),
  );
}

function rulesSignature(rules: readonly PacingRule[]): string {
  return [...rules]
    .map(
      (rule) =>
        `${rule.ruleId}:${rule.requestsPerWindow}:${rule.windowMs}:${rule.adaptiveMaxRps ?? ''}:${rule.maxConcurrency ?? ''}`,
    )
    .sort()
    .join('|');
}

function rateLaneKey(bucketId: string, rulesKey: string): string {
  return `${bucketId}\0${rulesKey}`;
}

function pendingSuccessKey(
  bucketId: string,
  rules: readonly PacingRule[],
): string {
  return rateLaneKey(bucketId, rulesSignature(rules));
}

function leasedBlockTtlMs(rules: readonly PacingRule[]): number {
  const shortestWindowMs = Math.min(
    ...rules
      .map((rule) => rule.windowMs)
      .filter((windowMs) => Number.isFinite(windowMs) && windowMs > 0),
  );
  if (!Number.isFinite(shortestWindowMs)) return LEASE_BLOCK_MIN_TTL_MS;
  return Math.max(
    LEASE_BLOCK_MIN_TTL_MS,
    Math.min(LEASE_BLOCK_MAX_TTL_MS, Math.floor(shortestWindowMs)),
  );
}

function normalizeError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error));
}