From 7bbed5e1c94a40818cd633e6f2567a7a7e9eced2 Mon Sep 17 00:00:00 2001 From: "dkoosis@gmail.com" Date: Sun, 2 Aug 2026 11:18:08 -0400 Subject: [PATCH 1/2] fix(supermemory): honor options.limit in search() instead of hardcoding 30 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit search() hardcoded limit: 30, silently overriding whatever limit the caller passed in SearchOptions — so a benchmark requesting a deeper retrieval pool (e.g. limit 50) still capped at 30. Honor options.limit, keeping 30 as the default. Flagged by @sohamd22 in PR #44 review. --- src/providers/supermemory/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/providers/supermemory/index.ts b/src/providers/supermemory/index.ts index 027bc32..cd595c8 100644 --- a/src/providers/supermemory/index.ts +++ b/src/providers/supermemory/index.ts @@ -123,7 +123,7 @@ export class SupermemoryProvider implements Provider { const response = await this.client.search.memories({ q: query, containerTag: options.containerTag, - limit: 30, + limit: options.limit ?? 30, threshold: options.threshold || 0.3, searchMode: "hybrid", include: { From da9c8fb1e9cc09505b181b88ddefba1c690ca574 Mon Sep 17 00:00:00 2001 From: "dkoosis@gmail.com" Date: Sun, 2 Aug 2026 11:29:31 -0400 Subject: [PATCH 2/2] fix(providers): rate-limit resilience + non-fatal ingest failures Running a full 500-question LongMemEval ingest surfaced three ways the mem0/supermemory providers and the ingest phase aborted whole runs: - mem0 + supermemory: default ingest concurrency (50/100) blew past each paid tier's per-second cap, triggering an immediate 429 storm. Add exponential-backoff retry (supermemory honors server retryAfterSeconds) and lower the provider-default ingest concurrency to 8. Per-run overrides via checkpoint.concurrency still win over these defaults. - mem0: skip sessions with no usable messages instead of erroring (abstention questions can carry empty haystacks; mem0 rejects empty adds). - ingest phase: a single failed question no longer throws and kills the run. The checkpoint records the failure, downstream phases skip it, and resume retries it. One bad question in 500 must not abort a multi-hour benchmark. --- src/orchestrator/phases/ingest.ts | 10 ++-- src/providers/mem0/index.ts | 47 ++++++++++++++++--- src/providers/supermemory/index.ts | 75 +++++++++++++++++++++--------- 3 files changed, 101 insertions(+), 31 deletions(-) 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 cd595c8..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: options.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 || [] }