qmd/src/embedding/openai.ts

/**
 * openai.ts - OpenAI-compatible HTTP embedding provider
 *
 * Talks to any endpoint that implements `POST /v1/embeddings` with the OpenAI
 * shape: request `{model, input: string|string[]}`, response
 * `{data: [{embedding: number[], index: number}, ...]}`.
 *
 * Used by qmd to delegate embeddings to an approved commercial HTTPS API.
 *
 * Features:
 *   - Batches input in groups of ≤64 (configurable via QMD_EMBED_BATCH_SIZE)
 *   - Retries 429 / 503 with exponential backoff (1s, 4s, 16s)
 *   - 429 gets its own, larger retry budget and honours `Retry-After`
 *     (header or `"Retry after 29s"` body text) — a shared gateway bucket
 *     refilling in 30s must not exhaust a 1s/4s/16s schedule (i-yghj098h)
 *   - Bulk lane (`embedBatch` with >1 input): a 429 pauses the WHOLE worker
 *     pool for the advertised cooldown and halves in-flight concurrency
 *     (AIMD), so reindex traffic yields the bucket to interactive callers
 *     instead of contending head-on with them
 *   - 4xx (non-429) → no retry, count as failure
 *   - Circuit breaker: >50% failures in a 60s window → OPEN for 5 min;
 *     callers receive failures and no model fallback is selected
 *   - Per-call timeout via AbortSignal (default QMD_EMBED_TIMEOUT_MS=30000)
 *   - Healthcheck via `GET /health` if available, else a probe embed call
 */

import os from "node:os";

import type {
  EmbeddingProvider,
  ProviderEmbedOptions,
  ProviderEmbedding,
  ProviderHealth,
  ProviderKind,
} from "./provider.js";

// ─────────────────────────── Configuration ───────────────────────────────────

/**
 * Default batch size — most OpenAI-compatible embedding endpoints accept up to
 * 2048 inputs per call but for memory and latency we cap at 64.
 */
export const DEFAULT_BATCH_SIZE = 64;

/**
 * Default in-flight concurrency cap for `embedBatch`. The qmd-embed-worker
 * exposes a 4-way semaphore (`MAX_CONCURRENT_REQUESTS=4`) and idles at
 * queue-depth 1.0 under sequential clients (i-fkpnar9i baseline). Defaulting
 * to 4 matches the worker's advertised concurrency without overshooting the
 * GPU. Override per-deploy via `QMD_EMBED_CONCURRENCY`. Setting to 1 reverts
 * to the legacy sequential dispatch.
 */
export const DEFAULT_CONCURRENCY = 4;

/**
 * Default per-request timeout (30 s). embeddinggemma-300M on RTX 4090 takes
 * <500ms per batch of 64 in practice; 30s is a safe upper bound.
 */
export const DEFAULT_TIMEOUT_MS = 30_000;

/**
 * Retry backoff schedule (ms) for 429/503 responses. 3 attempts total
 * (initial + 2 retries) — aligns with issue spec "1s/4s/16s".
 */
export const RETRY_BACKOFFS_MS: readonly number[] = [1_000, 4_000, 16_000];

/**
 * Rate-limit (429) retry budget, separate from `RETRY_BACKOFFS_MS`.
 *
 * A 429 is backpressure, not a fault: the shared ai.mm.mk gateway answers
 * `{"detail":"Rate limit exceeded (tokens). Retry after 29s."}` with observed
 * retry-after values of 7s/29s/30s, while the generic schedule waits at most
 * 1+4+16 = 21s in total. Bulk reindex runs therefore burned all three attempts
 * inside one bucket refill and reported the chunk as failed (i-yghj098h).
 * 429s get their own attempt count and always wait at least as long as the
 * server asked for, capped by `RATE_LIMIT_MAX_BACKOFF_MS`.
 */
export const DEFAULT_RATE_LIMIT_RETRIES = 5;

/** Upper bound on a single rate-limit wait, so a bogus Retry-After can't hang a run. */
export const RATE_LIMIT_MAX_BACKOFF_MS = 60_000;

/** Fallback wait when a 429 carries no parseable Retry-After (doubles per attempt). */
export const RATE_LIMIT_BASE_BACKOFF_MS = 5_000;

/**
 * Consecutive successful bulk requests before the lane additively recovers one
 * unit of concurrency after a 429-triggered halving (AIMD).
 */
export const LANE_RECOVERY_STREAK = 8;

/**
 * Circuit breaker — flips OPEN when error rate exceeds threshold within
 * window. While OPEN, every call fails fast so the caller can fall back.
 */
export const CIRCUIT_WINDOW_MS = 60_000;
export const CIRCUIT_OPEN_DURATION_MS = 5 * 60_000;
export const CIRCUIT_FAILURE_RATE_THRESHOLD = 0.5;
export const CIRCUIT_MIN_SAMPLES = 4;

