fix(search): reserve interactive embeddings from their own admission lane

A search embeds its query through the same per-credential admission bucket
as bulk indexing, so during a crawl one short interactive request competed
with hundreds of batches for the bucket's refill. Non-checkpointed callers
now reserve from a separate lane under the same identity; the provider
cooldown and quota gates stay shared, so a provider 429 or an exhausted
balance still pauses every caller.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JBacX6HGVhPMUySuMfANwn

Author: waleedlatif1

apps/sim/lib/core/rate-limiter/provider-admission.test.ts

@@ -86,6 +86,19 @@ describe('provider admission', () => {
     )
   })
 
+  it('reserves an interactive lane from its own buckets while sharing the provider gates', async () => {
+    await waitForProviderAdmission({ ...INPUT, lane: 'interactive' })
+    const [reservations, options] = consumeTokens.mock.calls[0]
+    expect(reservations.map((item: { key: string }) => item.key)).toEqual([
+      'provider:embedding:openai:hashed-credential:interactive:tokens',
+      'provider:embedding:openai:hashed-credential:interactive:requests',
+    ])
+    expect(options.cooldownKeys).toEqual([
+      'provider:embedding:openai:hashed-credential:cooldown',
+      'provider:embedding:openai:hashed-credential:quota',
+    ])
+  })
+
   it('isolates another credential and does not impose token costs on OCR', async () => {
     await waitForProviderAdmission({
       ...INPUT,

apps/sim/lib/core/rate-limiter/provider-admission.ts

@@ -12,10 +12,19 @@ export interface ProviderIdentity {
   operation: 'embedding' | 'ocr' | 'rerank'
 }
 
+/**
+ * A lane reserves from its own request and token buckets under the same
+ * provider identity, so a person waiting on one embedding never queues behind
+ * a bulk crawl's batches. Cooldown and quota gates stay per identity: a
+ * provider pause or an exhausted balance still stops every lane.
+ */
+export type ProviderAdmissionLane = 'interactive'
+
 interface ProviderAdmissionInput extends ProviderIdentity {
   inputTokens?: number
   signal?: AbortSignal
   maxWaitMs: number
+  lane?: ProviderAdmissionLane
 }
 
 /**
@@ -49,6 +58,7 @@ export async function waitForProviderAdmission(input: ProviderAdmissionInput): P
   input.signal?.throwIfAborted()
   const deadlineAt = Date.now() + input.maxWaitMs
   const key = providerKey(input)
+  const bucketKey = input.lane ? `${key}:${input.lane}` : key
   const requestsPerMinute =
     input.operation === 'embedding'
       ? envNumber(env.KB_CONFIG_EMBEDDING_REQUESTS_PER_MINUTE, 600, { min: 1 })
@@ -64,7 +74,7 @@ export async function waitForProviderAdmission(input: ProviderAdmissionInput): P
       throw new Error('Embedding request exceeds the configured per-credential token budget')
     }
     reservations.push({
-      key: `${key}:tokens`,
+      key: `${bucketKey}:tokens`,
       cost: input.inputTokens,
       config: {
         maxTokens: tokensPerMinute,
@@ -74,7 +84,7 @@ export async function waitForProviderAdmission(input: ProviderAdmissionInput): P
     })
   }
   reservations.push({
-    key: `${key}:requests`,
+    key: `${bucketKey}:requests`,
     cost: 1,
     config: {
       maxTokens: Math.min(

apps/sim/lib/embeddings/client.test.ts

@@ -1807,13 +1807,14 @@ describe('durable embedding batches', () => {
     expect(KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS).toBeLessThan(EMBEDDING_RETRY_BUDGET_MS)
   })
 
-  it('limits checkpointed admission waits while retaining the interactive request budget', async () => {
+  it('limits checkpointed admission waits and keeps interactive callers on their own lane', async () => {
     fetchMock.mockImplementation(() => Promise.resolve(jsonResponse(openAIBody([[1]], 7))))
     await embed(['text'], { apiKey: 'fixture-key', checkpoints: memoryCheckpoints() })
     expect(mockAdmit).toHaveBeenLastCalledWith(
-      expect.objectContaining({ maxWaitMs: KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS })
+      expect.objectContaining({ maxWaitMs: KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS, lane: undefined })
     )
     await embed(['text'], { apiKey: 'fixture-key' })
+    expect(mockAdmit).toHaveBeenLastCalledWith(expect.objectContaining({ lane: 'interactive' }))
     expect(mockAdmit.mock.lastCall?.[0].maxWaitMs).toBeGreaterThan(
       KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS
     )

apps/sim/lib/embeddings/client.ts

@@ -11,6 +11,7 @@ import {
 } from '@/lib/core/config/env-capabilities'
 import { isHosted } from '@/lib/core/config/env-flags'
 import {
+  type ProviderAdmissionLane,
   ProviderQuotaExhaustedError,
   recordProviderCooldown,
   waitForProviderAdmission,
@@ -544,7 +545,8 @@ async function callEmbeddingAPI(
   expectedDimensions: number | undefined,
   isBYOK: boolean,
   signal?: AbortSignal,
-  admissionWaitMs = EMBEDDING_RETRY_BUDGET_MS
+  admissionWaitMs = EMBEDDING_RETRY_BUDGET_MS,
+  lane?: ProviderAdmissionLane
 ): Promise<{ embeddings: number[][]; totalTokens: number; dimensions: number }> {
   const admissionIdentity = embeddingAdmissionIdentity({ providerId, quotaCircuitIdentity, isBYOK })
   return retryWithExponentialBackoff(
@@ -563,6 +565,7 @@ async function callEmbeddingAPI(
           ),
           signal: operationSignal,
           maxWaitMs: Math.min(admissionWaitMs, Math.max(0, deadlineAt - Date.now())),
+          lane,
         })
       } catch (error) {
         if (error instanceof ProviderQuotaExhaustedError)
@@ -848,7 +851,13 @@ async function callCheckpointedEmbeddingBatch(
     provider.dimensions,
     provider.isBYOK,
     signal,
-    checkpoints ? KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS : undefined
+    checkpoints ? KNOWLEDGE_EMBEDDING_ADMISSION_WAIT_MS : undefined,
+    /**
+     * Checkpoints mark the bulk indexing path. Everything else has a person
+     * waiting on it, so it reserves from the interactive lane and never
+     * queues behind a crawl's batches.
+     */
+    checkpoints ? undefined : 'interactive'
   )
   if (identity) await checkpoints!.save(identity, result, signal)
   return result