fix(qmd): stop one 429 from aborting a whole embedding run (i-yghj098h)

Bulk reindex traffic shares the interactive ai.mm.mk token bucket, and a
single rate-limit response could poison an entire pass:

- src/store.ts: the >80% error-rate guard compared CUMULATIVE errors and
  fed its own un-attempted remainders back into `errors`, while `break`
  exited only the inner batch loop — so every following document re-tripped
  the guard and added its remainder too. 1 real 429 was measured booking
  740,364 "failures" across 52 aborts. Now: the rate is judged over a
  sliding window of recently ATTEMPTED batches, un-attempted chunks are
  reported as `skipped` (new, additive field) instead of `errors`, and one
  abort ends the run exactly once.

- src/embedding/openai.ts: 429s get their own retry budget and honour the
  server cooldown (Retry-After header, or the "Retry after 29s" prose that
  ai.mm.mk puts in the body) — the old 1s/4s/16s schedule could not outlast
  a 30s bucket refill. 429s no longer count toward the circuit breaker:
  backpressure must not become a 5-minute hard OPEN.

- Bulk lane (BulkLaneGate): a 429 on any worker now pauses the WHOLE pool
  for one coalesced cooldown and halves in-flight concurrency (AIMD, floor
  1, recovering one unit per 8 successes), so reindex traffic yields the
  bucket to interactive callers instead of re-draining it. Multi-input
  batches are attributed `via=embeddings-bulk` in X-AI-Caller so the gateway
  can budget the two lanes separately.

Gateway-side per-caller budget and the dedicated qmd-embed.mm.mk vhost
(issue part 1a/1b) are unchanged and still open — this is the client-side
half. dist/ regenerated by the pre-commit build; no reindex run performed.

Refs: i-yghj098h
Session-Id: d83f4a3e
Resolves: i-yghj098h
Push-Allowed: yes

dist/embedding/openai.d.ts

@@ -10,6 +10,13 @@
  * 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; local fallback is forbidden
@@ -41,6 +48,27 @@ export declare const DEFAULT_TIMEOUT_MS = 30000;
  * (initial + 2 retries) — aligns with issue spec "1s/4s/16s".
  */
 export declare const RETRY_BACKOFFS_MS: readonly number[];