// ─────────────────────────── Types ───────────────────────────────────────────

export type OpenAIProviderConfig = {
  /** Endpoint base URL — e.g. "https://ai.mm.mk" (no trailing slash) */
  endpoint: string;
  /** Optional bearer token sent as `Authorization: Bearer ...` */
  apiKey?: string;
  /**
   * Stable model identifier to report up via `getModelId()`.
   * Defaults to "embeddinggemma" to match qmd's existing DB rows.
   */
  modelId?: string;
  /**
   * Upstream model name sent in the HTTP request body. Often differs from
   * `modelId` (e.g. modelId="embeddinggemma" but upstream model="embeddinggemma:300m").
   */
  upstreamModel?: string;
  /** Batch size cap (default DEFAULT_BATCH_SIZE = 64) */
  batchSize?: number;
  /**
   * Max in-flight HTTP requests during a single `embedBatch` call. Default
   * `DEFAULT_CONCURRENCY=4` matches the worker semaphore. Set to 1 to force
   * legacy sequential dispatch (useful for benchmarks / regression bisect).
   */
  concurrency?: number;
  /** Per-request timeout in ms (default DEFAULT_TIMEOUT_MS = 30_000) */
  timeoutMs?: number;
  /** Custom fetch (for testing). Defaults to global `fetch`. */
  fetchImpl?: typeof fetch;
  /** Custom retry schedule (for testing). Defaults to RETRY_BACKOFFS_MS. */
  retryBackoffsMs?: readonly number[];
  /**
   * Extra retry attempts granted to 429 responses on top of the generic
   * schedule (default DEFAULT_RATE_LIMIT_RETRIES = 5, env
   * `QMD_EMBED_RATE_LIMIT_RETRIES`). Rate limits are backpressure, not faults.
   */
  rateLimitRetries?: number;
  /** Custom sleep impl (for testing). Defaults to setTimeout. */
  sleep?: (ms: number) => Promise<void>;
  /** Custom clock (for testing). Defaults to Date.now. */
  now?: () => number;
};

export type OpenAIEmbeddingsResponse = {
  object?: string;
  model?: string;
  data: Array<{
    object?: string;
    index: number;
    embedding: number[];
  }>;
  usage?: {
    prompt_tokens?: number;
    total_tokens?: number;
  };
};

/**
 * Circuit breaker state — exported for tests
 */
export type CircuitState = "closed" | "open" | "half-open";

/**
 * Traffic lane for a single `embedBatch` call. `bulk` = reindex traffic
 * (multi-input batch), which self-throttles via `BulkLaneGate` and is
 * attributed separately in `X-AI-Caller`. `interactive` = query-time embeds,
 * which are never made to wait behind a bulk cooldown.
 */
export type EmbedLane = "interactive" | "bulk";

// ─────────────────────────── Helpers ─────────────────────────────────────────

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

/** Parse a non-negative integer env value; `undefined` when unset/invalid. */
function parseNonNegativeInt(raw: string | undefined): number | undefined {
  if (raw == null || raw.trim() === "") return undefined;
  const n = Number.parseInt(raw, 10);
  return Number.isFinite(n) && n >= 0 ? n : undefined;
}

/**
 * Build the advisory `X-AI-Caller` attribution header value (Oivo ai.mm.mk
 * "Observability-Driven Fleet Self-Improvement" rollout, Thread 1). Format
 * mirrors Oivo's `cli/src/shared/aiCallerHeader.ts` EXACTLY:
 *   site=<file:line | tool file>; mc=<machine>; sid=<session-8char>;
 *   comp=<component>; via=<logical-label>
 *
 * Advisory metadata ONLY — never an auth/authorization input. `mc`/`sid` come
 * from the Oivo fleet env when qmd runs as a delegated embedder; both degrade
 * to `-`/short-hostname safely when qmd runs standalone. Values are sanitized
 * (`;` + control chars stripped, capped at 120) so a pathological env value can
 * never break the header grammar.
 */
function aiCallerHeaderValue(via: string): string {
  const clean = (raw: string | undefined, fallback: string): string => {
    if (raw == null) return fallback;
    let out = "";
    for (const ch of String(raw)) {
      const code = ch.charCodeAt(0);
      out += code < 32 || code === 127 || ch === ";" ? " " : ch;
    }
    const collapsed = out.replace(/\s+/g, " ").trim();
    return (collapsed || fallback).slice(0, 120);
  };
  let mc = (process.env.OIVO_MACHINE_NAME ?? "").trim();
  if (!mc) {
    try {
      mc = os.hostname().split(".")[0] || "unknown";
    } catch {
      mc = "unknown";
    }
  }
  const sid = (process.env.OIVO_SESSION_ID || process.env.CLAUDE_CODE_SESSION_ID || "-").slice(0, 8);
  return `site=src/embedding/openai.ts; mc=${clean(mc, "unknown")}; sid=${clean(sid, "-")}; comp=qmd; via=${clean(via, "-")}`;
}

