diff --git a/src/orchestrator/phases/ingest.ts b/src/orchestrator/phases/ingest.ts index e38815d..f765079 100644 --- a/src/orchestrator/phases/ingest.ts +++ b/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 } } }, }) diff --git a/src/providers/mem0/index.ts b/src/providers/mem0/index.ts index e01d343..9a31472 100644 --- a/src/providers/mem0/index.ts +++ b/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(fn: () => Promise, label: string, maxAttempts = 6): Promise { + 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 ?? [] diff --git a/src/providers/supermemory/index.ts b/src/providers/supermemory/index.ts index 027bc32..59d8056 100644 --- a/src/providers/supermemory/index.ts +++ b/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(fn: () => Promise, label: string, maxAttempts = 6): Promise { + 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 { 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 || [] }