+/**
+ * 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 declare 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 declare const RATE_LIMIT_MAX_BACKOFF_MS = 60000;
+/** Fallback wait when a 429 carries no parseable Retry-After (doubles per attempt). */
+export declare const RATE_LIMIT_BASE_BACKOFF_MS = 5000;
+/**
+ * Consecutive successful bulk requests before the lane additively recovers one
+ * unit of concurrency after a 429-triggered halving (AIMD).
+ */
+export declare 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.
@@ -78,6 +106,12 @@ export type OpenAIProviderConfig = {
     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. */
@@ -100,11 +134,31 @@ export type OpenAIEmbeddingsResponse = {
  * 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";
 /**
  * Determine whether an HTTP status is retryable. 429 (Too Many Requests)
  * and 503 (Service Unavailable) are retried; 4xx (other than 429) are not.
  */
 export declare function isRetryableStatus(status: number): boolean;
+/**
+ * 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 declare function parseRetryAfterMs(headerValue: string | null | undefined, bodyPreview?: string, now?: () => number): number | undefined;
 /**
  * Chunk an array into pieces of ≤ size each. `size` MUST be ≥ 1.
  */
@@ -160,7 +214,55 @@ export declare class CircuitOpenError extends Error {
 export declare class HttpError extends Error {
     readonly status: number;
     readonly bodyPreview: string;
-    constructor(status: number, bodyPreview: string);
+    /** Server-advertised cooldown for 429s, when the response stated one. */
+    readonly retryAfterMs?: number;
+    constructor(status: number, bodyPreview: string, retryAfterMs?: number);
+}
+/**
+ * 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 declare class BulkLaneGate {
+    private readonly maxPermits;
+    private readonly sleep;
+    private permits;
+    private inFlight;
+    private okStreak;
+    private cooldown;
+    private waiters;
+    constructor(maxPermits: number, sleep: (ms: number) => Promise<void>);
+    /** Current in-flight cap — exported state for tests/diagnostics. */
+    get permitCount(): number;
+    /** True while the lane is serving a rate-limit cooldown. */
+    get isCoolingDown(): boolean;
+    /** Take a slot, waiting out any cooldown and respecting the current cap. */
+    acquire(): Promise<void>;
+    /** Return a slot. Always call from a `finally`. */
+    release(): void;
+    /**
+     * 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>;
+    /** Record a successful bulk request; recovers one permit per success streak. */
+    noteSuccess(): void;
+    private wakeOne;
+    private wakeAll;
 }
 export declare class OpenAIEmbeddingsProvider implements EmbeddingProvider {
     readonly kind: ProviderKind;
@@ -173,11 +275,14 @@ export declare class OpenAIEmbeddingsProvider implements EmbeddingProvider {
     private readonly timeoutMs;
     private readonly fetchImpl;
     private readonly retryBackoffsMs;
+    private readonly rateLimitRetries;
     private readonly sleep;
     private readonly now;
     private dimensions;
     private lastError;
     readonly breaker: CircuitBreaker;
+    /** Shared bulk-traffic lane — see `BulkLaneGate`. */
+    readonly lane: BulkLaneGate;
     constructor(config: OpenAIProviderConfig);
     getModelId(): string;
     getDimensions(): number | undefined;
@@ -210,6 +315,13 @@ export declare class OpenAIEmbeddingsProvider implements EmbeddingProvider {
      * exhausted.
      */
     private requestWithRetry;
+    /**
+     * 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;
     /**
      * Issue one HTTP attempt to `POST /v1/embeddings`. Does NOT retry.
      */

dist/embedding/openai.js

@@ -10,6 +10,13 @@
  * 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; local fallback is forbidden
@@ -42,6 +49,27 @@ export const DEFAULT_TIMEOUT_MS = 30_000;
  * (initial + 2 retries) — aligns with issue spec "1s/4s/16s".
  */
 export const RETRY_BACKOFFS_MS = [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.
@@ -54,6 +82,13 @@ export const CIRCUIT_MIN_SAMPLES = 4;
 function defaultSleep(ms) {
     return new Promise((resolve) => setTimeout(resolve, ms));
 }
+/** Parse a non-negative integer env value; `undefined` when unset/invalid. */
+function parseNonNegativeInt(raw) {
+    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
@@ -129,6 +164,44 @@ function buildAttemptSignal(userSignal, timeoutMs) {
 export function isRetryableStatus(status) {
     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, bodyPreview, now = Date.now) {
+    const clamp = (ms) => 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.
  */
@@ -255,11 +328,109 @@ export class CircuitOpenError extends Error {
 export class HttpError extends Error {
     status;
     bodyPreview;
-    constructor(status, bodyPreview) {
+    /** Server-advertised cooldown for 429s, when the response stated one. */
+    retryAfterMs;
+    constructor(status, bodyPreview, retryAfterMs) {
         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 {
+    maxPermits;
+    sleep;
+    permits;
+    inFlight = 0;
+    okStreak = 0;
+    cooldown = null;
+    waiters = [];
+    constructor(maxPermits, sleep) {
+        this.maxPermits = Math.max(1, maxPermits);
+        this.permits = this.maxPermits;
+        this.sleep = sleep;
+    }
+    /** Current in-flight cap — exported state for tests/diagnostics. */
+    get permitCount() {
+        return this.permits;
+    }
+    /** True while the lane is serving a rate-limit cooldown. */
+    get isCoolingDown() {
+        return this.cooldown !== null;
+    }
+    /** Take a slot, waiting out any cooldown and respecting the current cap. */
+    async acquire() {
+        while (this.cooldown !== null || this.inFlight >= this.permits) {
+            if (this.cooldown !== null) {
+                await this.cooldown;
+                continue;
+            }
+            await new Promise((resolve) => this.waiters.push(resolve));
+        }
+        this.inFlight++;
+    }
+    /** Return a slot. Always call from a `finally`. */
+    release() {
+        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) {
+        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() {
+        if (this.permits >= this.maxPermits)
+            return;
+        if (++this.okStreak >= LANE_RECOVERY_STREAK) {
+            this.okStreak = 0;
+            this.permits++;
+            this.wakeOne();
+        }
+    }
+    wakeOne() {
+        this.waiters.shift()?.();
+    }
+    wakeAll() {
+        const pending = this.waiters;
+        this.waiters = [];
+        for (const resolve of pending)
+            resolve();
     }
 }
 // ─────────────────────────── Provider ────────────────────────────────────────
@@ -274,11 +445,14 @@ export class OpenAIEmbeddingsProvider {
     timeoutMs;
     fetchImpl;
     retryBackoffsMs;
+    rateLimitRetries;
     sleep;
     now;
     dimensions = undefined;
     lastError = undefined;
     breaker;
+    /** Shared bulk-traffic lane — see `BulkLaneGate`. */
+    lane;
     constructor(config) {
         if (!config.endpoint) {
             throw new Error("OpenAIEmbeddingsProvider: endpoint is required");
@@ -292,9 +466,14 @@ export class OpenAIEmbeddingsProvider {
         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).");
@@ -392,6 +571,11 @@ export class OpenAIEmbeddingsProvider {
         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 = texts.length > 1 ? "bulk" : "interactive";
         const chunks = chunkArray(texts, this.batchSize);
         const results = new Array(texts.length).fill(null);
         // Pre-compute the input-array starting position for each chunk so each
@@ -442,7 +626,18 @@ export class OpenAIEmbeddingsProvider {
                     return;
                 }
                 try {
-                    const embeddings = await this.requestWithRetry(chunk, options);
+                    if (lane === "bulk")
+                        await this.lane.acquire();
+                    let embeddings;
+                    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) {
@@ -462,7 +657,13 @@ export class OpenAIEmbeddingsProvider {
                     this.breaker.recordSuccess();
                 }
                 catch (err) {
-                    this.breaker.recordFailure();
+                    // 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;
@@ -512,13 +713,16 @@ export class OpenAIEmbeddingsProvider {
         }
         return `endpoint=${this.endpoint}/v1/embeddings error="${String(err)}"`;
     }
-    buildHeaders() {
+    buildHeaders(lane = "interactive") {
         const headers = {
             "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.
-            "X-AI-Caller": aiCallerHeaderValue("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}`;
@@ -530,34 +734,65 @@ export class OpenAIEmbeddingsProvider {
      * the same as `texts`. Throws on non-retryable failure or all attempts
      * exhausted.
      */
-    async requestWithRetry(texts, options) {
+    async requestWithRetry(texts, options, lane = "interactive") {
         let lastErr = null;
-        const maxAttempts = this.retryBackoffsMs.length + 1;
-        for (let attempt = 0; attempt < maxAttempts; attempt++) {
+        // 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);
+                return await this.requestOnce(texts, options, lane);
             }
             catch (err) {
                 lastErr = err;
-                const retryable = err instanceof HttpError ? isRetryableStatus(err.status) : false;
-                if (!retryable)
+                const status = err instanceof HttpError ? err.status : 0;
+                if (!isRetryableStatus(status))
                     throw err;
-                if (attempt < this.retryBackoffsMs.length) {
-                    await this.sleep(this.retryBackoffsMs[attempt]);
+                if (status === 429) {
+                    if (rateLimitAttempt >= this.rateLimitRetries)
+                        throw err;
+                    const waitMs = this.rateLimitWaitMs(err, 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`.
+     */
+    rateLimitWaitMs(err, rateLimitAttempt) {
+        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.
      */
-    async requestOnce(texts, options) {
+    async requestOnce(texts, options, lane = "interactive") {
         const { signal: attemptSig, cleanup } = buildAttemptSignal(options.signal, this.timeoutMs);
         try {
             const body = JSON.stringify({
@@ -566,13 +801,18 @@ export class OpenAIEmbeddingsProvider {
             });
             const resp = await this.fetchImpl(`${this.endpoint}/v1/embeddings`, {
                 method: "POST",
-                headers: this.buildHeaders(),
+                headers: this.buildHeaders(lane),
                 body,
                 signal: attemptSig,
             });
             if (!resp.ok) {
                 const text = await resp.text().catch(() => "");
-                throw new HttpError(resp.status, text);
+                // 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;
             try {

dist/store.d.ts

@@ -327,12 +327,22 @@ export type EmbedProgress = {
     totalChunks: number;
     bytesProcessed: number;
     totalBytes: number;
+    /** Chunks that were sent to the provider and came back without an embedding. */
     errors: number;
+    /**
+     * Chunks the run gave up on WITHOUT sending them (abort / expired session).
+     * Kept separate from `errors` so "we stopped early" can never be reported as
+     * "N chunks failed" — see `generateEmbeddings` (i-yghj098h).
+     */
+    skipped?: number;
 };
 export type EmbedResult = {
     docsProcessed: number;
     chunksEmbedded: number;
+    /** Attempted-and-failed chunks only. Never includes un-attempted ones. */
     errors: number;
+    /** Un-attempted chunks left behind by an early abort. */
+    skipped?: number;
     durationMs: number;
 };
 export type EmbedOptions = {

dist/store.js

@@ -1203,9 +1203,16 @@ export async function generateEmbeddings(store, options) {
     const result = await withEmbedSession(store, provider, async (session) => {
         let chunksEmbedded = 0;
         let errors = 0;
+        let skipped = 0;
         let bytesProcessed = 0;
         let totalChunks = 0;
         let vectorTableInitialized = false;
+        // Set once the run decides to stop early. The high-error-rate guard used to
+        // `break` the INNER batch loop only, so the very next document re-evaluated
+        // the same cumulative ratio and broke again — one transient 429 could book
+        // ~740k never-attempted chunks as "failures" across ~50 aborts
+        // (i-yghj098h). The flag stops the OUTER loop too, exactly once.
+        let abortRun = false;
         // Inner batch size — number of chunks fed into each `embedMany` call.
         // Bumped 32 → 256 (i-fkpnar9i) so the openai provider's concurrent
         // dispatcher receives ≥ 4 sub-chunks of size 64 (worker MAX_BATCH) and
@@ -1213,6 +1220,32 @@ export async function generateEmbeddings(store, options) {
         // Override per-deploy via `QMD_EMBED_INNER_BATCH_SIZE`.
         const BATCH_SIZE = parseInt(process.env.QMD_EMBED_INNER_BATCH_SIZE ?? "256", 10) || 256;
         const batches = buildEmbeddingBatches(docsToEmbed, maxDocsPerBatch, maxBatchBytes);
+        // Sliding error-rate window (i-yghj098h). The abort guard used to compare
+        // CUMULATIVE `errors` against CUMULATIVE processed, and fed its own
+        // un-attempted remainders back into `errors` — so once tripped the ratio
+        // could never fall back under the threshold and the run was poisoned for
+        // good. Judging only the last few ATTEMPTED batches means a burst of 429s
+        // that the provider then rides out (retry + lane cooldown) no longer
+        // condemns the rest of the run.
+        const ERROR_WINDOW_BATCHES = parseInt(process.env.QMD_EMBED_ERROR_WINDOW_BATCHES ?? "8", 10) || 8;
+        const ERROR_RATE_ABORT_THRESHOLD = 0.8;
+        const recentBatches = [];
+        const noteBatchOutcome = (attempted, failed) => {
+            if (attempted <= 0)
+                return;
+            recentBatches.push({ attempted, errors: failed });
+            while (recentBatches.length > ERROR_WINDOW_BATCHES)
+                recentBatches.shift();
+        };
+        const windowTotals = () => {
+            let attempted = 0;
+            let failed = 0;
+            for (const b of recentBatches) {
+                attempted += b.attempted;
+                failed += b.errors;
+            }
+            return { attempted, errors: failed };
+        };
         // Embedding helpers — single point of provider/session selection.
         // Both return the same shape as ILLMSession.embed/embedBatch so the
         // rest of the loop is unchanged.
@@ -1271,7 +1304,7 @@ export async function generateEmbeddings(store, options) {
             totalChunks += batchChunks.length;
             if (batchChunks.length === 0) {
                 bytesProcessed += batchBytes;
-                options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors });
+                options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors, skipped });
                 continue;
             }
             if (!vectorTableInitialized) {
@@ -1313,16 +1346,23 @@ export async function generateEmbeddings(store, options) {
                 // Abort early if session has been invalidated (e.g. max duration exceeded)
                 if (!session.isValid) {
                     const remaining = batchChunks.length - batchStart;
-                    errors += remaining;
+                    skipped += remaining;
+                    abortRun = true;
                     console.warn(`⚠ Session expired — skipping ${remaining} remaining chunks`);
                     break;
                 }
-                // Abort early if error rate is too high (>80% of processed chunks failed)
-                const processed = chunksEmbedded + errors;
-                if (processed >= BATCH_SIZE && errors > processed * 0.8) {
+                // Abort early if the RECENT attempted batches are overwhelmingly failing
+                // (>80% over the sliding window). Un-attempted chunks are booked as
+                // `skipped`, never as `errors`.
+                const window = windowTotals();
+                if (window.attempted >= BATCH_SIZE &&
+                    window.errors > window.attempted * ERROR_RATE_ABORT_THRESHOLD) {
                     const remaining = batchChunks.length - batchStart;
-                    errors += remaining;
-                    console.warn(`⚠ Error rate too high (${errors}/${processed}) — aborting embedding`);
+                    skipped += remaining;
+                    abortRun = true;
+                    console.warn(`⚠ Error rate too high (${window.errors}/${window.attempted} over last ` +
+                        `${recentBatches.length} batches) — aborting embedding ` +
+                        `(${remaining} chunks in this document left unattempted)`);
                     break;
                 }
                 const batchEnd = Math.min(batchStart + BATCH_SIZE, batchChunks.length);
@@ -1361,6 +1401,7 @@ export async function generateEmbeddings(store, options) {
                     const { okCount, errCount } = insertBatchTxn();
                     chunksEmbedded += okCount;
                     errors += errCount;
+                    noteBatchOutcome(chunkBatch.length, errCount);
                     batchChunkBytesProcessed += chunkBatch.reduce((sum, c) => sum + c.bytes, 0);
                 }
                 catch {
@@ -1368,9 +1409,11 @@ export async function generateEmbeddings(store, options) {
                     // But skip if session is already invalid (avoids N doomed retries)
                     if (!session.isValid) {
                         errors += chunkBatch.length;
+                        noteBatchOutcome(chunkBatch.length, chunkBatch.length);
                         batchChunkBytesProcessed += chunkBatch.reduce((sum, c) => sum + c.bytes, 0);
                     }
                     else {
+                        let fallbackErrors = 0;
                         for (const chunk of chunkBatch) {
                             try {
                                 const text = formatDocForEmbedding(chunk.text, chunk.title, embedModelUri);
@@ -1381,13 +1424,16 @@ export async function generateEmbeddings(store, options) {
                                 }
                                 else {
                                     errors++;
+                                    fallbackErrors++;
                                 }
                             }
                             catch {
                                 errors++;
+                                fallbackErrors++;
                             }
                             batchChunkBytesProcessed += chunk.bytes;
                         }
+                        noteBatchOutcome(chunkBatch.length, fallbackErrors);
                     }
                 }
                 const proportionalBytes = totalBatchChunkBytes === 0
@@ -1399,17 +1445,22 @@ export async function generateEmbeddings(store, options) {
                     bytesProcessed: bytesProcessed + proportionalBytes,
                     totalBytes,
                     errors,
+                    skipped,
                 });
             }
             bytesProcessed += batchBytes;
-            options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors });
+            options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors, skipped });
+            // One abort ends the run — it must not re-trip on every remaining doc.
+            if (abortRun)
+                break;
         }
-        return { chunksEmbedded, errors };
+        return { chunksEmbedded, errors, skipped };
     }, { maxDuration: 30 * 60 * 1000, name: 'generateEmbeddings' });
     return {
         docsProcessed: totalDocs,
         chunksEmbedded: result.chunksEmbedded,
         errors: result.errors,
+        skipped: result.skipped,
         durationMs: Date.now() - startTime,
     };
 }

src/embedding/openai.ts

@@ -10,6 +10,13 @@
  * 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; local fallback is forbidden
@@ -57,6 +64,31 @@ export const DEFAULT_TIMEOUT_MS = 30_000;
  */
 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.
@@ -97,6 +129,12 @@ export type OpenAIProviderConfig = {
   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. */
@@ -122,12 +160,27 @@ export type OpenAIEmbeddingsResponse = {
  */
 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
@@ -208,6 +261,51 @@ 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.
  */
@@ -350,11 +448,115 @@ export class CircuitOpenError extends Error {
 export class HttpError extends Error {
   readonly status: number;
   readonly bodyPreview: string;
-  constructor(status: number, 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();
   }
 }
 
@@ -372,12 +574,15 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
   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) {
@@ -392,9 +597,14 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
     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(
@@ -507,6 +717,12 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
       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);
 
@@ -561,7 +777,14 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
         }
 
         try {
-          const embeddings = await this.requestWithRetry(chunk, options);
+          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) {
@@ -580,7 +803,12 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
           }
           this.breaker.recordSuccess();
         } catch (err) {
-          this.breaker.recordFailure();
+          // 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;
@@ -639,13 +867,18 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
     return `endpoint=${this.endpoint}/v1/embeddings error="${String(err)}"`;
   }
 
-  private buildHeaders(): Record<string, string> {
+  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.
-      "X-AI-Caller": aiCallerHeaderValue("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}`;
@@ -661,26 +894,45 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
   private async requestWithRetry(
     texts: string[],
     options: ProviderEmbedOptions,
+    lane: EmbedLane = "interactive",
   ): Promise<number[][]> {
     let lastErr: unknown = null;
-    const maxAttempts = this.retryBackoffsMs.length + 1;
+    // 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 (let attempt = 0; attempt < maxAttempts; attempt++) {
+    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);
+        return await this.requestOnce(texts, options, lane);
       } catch (err) {
         lastErr = err;
-        const retryable =
-          err instanceof HttpError ? isRetryableStatus(err.status) : false;
-        if (!retryable) throw err;
-        if (attempt < this.retryBackoffsMs.length) {
-          await this.sleep(this.retryBackoffsMs[attempt]!);
+        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++;
       }
     }
 
@@ -688,12 +940,32 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
     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 {
@@ -703,14 +975,24 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
       });
       const resp = await this.fetchImpl(`${this.endpoint}/v1/embeddings`, {
         method: "POST",
-        headers: this.buildHeaders(),
+        headers: this.buildHeaders(lane),
         body,
         signal: attemptSig,
       });
 
       if (!resp.ok) {
         const text = await resp.text().catch(() => "");
-        throw new HttpError(resp.status, text);
+        // 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;

src/store.ts

@@ -1353,13 +1353,23 @@ export type EmbedProgress = {
   totalChunks: number;
   bytesProcessed: number;
   totalBytes: number;
+  /** Chunks that were sent to the provider and came back without an embedding. */
   errors: number;
+  /**
+   * Chunks the run gave up on WITHOUT sending them (abort / expired session).
+   * Kept separate from `errors` so "we stopped early" can never be reported as
+   * "N chunks failed" — see `generateEmbeddings` (i-yghj098h).
+   */
+  skipped?: number;
 };
 
 export type EmbedResult = {
   docsProcessed: number;
   chunksEmbedded: number;
+  /** Attempted-and-failed chunks only. Never includes un-attempted ones. */
   errors: number;
+  /** Un-attempted chunks left behind by an early abort. */
+  skipped?: number;
   durationMs: number;
 };
 
@@ -1626,9 +1636,16 @@ export async function generateEmbeddings(
   const result = await withEmbedSession(store, provider, async (session) => {
     let chunksEmbedded = 0;
     let errors = 0;
+    let skipped = 0;
     let bytesProcessed = 0;
     let totalChunks = 0;
     let vectorTableInitialized = false;
+    // Set once the run decides to stop early. The high-error-rate guard used to
+    // `break` the INNER batch loop only, so the very next document re-evaluated
+    // the same cumulative ratio and broke again — one transient 429 could book
+    // ~740k never-attempted chunks as "failures" across ~50 aborts
+    // (i-yghj098h). The flag stops the OUTER loop too, exactly once.
+    let abortRun = false;
     // Inner batch size — number of chunks fed into each `embedMany` call.
     // Bumped 32 → 256 (i-fkpnar9i) so the openai provider's concurrent
     // dispatcher receives ≥ 4 sub-chunks of size 64 (worker MAX_BATCH) and
@@ -1637,6 +1654,33 @@ export async function generateEmbeddings(
     const BATCH_SIZE = parseInt(process.env.QMD_EMBED_INNER_BATCH_SIZE ?? "256", 10) || 256;
     const batches = buildEmbeddingBatches(docsToEmbed, maxDocsPerBatch, maxBatchBytes);
 
+    // Sliding error-rate window (i-yghj098h). The abort guard used to compare
+    // CUMULATIVE `errors` against CUMULATIVE processed, and fed its own
+    // un-attempted remainders back into `errors` — so once tripped the ratio
+    // could never fall back under the threshold and the run was poisoned for
+    // good. Judging only the last few ATTEMPTED batches means a burst of 429s
+    // that the provider then rides out (retry + lane cooldown) no longer
+    // condemns the rest of the run.
+    const ERROR_WINDOW_BATCHES = parseInt(
+      process.env.QMD_EMBED_ERROR_WINDOW_BATCHES ?? "8", 10,
+    ) || 8;
+    const ERROR_RATE_ABORT_THRESHOLD = 0.8;
+    const recentBatches: { attempted: number; errors: number }[] = [];
+    const noteBatchOutcome = (attempted: number, failed: number): void => {
+      if (attempted <= 0) return;
+      recentBatches.push({ attempted, errors: failed });
+      while (recentBatches.length > ERROR_WINDOW_BATCHES) recentBatches.shift();
+    };
+    const windowTotals = (): { attempted: number; errors: number } => {
+      let attempted = 0;
+      let failed = 0;
+      for (const b of recentBatches) {
+        attempted += b.attempted;
+        failed += b.errors;
+      }
+      return { attempted, errors: failed };
+    };
+
     // Embedding helpers — single point of provider/session selection.
     // Both return the same shape as ILLMSession.embed/embedBatch so the
     // rest of the loop is unchanged.
@@ -1715,7 +1759,7 @@ export async function generateEmbeddings(
 
       if (batchChunks.length === 0) {
         bytesProcessed += batchBytes;
-        options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors });
+        options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors, skipped });
         continue;
       }
 
@@ -1764,17 +1808,28 @@ export async function generateEmbeddings(
         // Abort early if session has been invalidated (e.g. max duration exceeded)
         if (!session.isValid) {
           const remaining = batchChunks.length - batchStart;
-          errors += remaining;
+          skipped += remaining;
+          abortRun = true;
           console.warn(`⚠ Session expired — skipping ${remaining} remaining chunks`);
           break;
         }
 
-        // Abort early if error rate is too high (>80% of processed chunks failed)
-        const processed = chunksEmbedded + errors;
-        if (processed >= BATCH_SIZE && errors > processed * 0.8) {
+        // Abort early if the RECENT attempted batches are overwhelmingly failing
+        // (>80% over the sliding window). Un-attempted chunks are booked as
+        // `skipped`, never as `errors`.
+        const window = windowTotals();
+        if (
+          window.attempted >= BATCH_SIZE &&
+          window.errors > window.attempted * ERROR_RATE_ABORT_THRESHOLD
+        ) {
           const remaining = batchChunks.length - batchStart;
-          errors += remaining;
-          console.warn(`⚠ Error rate too high (${errors}/${processed}) — aborting embedding`);
+          skipped += remaining;
+          abortRun = true;
+          console.warn(
+            `⚠ Error rate too high (${window.errors}/${window.attempted} over last ` +
+            `${recentBatches.length} batches) — aborting embedding ` +
+            `(${remaining} chunks in this document left unattempted)`,
+          );
           break;
         }
 
@@ -1814,14 +1869,17 @@ export async function generateEmbeddings(
           const { okCount, errCount } = insertBatchTxn();
           chunksEmbedded += okCount;
           errors += errCount;
+          noteBatchOutcome(chunkBatch.length, errCount);
           batchChunkBytesProcessed += chunkBatch.reduce((sum, c) => sum + c.bytes, 0);
         } catch {
           // Batch failed — try individual embeddings as fallback
           // But skip if session is already invalid (avoids N doomed retries)
           if (!session.isValid) {
             errors += chunkBatch.length;
+            noteBatchOutcome(chunkBatch.length, chunkBatch.length);
             batchChunkBytesProcessed += chunkBatch.reduce((sum, c) => sum + c.bytes, 0);
           } else {
+            let fallbackErrors = 0;
             for (const chunk of chunkBatch) {
               try {
                 const text = formatDocForEmbedding(chunk.text, chunk.title, embedModelUri);
@@ -1831,12 +1889,15 @@ export async function generateEmbeddings(
                   chunksEmbedded++;
                 } else {
                   errors++;
+                  fallbackErrors++;
                 }
               } catch {
                 errors++;
+                fallbackErrors++;
               }
               batchChunkBytesProcessed += chunk.bytes;
             }
+            noteBatchOutcome(chunkBatch.length, fallbackErrors);
           }
         }
 
@@ -1849,20 +1910,25 @@ export async function generateEmbeddings(
           bytesProcessed: bytesProcessed + proportionalBytes,
           totalBytes,
           errors,
+          skipped,
         });
       }
 
       bytesProcessed += batchBytes;
-      options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors });
+      options?.onProgress?.({ chunksEmbedded, totalChunks, bytesProcessed, totalBytes, errors, skipped });
+
+      // One abort ends the run — it must not re-trip on every remaining doc.
+      if (abortRun) break;
     }
 
-    return { chunksEmbedded, errors };
+    return { chunksEmbedded, errors, skipped };
   }, { maxDuration: 30 * 60 * 1000, name: 'generateEmbeddings' });
 
   return {
     docsProcessed: totalDocs,
     chunksEmbedded: result.chunksEmbedded,
     errors: result.errors,
+    skipped: result.skipped,
     durationMs: Date.now() - startTime,
   };
 }

test/embedding-openai.test.ts

@@ -1098,3 +1098,218 @@ describe("OpenAIEmbeddingsProvider — dispose", () => {
     expect(p.breaker.getState()).toBe("closed");
   });
 });
+
+// ────────── rate-limit budget + bulk lane (i-yghj098h) ───────────────────────
+
+import {
+  parseRetryAfterMs,
+  BulkLaneGate,
+  RATE_LIMIT_MAX_BACKOFF_MS,
+  DEFAULT_RATE_LIMIT_RETRIES,
+  LANE_RECOVERY_STREAK,
+} from "../src/embedding/openai.js";
+
+/** The shape ai.mm.mk actually returns when the token bucket is empty. */
+function rateLimitedResponse(seconds: number, opts?: { header?: boolean }): Response {
+  const headers: Record<string, string> = { "content-type": "application/json" };
+  if (opts?.header) headers["retry-after"] = String(seconds);
+  return new Response(
+    JSON.stringify({
+      detail: `Rate limit exceeded (tokens). Retry after ${seconds}s.`,
+      error: "rate_limited",
+    }),
+    { status: 429, headers },
+  );
+}
+
+describe("parseRetryAfterMs", () => {
+  test("delta-seconds header", () => {
+    expect(parseRetryAfterMs("29")).toBe(29_000);
+  });
+  test("HTTP-date header", () => {
+    const now = () => 1_000_000;
+    const when = new Date(now() + 12_000).toUTCString();
+    // toUTCString truncates to whole seconds — allow the rounding slack.
+    expect(parseRetryAfterMs(when, undefined, now)).toBeGreaterThan(11_000);
+  });
+  test("falls back to the ai.mm.mk body prose", () => {
+    expect(
+      parseRetryAfterMs(null, '{"detail":"Rate limit exceeded (tokens). Retry after 29s."}'),
+    ).toBe(29_000);
+  });
+  test("clamps absurd values", () => {
+    expect(parseRetryAfterMs("99999")).toBe(RATE_LIMIT_MAX_BACKOFF_MS);
+  });
+  test("undefined when nothing parseable", () => {
+    expect(parseRetryAfterMs(null, "no numbers here")).toBeUndefined();
+    expect(parseRetryAfterMs("0")).toBeUndefined();
+  });
+});
+
+describe("OpenAIEmbeddingsProvider — 429 honours Retry-After (i-yghj098h)", () => {
+  test("waits the server-advertised cooldown from the body, not the 1s schedule", async () => {
+    const sleepCalls: number[] = [];
+    const { fetchImpl } = makeFetchSequence([
+      () => rateLimitedResponse(29),
+      () => embeddingsResponse(["x"], 4),
+    ]);
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      retryBackoffsMs: [1_000, 4_000, 16_000],
+      sleep: async (ms) => { sleepCalls.push(ms); },
+    });
+    const r = await p.embed("x");
+    expect(r).not.toBeNull();
+    expect(sleepCalls).toEqual([29_000]);
+  });
+
+  test("Retry-After header wins over the body", async () => {
+    const sleepCalls: number[] = [];
+    const { fetchImpl } = makeFetchSequence([
+      () => rateLimitedResponse(7, { header: true }),
+      () => embeddingsResponse(["x"], 4),
+    ]);
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      retryBackoffsMs: [],
+      sleep: async (ms) => { sleepCalls.push(ms); },
+    });
+    expect(await p.embed("x")).not.toBeNull();
+    expect(sleepCalls).toEqual([7_000]);
+  });
+
+  test("429 gets its own budget — survives more rate limits than the generic schedule", async () => {
+    const sleepCalls: number[] = [];
+    const { fetchImpl, calls } = makeFetchSequence([
+      () => rateLimitedResponse(1),
+      () => rateLimitedResponse(1),
+      () => rateLimitedResponse(1),
+      () => rateLimitedResponse(1),
+      () => embeddingsResponse(["x"], 4),
+    ]);
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      // A single generic retry — before the fix this run gave up on call 2.
+      retryBackoffsMs: [10],
+      sleep: async (ms) => { sleepCalls.push(ms); },
+    });
+    expect(await p.embed("x")).not.toBeNull();
+    expect(calls.length).toBe(5);
+    expect(sleepCalls.length).toBe(4);
+    expect(DEFAULT_RATE_LIMIT_RETRIES).toBeGreaterThanOrEqual(4);
+  });
+
+  test("429 exhaustion still fails, and does not trip the circuit breaker", async () => {
+    const { fetchImpl } = makeFetchSequence(
+      Array.from({ length: 12 }, () => () => rateLimitedResponse(1)),
+    );
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      retryBackoffsMs: [],
+      rateLimitRetries: 1,
+      sleep: async () => {},
+    });
+    for (let i = 0; i < 6; i++) expect(await p.embed("x")).toBeNull();
+    // Backpressure must not become a 5-minute hard OPEN.
+    expect(p.breaker.getState()).toBe("closed");
+  });
+});
+
+describe("BulkLaneGate", () => {
+  test("halves permits per cooldown and coalesces concurrent penalties", async () => {
+    const slept: number[] = [];
+    const gate = new BulkLaneGate(4, async (ms) => { slept.push(ms); });
+    await Promise.all([gate.penalize(5_000), gate.penalize(5_000)]);
+    expect(slept).toEqual([5_000]);
+    expect(gate.permitCount).toBe(2);
+    await gate.penalize(5_000);
+    expect(gate.permitCount).toBe(1);
+    await gate.penalize(5_000);
+    expect(gate.permitCount).toBe(1); // floor
+  });
+
+  test("recovers one permit per success streak", async () => {
+    const gate = new BulkLaneGate(4, async () => {});
+    await gate.penalize(1_000);
+    expect(gate.permitCount).toBe(2);
+    for (let i = 0; i < LANE_RECOVERY_STREAK; i++) gate.noteSuccess();
+    expect(gate.permitCount).toBe(3);
+  });
+
+  test("acquire/release respects the current cap", async () => {
+    const gate = new BulkLaneGate(1, async () => {});
+    await gate.acquire();
+    let secondEntered = false;
+    const second = gate.acquire().then(() => { secondEntered = true; });
+    await Promise.resolve();
+    expect(secondEntered).toBe(false);
+    gate.release();
+    await second;
+    expect(secondEntered).toBe(true);
+  });
+});
+
+describe("bulk lane — X-AI-Caller attribution + shared cooldown", () => {
+  function callerHeader(init?: RequestInit): string {
+    const headers = (init?.headers ?? {}) as Record<string, string>;
+    return headers["X-AI-Caller"] ?? "";
+  }
+
+  test("multi-input batches are attributed as embeddings-bulk", async () => {
+    const { fetchImpl, calls } = makeFetchSequence([
+      () => embeddingsResponse(["a", "b"], 4),
+    ]);
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      sleep: async () => {},
+    });
+    await p.embedBatch(["a", "b"]);
+    expect(callerHeader(calls[0]!.init)).toContain("via=embeddings-bulk");
+    expect(callerHeader(calls[0]!.init)).toContain("comp=qmd");
+  });
+
+  test("single query-time embeds stay on the interactive label", async () => {
+    const { fetchImpl, calls } = makeFetchSequence([
+      () => embeddingsResponse(["a"], 4),
+    ]);
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      sleep: async () => {},
+    });
+    await p.embed("a");
+    expect(callerHeader(calls[0]!.init)).toContain("via=embeddings");
+    expect(callerHeader(calls[0]!.init)).not.toContain("via=embeddings-bulk");
+  });
+
+  test("one 429 pauses the whole bulk pool once and narrows concurrency", async () => {
+    const sleepCalls: number[] = [];
+    let call = 0;
+    const fetchImpl = (async (_input: unknown, _init?: RequestInit) => {
+      call++;
+      if (call === 1) return rateLimitedResponse(30);
+      return embeddingsResponse(["x"], 4);
+    }) as unknown as typeof fetch;
+
+    const p = new OpenAIEmbeddingsProvider({
+      endpoint: "https://ai.mm.mk",
+      fetchImpl,
+      batchSize: 1,
+      concurrency: 2,
+      retryBackoffsMs: [],
+      sleep: async (ms) => { sleepCalls.push(ms); },
+    });
+
+    const out = await p.embedBatch(["a", "b", "c", "d"]);
+    expect(out.every((r) => r !== null)).toBe(true);
+    // Exactly one cooldown, at the length the gateway asked for.
+    expect(sleepCalls).toEqual([30_000]);
+    // AIMD: the lane gave capacity back to interactive traffic.
+    expect(p.lane.permitCount).toBe(1);
+  });
+});

test/embedding-store-integration.test.ts

@@ -387,3 +387,80 @@ describe("first-chunk dimension probe — retry + rich error (i-vm1lxwry)", () =
     expect(provider2.callIdx).toBeGreaterThanOrEqual(2);
   });
 });
+
+// ────────── high-error-rate abort accounting (i-yghj098h) ────────────────────
+
+/**
+ * Insert `count` extra documents so a run has several chunks to walk through.
+ * Bodies differ so the chunker/embedder can't collapse them.
+ */
+function insertExtraDocs(s: Store, count: number): void {
+  const now = "2026-04-27T00:00:00Z";
+  for (let i = 0; i < count; i++) {
+    const hash = `extra${i}`;
+    s.db
+      .prepare(`INSERT INTO content (hash, doc, created_at) VALUES (?, ?, ?)`)
+      .run(hash, `Extra document number ${i} with its own body text to chunk.`, now);
+    s.db
+      .prepare(
+        `INSERT INTO documents (hash, collection, path, title, created_at, modified_at, active) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+      )
+      .run(hash, "test", `extra-${i}.md`, `Extra ${i}`, now, now, 1);
+  }
+}
+
+describe("generateEmbeddings — abort accounting (i-yghj098h)", () => {
+  const prevInnerBatch = process.env.QMD_EMBED_INNER_BATCH_SIZE;
+
+  afterEach(() => {
+    if (prevInnerBatch === undefined) delete process.env.QMD_EMBED_INNER_BATCH_SIZE;
+    else process.env.QMD_EMBED_INNER_BATCH_SIZE = prevInnerBatch;
+  });
+
+  test("an early abort books un-attempted chunks as skipped, never as errors", async () => {
+    // One chunk per HTTP batch so a single failure is enough to trip the
+    // >80% guard on the very next iteration.
+    process.env.QMD_EMBED_INNER_BATCH_SIZE = "1";
+    insertExtraDocs(store, 4);
+
+    // probe ok → first real batch fails → everything after would succeed,
+    // so anything counted as an error past the abort is phantom.
+    const provider = new FlakyProvider("embeddinggemma", 4, [true, false, true]);
+
+    let totalChunks = 0;
+    const result = await generateEmbeddings(store, {
+      embedProvider: provider,
+      onProgress: (p) => { totalChunks = Math.max(totalChunks, p.totalChunks); },
+    });
+
+    // Exactly one chunk was actually sent and came back empty.
+    expect(result.errors).toBe(1);
+    // The rest were never attempted — they are skipped, not failures. Before
+    // the fix `errors` absorbed every un-attempted remainder on every
+    // subsequent document, so it could exceed the corpus several times over.
+    expect(result.skipped ?? 0).toBeGreaterThan(0);
+    expect(result.chunksEmbedded).toBe(0);
+    // The books must balance: nothing can be counted twice.
+    expect(result.chunksEmbedded + result.errors + (result.skipped ?? 0)).toBe(totalChunks);
+    expect(result.errors).toBeLessThanOrEqual(totalChunks);
+  });
+
+  test("an isolated failed batch does not abort the rest of the run", async () => {
+    process.env.QMD_EMBED_INNER_BATCH_SIZE = "2";
+    insertExtraDocs(store, 8);
+
+    // probe ok, batch1 ok, batch2 empty, everything after ok → the window
+    // error rate falls back under the threshold and the run continues.
+    const provider = new FlakyProvider("embeddinggemma", 4, [true, true, false, true]);
+
+    let totalChunks = 0;
+    const result = await generateEmbeddings(store, {
+      embedProvider: provider,
+      onProgress: (p) => { totalChunks = Math.max(totalChunks, p.totalChunks); },
+    });
+
+    expect(result.chunksEmbedded).toBeGreaterThan(0);
+    expect(result.skipped ?? 0).toBe(0);
+    expect(result.chunksEmbedded + result.errors).toBe(totalChunks);
+  });
+});