/**
 * Build the merged AbortSignal for a single HTTP attempt: combines an
 * external `userSignal` (from caller / withLLMSession) with a per-attempt
 * timeout signal. Returns the merged signal AND the timeout id so the
 * caller can `clearTimeout` after the attempt completes (avoids leaks).
 */
function buildAttemptSignal(
  userSignal: AbortSignal | undefined,
  timeoutMs: number,
): { signal: AbortSignal; cleanup: () => void } {
  const ctrl = new AbortController();
  const timeoutId = setTimeout(() => {
    ctrl.abort(new Error(`Request timed out after ${timeoutMs}ms`));
  }, timeoutMs);
  // Don't keep process alive just for this timer
  if (typeof timeoutId === "object" && timeoutId !== null && "unref" in timeoutId) {
    (timeoutId as { unref: () => void }).unref();
  }

  const onUserAbort = () => ctrl.abort(userSignal?.reason);
  if (userSignal) {
    if (userSignal.aborted) {
      ctrl.abort(userSignal.reason);
    } else {
      userSignal.addEventListener("abort", onUserAbort, { once: true });
    }
  }

  const cleanup = () => {
    clearTimeout(timeoutId);
    if (userSignal) userSignal.removeEventListener("abort", onUserAbort);
  };

  return { signal: ctrl.signal, cleanup };
}

/**
 * Determine whether an HTTP status is retryable. 429 (Too Many Requests)
 * and 503 (Service Unavailable) are retried; 4xx (other than 429) are not.
 */
export function isRetryableStatus(status: number): boolean {
  return status === 429 || status === 503;
}

/**
 * Extract the server-advertised cooldown from a rate-limited response.
 *
 * Two sources, in priority order:
 *   1. the standard `Retry-After` header — delta-seconds or an HTTP-date;
 *   2. the ai.mm.mk body text, which states the cooldown in prose only:
 *      `{"detail":"Rate limit exceeded (tokens). Retry after 29s.", ...}`.
 *
 * Returns `undefined` when neither source yields a sane positive duration, so
 * the caller falls back to its own schedule. Values are clamped to
 * `RATE_LIMIT_MAX_BACKOFF_MS`.
 */
export function parseRetryAfterMs(
  headerValue: string | null | undefined,
  bodyPreview?: string,
  now: () => number = Date.now,
): number | undefined {
  const clamp = (ms: number): number | undefined =>
    Number.isFinite(ms) && ms > 0 ? Math.min(ms, RATE_LIMIT_MAX_BACKOFF_MS) : undefined;

  const header = headerValue?.trim();
  if (header) {
    if (/^\d+(\.\d+)?$/.test(header)) {
      const fromSeconds = clamp(Number(header) * 1000);
      if (fromSeconds !== undefined) return fromSeconds;
    }
    const asDate = Date.parse(header);
    if (!Number.isNaN(asDate)) {
      const fromDate = clamp(asDate - now());
      if (fromDate !== undefined) return fromDate;
    }
  }

  if (bodyPreview) {
    const m = /retry\s+after\s+(\d+(?:\.\d+)?)\s*(ms|s|seconds?)?/i.exec(bodyPreview);
    if (m) {
      const value = Number(m[1]);
      const unit = (m[2] ?? "s").toLowerCase();
      return clamp(unit === "ms" ? value : value * 1000);
    }
  }

  return undefined;
}

/**
 * Chunk an array into pieces of ≤ size each. `size` MUST be ≥ 1.
 */
export function chunkArray<T>(items: T[], size: number): T[][] {
  if (size < 1) throw new Error(`chunkArray: size must be ≥ 1, got ${size}`);
  if (items.length <= size) return items.length === 0 ? [] : [items];
  const out: T[][] = [];
  for (let i = 0; i < items.length; i += size) {
    out.push(items.slice(i, i + size));
  }
  return out;
}

// ─────────────────────────── Circuit Breaker ─────────────────────────────────

/**
 * Sliding-window circuit breaker. Tracks the last N samples (min 4) over a
 * 60-second window; flips OPEN when failure rate exceeds 50%, then auto-
 * resets to HALF-OPEN after 5 minutes — at which point the next probe
 * decides whether to close (success) or re-open (failure).
 */
export class CircuitBreaker {
  private samples: { ts: number; ok: boolean }[] = [];
  private state: CircuitState = "closed";
  private openedAt: number | null = null;
  private readonly windowMs: number;
  private readonly openDurationMs: number;
  private readonly threshold: number;
  private readonly minSamples: number;
  private readonly now: () => number;

