fix(providers): rate-limit resilience + non-fatal ingest failures

src/orchestrator/phases/ingest.ts

@@ -115,10 +115,12 @@ export async function runIngestPhase(
           status: "failed",
           error,
         })
-        logger.error(`Failed to ingest ${question.questionId}: ${error}`)
-        throw new Error(
-          `Ingest failed at ${question.questionId}: ${error}. Fix the issue and resume with the same run ID.`
-        )
+        // Continue the run: one bad question must not abort a multi-hour benchmark.
+        // The checkpoint carries the failure — indexing only picks up questions whose
+        // ingest completed, so failed ones are skipped downstream, and resume (which
+        // filters on status !== "completed") retries them under the same run ID.
+        logger.error(`Failed to ingest ${question.questionId}: ${error} (skipping, run continues)`)
+        return { questionId: question.questionId, durationMs: Date.now() - startTime }
       }
     },
   })

src/providers/mem0/index.ts

@@ -51,11 +51,35 @@ const CUSTOM_INSTRUCTIONS = `Generate personal memories that follow these guidel
 
 5. Format each memory as a paragraph with a clear narrative structure that captures the person's experience, challenges, and aspirations`
 
+/**
+ * Retry an async op with exponential backoff on mem0 rate-limit (429) errors.
+ * Non-rate-limit errors rethrow immediately.
+ */
+async function withRetry<T>(fn: () => Promise<T>, label: string, maxAttempts = 6): Promise<T> {
+  let delayMs = 1000
+  for (let attempt = 1; ; attempt++) {
+    try {
+      return await fn()
+    } catch (e) {
+      const msg = e instanceof Error ? e.message : String(e)
+      const isRateLimit = /rate limit|too many|429|too frequently/i.test(msg)
+      if (!isRateLimit || attempt >= maxAttempts) throw e
+      logger.warn(`mem0 ${label}: rate-limited (attempt ${attempt}/${maxAttempts}), backing off ${delayMs}ms`)
+      await new Promise((r) => setTimeout(r, delayMs))
+      delayMs = Math.min(delayMs * 2, 30000)
+    }
+  }
+}
+
 export class Mem0Provider implements Provider {
   name = "mem0"
   prompts = MEM0_PROMPTS
+  // mem0's paid tier raises the monthly quota but keeps a per-second RPS cap;
+  // ingest fans out one add per session, so cap it low and lean on withRetry.
   concurrency = {
     default: 50,
+    ingest: 8,
+    search: 20,
   }
   private client: MemoryClient | null = null
   private apiKey: string = ""
@@ -81,10 +105,15 @@ export class Mem0Provider implements Provider {
     const eventIds: string[] = []
 
     for (const session of sessions) {
-      const messages = session.messages.map((m) => ({
-        role: m.role,
-        content: m.content,
-      }))
+      const messages = session.messages
+        .filter((m) => m.content && m.content.trim().length > 0)
+        .map((m) => ({
+          role: m.role,
+          content: m.content,
+        }))
+
+      // mem0 rejects empty adds; skip sessions with no usable messages.
+      if (messages.length === 0) continue
 
       const addOptions: MemoryOptions = {
         user_id: options.containerTag,
@@ -99,7 +128,10 @@ export class Mem0Provider implements Provider {
         },
       }
 
-      const result = (await this.client.add(messages, addOptions)) as Array<{
+      const result = (await withRetry(
+        () => this.client!.add(messages, addOptions),
+        `add ${session.sessionId}`
+      )) as Array<{
         event_id?: string
       }>
       for (const event of result) {
@@ -182,7 +214,10 @@ export class Mem0Provider implements Provider {
       output_format: "v1.1",
     }
 
-    const response = await this.client.search(query, searchOptions)
+    const response = await withRetry(
+      () => this.client!.search(query, searchOptions),
+      "search"
+    )
 
     const res = response as { results?: unknown[] }
     return res.results ?? []

src/providers/supermemory/index.ts

@@ -11,13 +11,38 @@ import type { UnifiedSession } from "../../types/unified"
 import { logger } from "../../utils/logger"
 import { SUPERMEMORY_PROMPTS } from "./prompts"
 
+/**
+ * Retry an async op with backoff on supermemory rate-limit (429) errors.
+ * Honors the server's retryAfterSeconds when present; else exponential.
+ * Non-rate-limit errors rethrow immediately.
+ */
+async function withRetry<T>(fn: () => Promise<T>, label: string, maxAttempts = 6): Promise<T> {
+  let delayMs = 1000
+  for (let attempt = 1; ; attempt++) {
+    try {
+      return await fn()
+    } catch (e) {
+      const msg = e instanceof Error ? e.message : String(e)
+      const isRateLimit = /rate limit|too many|429|rate_limited/i.test(msg)
+      if (!isRateLimit || attempt >= maxAttempts) throw e
+      const retryAfter = msg.match(/retryafterseconds"?\s*:?\s*(\d+)/i)
+      const waitMs = retryAfter ? (parseInt(retryAfter[1], 10) + 1) * 1000 : delayMs
+      logger.warn(`supermemory ${label}: rate-limited (attempt ${attempt}/${maxAttempts}), backing off ${waitMs}ms`)
+      await new Promise((r) => setTimeout(r, waitMs))
+      delayMs = Math.min(delayMs * 2, 30000)
+    }
+  }
+}
+
 export class SupermemoryProvider implements Provider {
   name = "supermemory"
   prompts = SUPERMEMORY_PROMPTS
+  // Free/standard tier enforces a per-key RPS cap (429 with retryAfterSeconds);
+  // keep ingest modest and lean on withRetry.
   concurrency = {
     default: 50,
-    ingest: 100,
-    indexing: 200,
+    ingest: 8,
+    indexing: 50,
   }
   private client: Supermemory | null = null
 
@@ -44,14 +69,18 @@ export class SupermemoryProvider implements Provider {
         ? `Here is the date the following session took place: ${formattedDate}\n\nHere is the session as a stringified JSON:\n${sessionStr}`
         : `Here is the session as a stringified JSON:\n${sessionStr}`
 
-      const response = await this.client.add({
-        content,
-        containerTag: options.containerTag,
-        metadata: {
-          sessionId: session.sessionId,
-          ...(isoDate ? { date: isoDate } : {}),
-        },
-      })
+      const response = await withRetry(
+        () =>
+          this.client!.add({
+            content,
+            containerTag: options.containerTag,
+            metadata: {
+              sessionId: session.sessionId,
+              ...(isoDate ? { date: isoDate } : {}),
+            },
+          }),
+        `add ${session.sessionId}`
+      )
       documentIds.push(response.id)
       logger.debug(`Ingested session ${session.sessionId}`)
     }
@@ -120,17 +149,21 @@ export class SupermemoryProvider implements Provider {
   async search(query: string, options: SearchOptions): Promise<unknown[]> {
     if (!this.client) throw new Error("Provider not initialized")
 
-    const response = await this.client.search.memories({
-      q: query,
-      containerTag: options.containerTag,
-      limit: 30,
-      threshold: options.threshold || 0.3,
-			searchMode: "hybrid",
-			include: {
-				summaries: true,
-				chunks: true
-      }
-    })
+    const response = await withRetry(
+      () =>
+        this.client!.search.memories({
+          q: query,
+          containerTag: options.containerTag,
+          limit: options.limit ?? 30,
+          threshold: options.threshold || 0.3,
+          searchMode: "hybrid",
+          include: {
+            summaries: true,
+            chunks: true,
+          },
+        }),
+      "search"
+    )
 
     return response.results || []
   }