  constructor(opts: {
    windowMs?: number;
    openDurationMs?: number;
    threshold?: number;
    minSamples?: number;
    now?: () => number;
  } = {}) {
    this.windowMs = opts.windowMs ?? CIRCUIT_WINDOW_MS;
    this.openDurationMs = opts.openDurationMs ?? CIRCUIT_OPEN_DURATION_MS;
    this.threshold = opts.threshold ?? CIRCUIT_FAILURE_RATE_THRESHOLD;
    this.minSamples = opts.minSamples ?? CIRCUIT_MIN_SAMPLES;
    this.now = opts.now ?? Date.now;
  }

  getState(): CircuitState {
    this.tickAutoReset();
    return this.state;
  }

  /**
   * Returns true when calls should be short-circuited (skip HTTP, fall back).
   * Side-effects: may transition OPEN → HALF-OPEN if the open window expired.
   */
  shouldFailFast(): boolean {
    return this.getState() === "open";
  }

  /** Record a successful call. */
  recordSuccess(): void {
    // Honor the time-based OPEN→HALF-OPEN transition before deciding what
    // to do with this sample. Without this, a success that lands AFTER the
    // open window expired would still see state==="open" and never close
    // the breaker (a probe call could only flip it via getState()).
    this.tickAutoReset();
    this.pushSample(true);
    if (this.state === "half-open") {
      this.state = "closed";
      this.openedAt = null;
    }
  }

  /** Record a failed call. May trigger OPEN. */
  recordFailure(): void {
    // Same reasoning as recordSuccess — apply lazy auto-reset before
    // classifying the sample.
    this.tickAutoReset();
    this.pushSample(false);
    if (this.state === "half-open") {
      // Probe failed — re-open
      this.state = "open";
      this.openedAt = this.now();
      return;
    }
    if (this.state === "closed") this.evaluate();
  }

  /** Force-reset the breaker (used by tests / admin) */
  reset(): void {
    this.samples = [];
    this.state = "closed";
    this.openedAt = null;
  }

  private pushSample(ok: boolean): void {
    const ts = this.now();
    this.samples.push({ ts, ok });
    // Drop samples outside the window
    const cutoff = ts - this.windowMs;
    while (this.samples.length > 0 && this.samples[0]!.ts < cutoff) {
      this.samples.shift();
    }
  }

  private evaluate(): void {
    if (this.samples.length < this.minSamples) return;
    const failures = this.samples.filter((s) => !s.ok).length;
    const rate = failures / this.samples.length;
    if (rate > this.threshold) {
      this.state = "open";
      this.openedAt = this.now();
    }
  }

  private tickAutoReset(): void {
    if (this.state === "open" && this.openedAt !== null) {
      if (this.now() - this.openedAt >= this.openDurationMs) {
        this.state = "half-open";
      }
    }
  }
}

// ─────────────────────────── Errors ──────────────────────────────────────────

/**
 * Raised when the circuit breaker is OPEN and a call is short-circuited.
 */
export class CircuitOpenError extends Error {
  constructor(message = "OpenAIEmbeddingsProvider circuit is OPEN") {
    super(message);
    this.name = "CircuitOpenError";
  }
}

/**
 * Persistent (non-retryable) HTTP error from upstream. Includes status code.
 */
export class HttpError extends Error {
  readonly status: number;
  readonly bodyPreview: string;
  /** Server-advertised cooldown for 429s, when the response stated one. */
  readonly retryAfterMs?: number;
  constructor(status: number, bodyPreview: string, retryAfterMs?: number) {
    super(`HTTP ${status}: ${bodyPreview.slice(0, 200)}`);
    this.name = "HttpError";
    this.status = status;
    this.bodyPreview = bodyPreview.slice(0, 1024);
    this.retryAfterMs = retryAfterMs;
  }
}

// ─────────────────────────── Bulk lane ───────────────────────────────────────

/**
 * Client-side bulk lane for reindex traffic (i-yghj098h).
 *
 * qmd's bulk embedding shares the interactive ai.mm.mk token bucket. Without a
 * server-side per-caller budget, the only way bulk traffic can stop contending
 * head-on with interactive callers is to police itself:
 *
 *   - a 429 on ANY worker pauses the ENTIRE pool for the advertised cooldown
 *     (one shared promise — concurrent 429s coalesce instead of stacking N
 *     cooldowns), so the bucket refills for interactive callers rather than
 *     being re-drained by the remaining workers;
 *   - the in-flight cap halves on each cooldown (floor 1) and recovers one unit
 *     per `LANE_RECOVERY_STREAK` successes — classic AIMD, so a run settles at
 *     whatever share the bucket actually has spare.
 *
 * Deliberately clock-free: cooldowns are modelled as a promise produced by the
 * injected `sleep`, so tests drive them with a fake sleep and no fake clock.
 */
export class BulkLaneGate {
  private readonly maxPermits: number;
  private readonly sleep: (ms: number) => Promise<void>;
  private permits: number;
  private inFlight = 0;
  private okStreak = 0;
  private cooldown: Promise<void> | null = null;
  private waiters: (() => void)[] = [];

  constructor(maxPermits: number, sleep: (ms: number) => Promise<void>) {
    this.maxPermits = Math.max(1, maxPermits);
    this.permits = this.maxPermits;
    this.sleep = sleep;
  }

  /** Current in-flight cap — exported state for tests/diagnostics. */
  get permitCount(): number {
    return this.permits;
  }

  /** True while the lane is serving a rate-limit cooldown. */
  get isCoolingDown(): boolean {
    return this.cooldown !== null;
  }

  /** Take a slot, waiting out any cooldown and respecting the current cap. */
  async acquire(): Promise<void> {
    while (this.cooldown !== null || this.inFlight >= this.permits) {
      if (this.cooldown !== null) {
        await this.cooldown;
        continue;
      }
      await new Promise<void>((resolve) => this.waiters.push(resolve));
    }
    this.inFlight++;
  }

  /** Return a slot. Always call from a `finally`. */
  release(): void {
    if (this.inFlight > 0) this.inFlight--;
    this.wakeOne();
  }

  /**
   * Enter (or join) a cooldown of `waitMs` and halve the in-flight cap.
   * Returns the shared cooldown promise — the caller awaits it INSTEAD of
   * sleeping itself, so a burst of 429s costs one cooldown, not one each.
   */
  penalize(waitMs: number): Promise<void> {
    if (this.cooldown !== null) return this.cooldown;
    this.okStreak = 0;
    this.permits = Math.max(1, Math.floor(this.permits / 2));
    const cooldown = this.sleep(waitMs).then(() => {
      if (this.cooldown === cooldown) this.cooldown = null;
      this.wakeAll();
    });
    this.cooldown = cooldown;
    return cooldown;
  }

  /** Record a successful bulk request; recovers one permit per success streak. */
  noteSuccess(): void {
    if (this.permits >= this.maxPermits) return;
    if (++this.okStreak >= LANE_RECOVERY_STREAK) {
      this.okStreak = 0;
      this.permits++;
      this.wakeOne();
    }
  }

  private wakeOne(): void {
    this.waiters.shift()?.();
  }

  private wakeAll(): void {
    const pending = this.waiters;
    this.waiters = [];
    for (const resolve of pending) resolve();
  }
}

// ─────────────────────────── Provider ────────────────────────────────────────

export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
  readonly kind: ProviderKind = "openai";

  private readonly endpoint: string;
  private readonly apiKey?: string;
  private readonly modelId: string;
  private readonly upstreamModel: string;
  private readonly batchSize: number;
  private readonly concurrency: number;
  private readonly timeoutMs: number;
  private readonly fetchImpl: typeof fetch;
  private readonly retryBackoffsMs: readonly number[];
  private readonly rateLimitRetries: number;
  private readonly sleep: (ms: number) => Promise<void>;
  private readonly now: () => number;

  private dimensions: number | undefined = undefined;
  private lastError: string | undefined = undefined;
  readonly breaker: CircuitBreaker;
  /** Shared bulk-traffic lane — see `BulkLaneGate`. */
  readonly lane: BulkLaneGate;

  constructor(config: OpenAIProviderConfig) {
    if (!config.endpoint) {
      throw new Error("OpenAIEmbeddingsProvider: endpoint is required");
    }
    this.endpoint = config.endpoint.replace(/\/+$/, "");
    this.apiKey = config.apiKey;
    this.modelId = config.modelId ?? "embeddinggemma";
    this.upstreamModel = config.upstreamModel ?? this.modelId;
    this.batchSize = config.batchSize ?? DEFAULT_BATCH_SIZE;
    this.concurrency = config.concurrency ?? DEFAULT_CONCURRENCY;
    this.timeoutMs = config.timeoutMs ?? DEFAULT_TIMEOUT_MS;
    this.fetchImpl = config.fetchImpl ?? globalThis.fetch;
    this.retryBackoffsMs = config.retryBackoffsMs ?? RETRY_BACKOFFS_MS;
    this.rateLimitRetries =
      config.rateLimitRetries ??
      parseNonNegativeInt(process.env.QMD_EMBED_RATE_LIMIT_RETRIES) ??
      DEFAULT_RATE_LIMIT_RETRIES;
    this.sleep = config.sleep ?? defaultSleep;
    this.now = config.now ?? Date.now;
    this.breaker = new CircuitBreaker({ now: this.now });
    this.lane = new BulkLaneGate(this.concurrency, this.sleep);

    if (!this.fetchImpl) {
      throw new Error(
        "OpenAIEmbeddingsProvider: global fetch is unavailable. " +
        "Provide a `fetchImpl` config option (Node ≥18 ships fetch by default).",
      );
    }
    if (this.batchSize < 1) {
      throw new Error(`OpenAIEmbeddingsProvider: batchSize must be ≥ 1, got ${this.batchSize}`);
    }
    if (this.concurrency < 1) {
      throw new Error(`OpenAIEmbeddingsProvider: concurrency must be ≥ 1, got ${this.concurrency}`);
    }
  }

  getModelId(): string {
    return this.modelId;
  }

  getDimensions(): number | undefined {
    return this.dimensions;
  }

  /**
   * Most recent per-chunk failure message (HTTP status + body preview, malformed
   * JSON, timeout, abort reason). Returns `undefined` after a successful call
   * or before the first call. See `EmbeddingProvider.getLastError`.
   */
  getLastError(): string | undefined {
    return this.lastError;
  }

  /** Endpoint URL configured at construction time — used by callers when
   *  building error messages for failed first-chunk probes. */
  getEndpoint(): string {
    return this.endpoint;
  }

  async healthcheck(signal?: AbortSignal): Promise<ProviderHealth> {
    // Try GET /health first (worker exposes it). Fall back to probe embed.
    try {
      const { signal: attemptSig, cleanup } = buildAttemptSignal(signal, this.timeoutMs);
      try {
        const resp = await this.fetchImpl(`${this.endpoint}/health`, {
          method: "GET",
          headers: this.buildHeaders(),
          signal: attemptSig,
        });
        if (resp.ok) {
          return {
            ok: true,
            model: this.modelId,
            dimensions: this.dimensions,
            detail: `GET /health → ${resp.status}`,
          };
        }
        return {
          ok: false,
          model: this.modelId,
          detail: `GET /health → HTTP ${resp.status}`,
        };
      } finally {
        cleanup();
      }
    } catch (err) {
      // Endpoint may not implement /health — try a single embed probe instead.
      try {
        const probe = await this.embed("healthcheck", { signal });
        if (probe) {
          return {
            ok: true,
            model: this.modelId,
            dimensions: probe.embedding.length,
            detail: "embed probe ok",
          };
        }
        return {
          ok: false,
          model: this.modelId,
          detail: "embed probe returned null",
        };
      } catch (probeErr) {
        return {
          ok: false,
          model: this.modelId,
          detail:
            (err instanceof Error ? err.message : String(err)) +
            " | probe: " +
            (probeErr instanceof Error ? probeErr.message : String(probeErr)),
        };
      }
    }
  }

  async embed(
    text: string,
    options: ProviderEmbedOptions = {},
  ): Promise<ProviderEmbedding | null> {
    const batch = await this.embedBatch([text], options);
    return batch[0] ?? null;
  }

  async embedBatch(
    texts: string[],
    options: ProviderEmbedOptions = {},
  ): Promise<(ProviderEmbedding | null)[]> {
    if (texts.length === 0) return [];

    if (this.breaker.shouldFailFast()) {
      throw new CircuitOpenError();
    }

    // Bulk = anything that isn't a single query-time embed. Reindex runs go
    // through the self-throttling lane and are attributed as `embeddings-bulk`;
    // `embed()` (search path) stays interactive and is never parked behind a
    // bulk cooldown (i-yghj098h).
    const lane: EmbedLane = texts.length > 1 ? "bulk" : "interactive";

    const chunks = chunkArray(texts, this.batchSize);
    const results: (ProviderEmbedding | null)[] = new Array(texts.length).fill(null);

    // Pre-compute the input-array starting position for each chunk so each
    // worker can write its slice of `results` independently — input order is
    // preserved end-to-end without a final re-sort step.
    const chunkStarts: number[] = new Array(chunks.length);
    {
      let cursor = 0;
      for (let i = 0; i < chunks.length; i++) {
        chunkStarts[i] = cursor;
        cursor += chunks[i]!.length;
      }
    }

    // Shared state across the worker pool. Each transition is final-write,
    // so plain JS scalars are safe — no atomics or locks needed since
    // workers only contend on these via cooperative-scheduled awaits.
    let nextChunkIdx = 0;
    let anySucceeded = false;
    let aborted = false;
    let circuitTrippedDuringRun: CircuitOpenError | null = null;

    // Workers run as parallel async tasks pulling chunks off `nextChunkIdx`
    // until the queue is drained or one of the early-exit flags is set.
    // Concurrency is capped at min(this.concurrency, chunks.length) so we
    // don't spin up idle workers for tiny inputs.
    const workerCount = Math.min(this.concurrency, chunks.length);
    const dispatchOne = async (): Promise<void> => {
      while (true) {
        if (aborted || circuitTrippedDuringRun) return;
        const idx = nextChunkIdx++;
        if (idx >= chunks.length) return;

        const chunk = chunks[idx]!;
        const start = chunkStarts[idx]!;

        // Honor abort/breaker BEFORE issuing the request so we don't waste
        // network for a dispatch we know will be discarded.
        if (options.signal?.aborted) {
          aborted = true;
          this.lastError = `aborted by caller${options.signal.reason ? `: ${String(options.signal.reason)}` : ""}`;
          return;
        }
        if (this.breaker.shouldFailFast()) {
          // Capture the breaker-open intent so we throw it AFTER all
          // currently in-flight workers settle, instead of leaking
          // half-completed results. The thrown error is a fresh instance
          // (matching legacy behavior).
          circuitTrippedDuringRun = new CircuitOpenError();
          return;
        }

        try {
          if (lane === "bulk") await this.lane.acquire();
          let embeddings: number[][];
          try {
            embeddings = await this.requestWithRetry(chunk, options, lane);
          } finally {
            if (lane === "bulk") this.lane.release();
          }
          if (lane === "bulk") this.lane.noteSuccess();
          for (let i = 0; i < chunk.length; i++) {
            const embedding = embeddings[i];
            if (embedding) {
              results[start + i] = {
                embedding,
                model: this.modelId,
              };
              anySucceeded = true;
              // Record dimensions on first success. Concurrent workers may
              // race on this assignment, but they all observe the same
              // length so the race is benign.
              if (this.dimensions === undefined) {
                this.dimensions = embedding.length;
              }
            }
          }
          this.breaker.recordSuccess();
        } catch (err) {
          // A rate limit is backpressure, not a fault. Counting 429s toward the
          // breaker turns "the bucket is empty for 30s" into a 5-minute hard
          // OPEN, which is the outage the retry budget above exists to avoid.
          // The lane cooldown is already the correct control response.
          const rateLimited = err instanceof HttpError && err.status === 429;
          if (!rateLimited) this.breaker.recordFailure();
          if (err instanceof CircuitOpenError) {
            circuitTrippedDuringRun = err;
            return;
          }
          // Last-write-wins on lastError matches the legacy semantics — under
          // concurrency multiple workers may fail in the same call, but the
          // lastError just needs to surface "the most recent cause."
          this.lastError = this.formatErrorContext(err);
          if (process.env.QMD_EMBED_DEBUG) {
            process.stderr.write(
              `OpenAIEmbeddingsProvider: chunk failed (${err instanceof Error ? err.message : String(err)})\n`,
            );
          }
        }
      }
    };

    await Promise.all(Array.from({ length: workerCount }, () => dispatchOne()));

    // If a worker observed `shouldFailFast()` mid-run, surface the error
    // after all in-flight workers have settled.
    if (circuitTrippedDuringRun) throw circuitTrippedDuringRun;

    // Clear lastError on a fully-successful sweep (every input got an embedding).
    if (anySucceeded && results.every((r) => r !== null)) {
      this.lastError = undefined;
    }

    return results;
  }

  async dispose(): Promise<void> {
    // Nothing to release — fetch handles its own connection pooling.
    // Reset the breaker so a re-instantiation starts fresh.
    this.breaker.reset();
  }

  // ────────────────────── Internals ──────────────────────

  /**
   * Format a request-failure context string for `lastError`. Includes endpoint
   * + HTTP status + body preview when the error was an `HttpError`, otherwise
   * falls back to the message of the underlying error (or the value itself
   * when not an Error). Kept short — body preview is already capped at 1024
   * chars by `HttpError`, but we trim further here for the dimension-probe
   * thrown error which surfaces directly to users.
   */
  private formatErrorContext(err: unknown): string {
    if (err instanceof HttpError) {
      const preview = err.bodyPreview.replace(/\s+/g, " ").trim().slice(0, 240);
      return `endpoint=${this.endpoint}/v1/embeddings status=${err.status}${preview ? ` body="${preview}"` : ""}`;
    }
    if (err instanceof Error) {
      return `endpoint=${this.endpoint}/v1/embeddings error="${err.message}"`;
    }
    return `endpoint=${this.endpoint}/v1/embeddings error="${String(err)}"`;
  }

  private buildHeaders(lane: EmbedLane = "interactive"): Record<string, string> {
    const headers: Record<string, string> = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      // Advisory caller attribution for the ai.mm.mk gateway (NULL-safe, never
      // auth). Single chokepoint for both /health and /v1/embeddings.
      // `via` distinguishes bulk reindex traffic from query-time embeds so the
      // gateway can budget the two separately (i-yghj098h part 1b) — the
      // gateway-side per-caller bucket keys off this label.
      "X-AI-Caller": aiCallerHeaderValue(
        lane === "bulk" ? "embeddings-bulk" : "embeddings",
      ),
    };
    if (this.apiKey) {
      headers["Authorization"] = `Bearer ${this.apiKey}`;
    }
    return headers;
  }

  /**
   * Single HTTP request with retry on 429/503. Returns embeddings indexed
   * the same as `texts`. Throws on non-retryable failure or all attempts
   * exhausted.
   */
  private async requestWithRetry(
    texts: string[],
    options: ProviderEmbedOptions,
    lane: EmbedLane = "interactive",
  ): Promise<number[][]> {
    let lastErr: unknown = null;
    // 429s draw on their own budget: a token bucket that refills in 30s must
    // not be able to exhaust a schedule whose total wait is 21s (i-yghj098h).
    let backoffAttempt = 0;
    let rateLimitAttempt = 0;

    for (;;) {
      // Honor user abort BEFORE issuing the call (avoids wasted network)
      if (options.signal?.aborted) {
        throw new Error("aborted by caller");
      }

      try {
        return await this.requestOnce(texts, options, lane);
      } catch (err) {
        lastErr = err;
        const status = err instanceof HttpError ? err.status : 0;
        if (!isRetryableStatus(status)) throw err;

        if (status === 429) {
          if (rateLimitAttempt >= this.rateLimitRetries) throw err;
          const waitMs = this.rateLimitWaitMs(err as HttpError, rateLimitAttempt);
          rateLimitAttempt++;
          if (lane === "bulk") {
            // Pause the whole pool, not just this worker: the other workers
            // hammering the same empty bucket is exactly what turns one 429
            // into a cascade, and what starves interactive callers.
            await this.lane.penalize(waitMs);
          } else {
            await this.sleep(waitMs);
          }
          continue;
        }

        if (backoffAttempt >= this.retryBackoffsMs.length) break;
        await this.sleep(this.retryBackoffsMs[backoffAttempt]!);
        backoffAttempt++;
      }
    }

    // Exhausted retries → throw the last error so caller marks the chunk null
    throw lastErr ?? new Error("requestWithRetry exhausted");
  }

  /**
   * How long to wait after a 429. The server's own `Retry-After` wins when it
   * gave one; otherwise fall back to the configured schedule (so injected test
   * schedules stay authoritative) and then to an exponential 5s/10s/20s… ramp.
   * Always clamped to `RATE_LIMIT_MAX_BACKOFF_MS`.
   */
  private rateLimitWaitMs(err: HttpError, rateLimitAttempt: number): number {
    const scheduled =
      this.retryBackoffsMs[
        Math.min(rateLimitAttempt, this.retryBackoffsMs.length - 1)
      ];
    const fallback =
      scheduled ?? RATE_LIMIT_BASE_BACKOFF_MS * Math.pow(2, rateLimitAttempt);
    return Math.min(
      Math.max(err.retryAfterMs ?? 0, fallback),
      RATE_LIMIT_MAX_BACKOFF_MS,
    );
  }

  /**
   * Issue one HTTP attempt to `POST /v1/embeddings`. Does NOT retry.
   */
  private async requestOnce(
    texts: string[],
    options: ProviderEmbedOptions,
    lane: EmbedLane = "interactive",
  ): Promise<number[][]> {
    const { signal: attemptSig, cleanup } = buildAttemptSignal(options.signal, this.timeoutMs);
    try {
      const body = JSON.stringify({
        model: options.model ?? this.upstreamModel,
        input: texts,
      });
      const resp = await this.fetchImpl(`${this.endpoint}/v1/embeddings`, {
        method: "POST",
        headers: this.buildHeaders(lane),
        body,
        signal: attemptSig,
      });

      if (!resp.ok) {
        const text = await resp.text().catch(() => "");
        // ai.mm.mk states the cooldown in the JSON body, not always in a
        // Retry-After header — read both (i-yghj098h).
        const retryAfterMs =
          resp.status === 429
            ? parseRetryAfterMs(
                resp.headers?.get?.("retry-after"),
                text,
                this.now,
              )
            : undefined;
        throw new HttpError(resp.status, text, retryAfterMs);
      }

      let parsed: OpenAIEmbeddingsResponse;
      try {
        parsed = (await resp.json()) as OpenAIEmbeddingsResponse;
      } catch (err) {
        throw new Error(
          `OpenAIEmbeddingsProvider: malformed JSON from ${this.endpoint}/v1/embeddings: ${err instanceof Error ? err.message : String(err)}`,
        );
      }

      if (!parsed || !Array.isArray(parsed.data)) {
        throw new Error(
          `OpenAIEmbeddingsProvider: response missing "data" array (got ${typeof parsed})`,
        );
      }

      // Sort by index to match input order (in case server returns out-of-order).
      const out: number[][] = new Array(texts.length);
      for (const item of parsed.data) {
        if (
          typeof item.index !== "number" ||
          item.index < 0 ||
          item.index >= texts.length
        ) {
          throw new Error(
            `OpenAIEmbeddingsProvider: data item index out of range (${item.index}, expected 0..${texts.length - 1})`,
          );
        }
        if (!Array.isArray(item.embedding)) {
          throw new Error(
            `OpenAIEmbeddingsProvider: data[${item.index}].embedding is not an array`,
          );
        }
        out[item.index] = item.embedding;
      }

      // Sanity check — every slot must be filled
      for (let i = 0; i < texts.length; i++) {
        if (!out[i]) {
          throw new Error(
            `OpenAIEmbeddingsProvider: response missing embedding for index ${i}`,
          );
        }
      }
      return out;
    } finally {
      cleanup();
    }
  }
}