diff --git a/apps/memos-local-plugin/core/config/defaults.ts b/apps/memos-local-plugin/core/config/defaults.ts index 5c9dff305..9e2c6b03b 100644 --- a/apps/memos-local-plugin/core/config/defaults.ts +++ b/apps/memos-local-plugin/core/config/defaults.ts @@ -4,6 +4,7 @@ * to change. */ +import { DEFAULT_MAX_INPUT_CHARS } from "../embedding/constants.js"; import type { ResolvedConfig } from "./schema.js"; export const DEFAULT_CONFIG: ResolvedConfig = { @@ -34,6 +35,7 @@ export const DEFAULT_CONFIG: ResolvedConfig = { enabled: true, maxItems: 20_000, }, + maxInputChars: DEFAULT_MAX_INPUT_CHARS, }, llm: { provider: "", diff --git a/apps/memos-local-plugin/core/config/schema.ts b/apps/memos-local-plugin/core/config/schema.ts index 8566f90f3..09719fef9 100644 --- a/apps/memos-local-plugin/core/config/schema.ts +++ b/apps/memos-local-plugin/core/config/schema.ts @@ -10,6 +10,8 @@ import { Type, type Static } from "@sinclair/typebox"; +import { DEFAULT_MAX_INPUT_CHARS } from "../embedding/constants.js"; + // ─── Reusable building blocks ─────────────────────────────────────────────── const StringWithDefault = (def = "") => Type.String({ default: def }); @@ -49,6 +51,18 @@ const EmbeddingSchema = Type.Object({ enabled: Bool(true), maxItems: NumberInRange(20_000, 0), }, { default: {} }), + /** + * Per-input character cap applied inside the `Embedder` facade before + * hashing / calling the provider. Guards against remote embedding + * models with a per-request token limit — most notably 智谱 + * `embedding-3` (3072-token single-input cap; CJK averages ~1.3–1.5 + * chars per token, so the 4000-char default keeps CJK-dominant text + * under the limit), which returns HTTP 400 `code:1210` for + * over-length inputs and used to nuke the whole rebuild batch + * (issue #2121). Set to `0` to disable truncation. The default is + * `DEFAULT_MAX_INPUT_CHARS` in `core/embedding/constants.ts`. + */ + maxInputChars: NumberInRange(DEFAULT_MAX_INPUT_CHARS, 0), }, { default: {} }); const ReasoningSchema = Type.Object({ diff --git a/apps/memos-local-plugin/core/embedding/README.md b/apps/memos-local-plugin/core/embedding/README.md index ab4393395..25df3e485 100644 --- a/apps/memos-local-plugin/core/embedding/README.md +++ b/apps/memos-local-plugin/core/embedding/README.md @@ -157,6 +157,14 @@ Unit tests live in `tests/unit/embedding/`: - `gemini`'s `?key=` puts the secret in the URL; `fetcher.ts` redacts query string via the logger's redaction pipeline. Do not log the raw URL elsewhere. +- `fetcher.ts` attaches a truncated (≤ 512 chars) provider response body to + `http.non_ok` warn logs so operators can see error codes like 智谱's + `code:1210` (issue #2121). This field is a **verbatim third-party + payload**: some providers echo fragments of the submitted embedding input + (i.e. potentially personal memory content) back inside error responses. + Treat any log sink that carries `http.non_ok` (e.g. `gateway.log`) as + potentially containing user data — apply the same retention/access rules + as for raw memory content. - `voyage` and `cohere` charge per token; be mindful when bumping `batchSize` — large batches amortize HTTP overhead but hit TPM ceilings. - Changing `dimensions` in config after writing vectors to SQLite breaks diff --git a/apps/memos-local-plugin/core/embedding/constants.ts b/apps/memos-local-plugin/core/embedding/constants.ts new file mode 100644 index 000000000..9516784a1 --- /dev/null +++ b/apps/memos-local-plugin/core/embedding/constants.ts @@ -0,0 +1,29 @@ +/** + * Shared embedding constants. Kept dependency-free so `core/config/` + * (schema + defaults) can import from here without dragging in the + * provider implementations behind `embedder.ts`. + */ + +/** + * Default per-input character cap for embedding inputs. + * + * Chosen at 4000 chars with 智谱 embedding-3's 3072-token single-input + * hard limit as the reference worst case: GLM tokenizers average + * ~1.3–1.5 chars per token for Chinese, so 4000 CJK chars ≈ 2700–3000 + * tokens — under the cap for typical content (the previous 6000 + * default mapped to ≈ 4000–4600 tokens and could still trip HTTP 400 + * `code:1210` on CJK-dominant inputs). ASCII tokenizes at ~4 + * chars/token, so 4000 chars ≈ 1000 tokens, safe for every supported + * provider. The cap is a guard, not a hard guarantee — pathological + * inputs that still overflow are isolated per-slot by the + * divide-and-conquer retry in `pipeline/memory-core.ts`. + * + * Callers can override via `EmbeddingConfig.maxInputChars`; `0`, a + * negative value, or `Infinity` disables truncation (see + * `resolveMaxInputChars` in `embedder.ts`). See issue #2121. + * + * This constant is the single source of truth — `config/schema.ts` and + * `config/defaults.ts` import it so the schema default, the runtime + * default, and the facade fallback can never drift apart. + */ +export const DEFAULT_MAX_INPUT_CHARS = 4000; diff --git a/apps/memos-local-plugin/core/embedding/embedder.ts b/apps/memos-local-plugin/core/embedding/embedder.ts index 5d035b6ce..1acb0f17b 100644 --- a/apps/memos-local-plugin/core/embedding/embedder.ts +++ b/apps/memos-local-plugin/core/embedding/embedder.ts @@ -25,6 +25,7 @@ import { makeCacheKey, type EmbedCache, } from "./cache.js"; +import { DEFAULT_MAX_INPUT_CHARS } from "./constants.js"; import { postProcess } from "./normalize.js"; import { CohereEmbeddingProvider } from "./providers/cohere.js"; import { GeminiEmbeddingProvider } from "./providers/gemini.js"; @@ -117,7 +118,29 @@ export function createEmbedderWithProvider( requests += inputs.length; if (inputs.length === 0) return []; - const normalized = inputs.map(toInput); + // Per-input character cap. Truncation runs BEFORE cache-key hashing + // so a repeated call with the same head text hits the LRU. Guards + // against provider single-input token caps (e.g. 智谱 embedding-3 + // rejects >3072 tokens with HTTP 400 code:1210 — see issue #2121). + const cap = resolveMaxInputChars(config.maxInputChars); + let truncatedCount = 0; + const normalized = inputs.map(toInput).map((inp) => { + if (cap > 0 && inp.text.length > cap) { + truncatedCount++; + return { ...inp, text: inp.text.slice(0, cap) }; + } + return inp; + }); + if (truncatedCount > 0) { + logger.warn("input_truncated", { + provider: provider.name, + model: config.model, + cap, + count: truncatedCount, + of: normalized.length, + }); + } + const results = new Array(normalized.length).fill(null); const dedupEnabled = config.cache.enabled; const keys = normalized.map((inp, i) => { @@ -346,6 +369,26 @@ export function createEmbedderWithProvider( // ─── Provider lookup ───────────────────────────────────────────────────────── +/** + * Resolve the effective per-input character cap from config. + * + * Semantics (mirrors the `EmbeddingConfig.maxInputChars` JSDoc): + * - `undefined` → `DEFAULT_MAX_INPUT_CHARS` (guard on by default) + * - `NaN` → `DEFAULT_MAX_INPUT_CHARS` (invalid value — e.g. + * a typo'd config — must NOT silently disable the + * guard, or issue #2121 sneaks back in) + * - `0` / negative → `0` (documented explicit opt-out) + * - `Infinity` → `0` ("no cap" — explicit opt-out) + * - any other number → `Math.floor(value)` + */ +export function resolveMaxInputChars(configured: number | undefined): number { + if (configured === undefined || Number.isNaN(configured)) { + return DEFAULT_MAX_INPUT_CHARS; + } + if (configured < 0 || !Number.isFinite(configured)) return 0; + return Math.floor(configured); +} + export function makeProviderFor(name: EmbeddingProviderName): EmbeddingProvider { switch (name) { case "local": diff --git a/apps/memos-local-plugin/core/embedding/fetcher.ts b/apps/memos-local-plugin/core/embedding/fetcher.ts index 303dae28e..b29bf5ded 100644 --- a/apps/memos-local-plugin/core/embedding/fetcher.ts +++ b/apps/memos-local-plugin/core/embedding/fetcher.ts @@ -91,6 +91,8 @@ export async function httpPostJson(opts: HttpPostOpts): Promise< status: resp.status, }); } + // `body` is a verbatim third-party response excerpt. Providers may + // echo submitted memory content, so log sinks must treat it as sensitive. opts.log.warn("http.non_ok", { url: opts.url, status: resp.status, @@ -98,6 +100,7 @@ export async function httpPostJson(opts: HttpPostOpts): Promise< transient, retryAfterMs, durationMs: Date.now() - start, + body: text ? text.slice(0, 512) : undefined, }); if (transient && attempt <= maxRetries) { const plan = planRetry({ diff --git a/apps/memos-local-plugin/core/embedding/index.ts b/apps/memos-local-plugin/core/embedding/index.ts index f6f4b1ed9..040d593ac 100644 --- a/apps/memos-local-plugin/core/embedding/index.ts +++ b/apps/memos-local-plugin/core/embedding/index.ts @@ -6,7 +6,9 @@ export { createEmbedder, createEmbedderWithProvider, makeProviderFor, + resolveMaxInputChars, } from "./embedder.js"; +export { DEFAULT_MAX_INPUT_CHARS } from "./constants.js"; export { LruEmbedCache, NullEmbedCache, diff --git a/apps/memos-local-plugin/core/embedding/types.ts b/apps/memos-local-plugin/core/embedding/types.ts index 95726703c..11e7697b2 100644 --- a/apps/memos-local-plugin/core/embedding/types.ts +++ b/apps/memos-local-plugin/core/embedding/types.ts @@ -44,6 +44,21 @@ export interface EmbeddingConfig { maxRetries?: number; /** Max texts per HTTP round trip. Default: 32. */ batchSize?: number; + /** + * Per-input character cap. Inputs longer than this are truncated + * (character-wise, not token-wise) before being hashed / sent to the + * provider. Guards against provider single-input token caps such as + * 智谱 embedding-3 (3072 tokens; CJK ≈ 1.3–1.5 chars/token). Set `0`, + * a negative value, or `Infinity` to disable; `NaN` (invalid config) + * falls back to the default rather than disabling the guard. + * Default: `DEFAULT_MAX_INPUT_CHARS` (4000) from `constants.ts`. + * + * Truncation happens at the facade boundary so all providers (local, + * openai_compatible, gemini, cohere, voyage, mistral) benefit; the + * cache key is derived from the *truncated* text so repeat calls that + * share the same head text hit the LRU. + */ + maxInputChars?: number; /** Extra headers to tack on outgoing HTTP. */ headers?: Record; /** If true, all output vectors are L2-normalized. Default: true. */ diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index a0354bf64..60580aa3b 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts @@ -4487,28 +4487,10 @@ export function createMemoryCore( let failed = 0; let error: string | undefined; if (batch.length > 0) { - try { - const vecs = await handle.embedder.embedMany( - batch.map((slot) => ({ text: slot.sourceText || "(empty)", role: "document" as const })), - ); - for (let i = 0; i < batch.length; i++) { - const slot = batch[i]!; - const vec = vecs[i]; - if (!vec) { - failed++; - continue; - } - try { - if (slot.update(vec)) updated++; - else failed++; - } catch { - failed++; - } - } - } catch (err) { - failed = batch.length; - error = err instanceof Error ? err.message : String(err); - } + const outcome = await embedAndApplySlots(batch, handle.embedder); + updated = outcome.updated; + failed = outcome.failed; + error = outcome.firstError; } const statsAfter = computeEmbeddingMaintenanceStats(); @@ -4530,6 +4512,108 @@ export function createMemoryCore( }; } + /** + * Divide-and-conquer embed + write for `rebuildEmbeddings`. + * + * Before this refactor `rebuildEmbeddings` blanket-failed a whole + * batch on any provider throw — one 30 KB trace nuked the counters + * for every short trace next to it (issue #2121). Now, on provider + * failure the sub-batch is halved and each half retried; when a + * single-slot sub-batch still fails, only that one slot is counted + * as failed. + * + * Splitting only pays off for *content-specific* failures (one + * poisonous input rejected by the provider). For transient/systemic + * failures (network down, 5xx, 429) the fetcher has already + * exhausted its internal retries before the throw reaches us — + * halving would multiply total provider calls by O(log N) while + * every half fails for the same systemic reason. Those errors + * short-circuit: the whole sub-batch is marked failed in one step + * and the next `rebuildEmbeddings` run retries it. Worst case for + * content errors is O(N log N) round trips; in practice poison + * slots are ≪ 1 %. + */ + async function embedAndApplySlots( + sub: EmbeddingSlot[], + embedder: NonNullable, + ): Promise<{ + updated: number; + failed: number; + firstError?: string; + }> { + if (sub.length === 0) return { updated: 0, failed: 0 }; + try { + const vecs = await embedder.embedMany( + sub.map((slot) => ({ text: slot.sourceText || "(empty)", role: "document" as const })), + ); + let updated = 0; + let failed = 0; + let firstError: string | undefined; + for (let i = 0; i < sub.length; i++) { + const slot = sub[i]!; + const vec = vecs[i]; + if (!vec) { + // Contract violation: embedMany returned fewer vectors than + // inputs. Surface a diagnostic so the operator-facing `error` + // field is not silently empty for this failure mode. + failed++; + firstError ??= `embedMany returned no vector for slot ${slot.id} (index ${i} of ${sub.length})`; + continue; + } + try { + if (slot.update(vec)) updated++; + else failed++; + } catch { + failed++; + } + } + return { updated, failed, firstError }; + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + // Base case: a single-slot sub-batch still failed — mark this + // one slot as failed and bubble the message up. + if (sub.length === 1) return { updated: 0, failed: 1, firstError: msg }; + // Transient/systemic errors fail every half identically — + // splitting would only amplify already-exhausted retries + // (O(N log N × maxRetries) HTTP calls). Fail the sub-batch in + // one step instead; a later run retries it. + if (isTransientEmbeddingError(err)) { + return { updated: 0, failed: sub.length, firstError: msg }; + } + const mid = sub.length >> 1; + const left = await embedAndApplySlots(sub.slice(0, mid), embedder); + const right = await embedAndApplySlots(sub.slice(mid), embedder); + return { + updated: left.updated + right.updated, + failed: left.failed + right.failed, + firstError: left.firstError ?? right.firstError ?? msg, + }; + } + } + + /** + * Positively identify transient/systemic embedding failures so the + * divide-and-conquer in `embedAndApplySlots` can short-circuit + * instead of amplifying retries. Anything we cannot classify with + * confidence is treated as content-specific (split) — mis-splitting + * a systemic error costs extra HTTP calls, but mis-short-circuiting + * a content error re-nukes whole batches, which is the very bug + * (#2121) the splitting exists to fix. + * + * Classification sources (see `core/embedding/fetcher.ts`): + * - `details.status` 429 / 5xx → transient (retries already + * exhausted inside the fetcher before the throw). + * - No status + fetcher's stable network / retry-exhaustion + * message prefixes → transient. + * - 4xx statuses (e.g. 智谱 400 `code:1210`) → content-specific. + */ + function isTransientEmbeddingError(err: unknown): boolean { + if (!(err instanceof MemosError)) return false; + const status = (err.details as { status?: unknown } | undefined)?.status; + if (typeof status === "number") return status === 429 || status >= 500; + return /^Network error calling |^Exhausted retries to /.test(err.message); + } + type EmbeddingSlotKind = "trace" | "policy" | "world_model" | "skill"; type EmbeddingSlot = { kind: EmbeddingSlotKind; diff --git a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts index 617bc4a72..275f58b23 100644 --- a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts +++ b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts @@ -1,8 +1,12 @@ import { beforeAll, describe, expect, it } from "vitest"; import { MemosError } from "../../../agent-contract/errors.js"; -import { createEmbedderWithProvider } from "../../../core/embedding/embedder.js"; -import { initTestLogger } from "../../../core/logger/index.js"; +import { DEFAULT_MAX_INPUT_CHARS } from "../../../core/embedding/constants.js"; +import { + createEmbedderWithProvider, + resolveMaxInputChars, +} from "../../../core/embedding/embedder.js"; +import { initTestLogger, memoryBuffer } from "../../../core/logger/index.js"; import type { EmbedRole, EmbeddingConfig, @@ -287,4 +291,99 @@ describe("embedder facade", () => { expect(closed).toBe(1); expect(e.stats().hits).toBe(0); }); + + /** + * Regression: issue #2121. Long trace text (e.g. 30+ KB agent_text) + * caused HTTP 400 code:1210 from 智谱 embedding-3 (3072-token cap). + * Facade must truncate before hashing / calling the provider, so: + * 1. Provider never sees an over-cap input. + * 2. Cache key is derived from the truncated text (so a repeat call + * with the same head text hits the LRU). + */ + describe("input truncation guard (#2121)", () => { + it("truncates inputs above maxInputChars before the provider call", async () => { + const p = new FakeProvider(); + const e = createEmbedderWithProvider(cfg({ maxInputChars: 10 }), p); + const longText = "0123456789ABCDEFGHIJ"; // 20 chars + await e.embedMany([longText]); + expect(p.calls).toHaveLength(1); + expect(p.calls[0]!.texts).toEqual(["0123456789"]); + }); + + it("hits cache when repeat calls share the truncated prefix", async () => { + const p = new FakeProvider(); + const e = createEmbedderWithProvider(cfg({ maxInputChars: 4 }), p); + // Two distinct 10-char inputs with an identical first 4 chars. + await e.embedMany(["headTAILONE"]); + await e.embedMany(["headTAILTWO"]); + // Only the truncated `head` is embedded once; second call is a hit. + const s = e.stats(); + expect(s.roundTrips).toBe(1); + expect(s.hits).toBe(1); + expect(s.misses).toBe(1); + expect(p.calls).toHaveLength(1); + expect(p.calls[0]!.texts).toEqual(["head"]); + }); + + it("does not truncate when maxInputChars is 0 (disabled)", async () => { + const p = new FakeProvider(); + const e = createEmbedderWithProvider(cfg({ maxInputChars: 0 }), p); + const longText = "x".repeat(9000); + await e.embedMany([longText]); + expect(p.calls[0]!.texts[0]!.length).toBe(9000); + }); + + it("defaults maxInputChars to DEFAULT_MAX_INPUT_CHARS when the field is absent", async () => { + const p = new FakeProvider(); + const c = cfg(); + // Explicitly delete maxInputChars so we can prove the default kicks in. + delete (c as { maxInputChars?: number }).maxInputChars; + const e = createEmbedderWithProvider(c, p); + const longText = "y".repeat(DEFAULT_MAX_INPUT_CHARS + 3000); + await e.embedMany([longText]); + expect(p.calls[0]!.texts[0]!.length).toBe(DEFAULT_MAX_INPUT_CHARS); + }); + + /** + * `resolveMaxInputChars` boundary semantics: + * - invalid values (NaN) must fall back to the default — a typo'd + * config must not silently disable the #2121 guard + * - 0 / negative / Infinity are explicit opt-outs → 0 (disabled) + */ + it("resolveMaxInputChars: invalid falls back to default; opt-outs disable", () => { + expect(resolveMaxInputChars(undefined)).toBe(DEFAULT_MAX_INPUT_CHARS); + expect(resolveMaxInputChars(Number.NaN)).toBe(DEFAULT_MAX_INPUT_CHARS); + expect(resolveMaxInputChars(0)).toBe(0); + expect(resolveMaxInputChars(-1)).toBe(0); + expect(resolveMaxInputChars(Number.NEGATIVE_INFINITY)).toBe(0); + expect(resolveMaxInputChars(Number.POSITIVE_INFINITY)).toBe(0); + expect(resolveMaxInputChars(1234.9)).toBe(1234); + }); + + it("emits a warn per embedMany call when truncation fires (batched, not per input)", async () => { + const p = new FakeProvider(); + const e = createEmbedderWithProvider(cfg({ maxInputChars: 3 }), p); + // Snapshot the shared memory-buffer state so we count only the + // warns triggered by THIS embedMany call (other tests in this + // file also fire input_truncated). Note: `initTestLogger` wires + // AppLogSink + ErrorLogSink at both writing to the same memory + // buffer, so each warn shows up TWICE in the buffer (one per + // sink). We only care that our single call added a well-formed + // batched warn, not the exact count. + const before = memoryBuffer() + .tail({ level: "warn", channel: "embedding", limit: 200 }) + .filter((r) => r.msg === "input_truncated").length; + // Three distinct inputs, all over cap — should batch into one warn. + await e.embedMany(["aaaaaa", "bbbbbb", "cccccc"]); + const after = memoryBuffer() + .tail({ level: "warn", channel: "embedding", limit: 200 }) + .filter((r) => r.msg === "input_truncated"); + // Two sinks × 1 emit = 2 records — key property is "not one per + // input" (which would be 3×2=6). + expect(after.length - before).toBeLessThanOrEqual(2); + expect(after.length - before).toBeGreaterThanOrEqual(1); + // `tail` returns newest-first — the just-emitted warn is at index 0. + expect(after[0]!.data).toMatchObject({ count: 3, cap: 3 }); + }); + }); }); diff --git a/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts b/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts index 2811e94ef..a5802c537 100644 --- a/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts +++ b/apps/memos-local-plugin/tests/unit/embedding/fetcher.test.ts @@ -233,4 +233,67 @@ describe("embedding/fetcher", () => { signal: ctrl.signal, }); }); + + /** + * Regression: issue #2121. The gateway.log line + * "embedding_unavailable: HTTP 400 from openai_compatible" carried no + * response body, so operators could not see 智谱's code:1210 message. + * Truncated body is now attached to the warn detail (first 512 chars). + */ + it("http.non_ok warn detail carries a truncated response body", async () => { + mockFetch([ + new Response( + '{"error":{"code":"1210","message":"API 调用参数有误"}}', + { status: 400 }, + ), + ]); + const warns: Array<{ msg: string; detail?: Record }> = []; + const log: ProviderLogger = { + trace: () => {}, + debug: () => {}, + info: () => {}, + warn: (msg, detail) => warns.push({ msg, detail }), + error: () => {}, + }; + await expect( + httpPostJson({ + url: "https://x", + body: {}, + provider: "openai_compatible", + log, + maxRetries: 0, + }), + ).rejects.toBeInstanceOf(MemosError); + const nonOk = warns.find((w) => w.msg === "http.non_ok"); + expect(nonOk, "http.non_ok warn should have been emitted").toBeTruthy(); + expect(nonOk!.detail).toMatchObject({ status: 400 }); + expect(typeof nonOk!.detail!.body).toBe("string"); + expect(nonOk!.detail!.body).toContain("code"); + expect(nonOk!.detail!.body).toContain("1210"); + }); + + it("http.non_ok body is truncated to at most 512 chars", async () => { + const bigBody = "e".repeat(2000); + mockFetch([new Response(bigBody, { status: 400 })]); + const warns: Array<{ msg: string; detail?: Record }> = []; + const log: ProviderLogger = { + trace: () => {}, + debug: () => {}, + info: () => {}, + warn: (msg, detail) => warns.push({ msg, detail }), + error: () => {}, + }; + await expect( + httpPostJson({ + url: "https://x", + body: {}, + provider: "openai_compatible", + log, + maxRetries: 0, + }), + ).rejects.toBeInstanceOf(MemosError); + const nonOk = warns.find((w) => w.msg === "http.non_ok"); + expect(nonOk).toBeTruthy(); + expect((nonOk!.detail!.body as string).length).toBeLessThanOrEqual(512); + }); }); diff --git a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts index 52853ff4a..d69bfb7ec 100644 --- a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts +++ b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts @@ -30,7 +30,7 @@ import { RECOVERY_REASONS } from "../../../core/pipeline/recovery-constants.js"; import { makeTmpDb, type TmpDbHandle } from "../../helpers/tmp-db.js"; import { makeTmpHome, type TmpHomeContext } from "../../helpers/tmp-home.js"; import { fakeEmbedder } from "../../helpers/fake-embedder.js"; -import type { MemosError } from "../../../agent-contract/errors.js"; +import { MemosError } from "../../../agent-contract/errors.js"; import type { SkillId, SkillRow, TraceRow } from "../../../core/types.js"; let db: TmpDbHandle | null = null; @@ -280,6 +280,220 @@ describe("MemoryCore façade", () => { expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS); }); + /** + * Regression: issue #2121. Before the fix, `rebuildEmbeddings` caught + * any provider throw and set `failed = batch.length` — one 30 KB + * trace nuked the whole batch, so short valid rows next to it were + * counted as failed. The divide-and-conquer retry must isolate the + * poisonous input to only its own slot. + */ + it("rebuildEmbeddings isolates a single-slot provider failure", async () => { + const poison = "POISON"; + // Custom embedder that throws whenever the batch contains `poison`. + const deps = buildDeps(db!); + deps.embedder = { + ...deps.embedder!, + async embedMany(inputs) { + const texts = inputs.map((i) => (typeof i === "string" ? i : i.text)); + if (texts.some((t) => t.includes(poison))) { + throw new Error("boom: over-length input"); + } + // Deterministic non-zero vector. + return texts.map(() => { + const v = new Float32Array(TEST_EMBED_DIMENSIONS); + v[0] = 1; + return v; + }); + }, + async embedOne(input) { + const text = typeof input === "string" ? input : input.text; + if (text.includes(poison)) throw new Error("boom"); + const v = new Float32Array(TEST_EMBED_DIMENSIONS); + v[0] = 1; + return v; + }, + } as typeof deps.embedder; + + pipeline = createPipeline(deps); + core = createMemoryCore( + pipeline, + resolveHome("openclaw", "/tmp/memos-mc-test"), + "test", + ); + await core.init(); + + const rows = [ + { id: "tr_ok_1", summary: "clean summary one" }, + { id: "tr_ok_2", summary: "clean summary two" }, + { id: "tr_bad", summary: `${poison} tail text` }, + { id: "tr_ok_3", summary: "clean summary three" }, + { id: "tr_ok_4", summary: "clean summary four" }, + ]; + await core.importBundle({ + version: 1, + traces: rows.map((r, i) => ({ + id: r.id, + episodeId: `ep_iso_${i}`, + sessionId: `se_iso_${i}`, + ts: 1_700_000_000_000 + i, + userText: `user text ${i}`, + agentText: `agent text ${i}`, + summary: r.summary, + toolCalls: [], + value: 0, + alpha: 0, + priority: 0, + turnId: 1_700_000_000_000 + i, + })), + }); + + const before = await core.embeddingMaintenanceStats(); + expect(before.byKind.trace.missing).toBe(rows.length * 2); + + const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 }); + // Every row has 2 slots (summary + action). Only the poison row's + // `vec_summary` slot must fail. `vec_action` is derived from + // `agentText` which contains no poison, so it must still succeed. + // Expected: 9 updated, 1 failed. + expect(result.processed).toBe(rows.length * 2); + expect(result.updated).toBe(rows.length * 2 - 1); + expect(result.failed).toBe(1); + expect(result.error).toMatch(/boom/); + + // The clean rows must actually have vectors written. + for (const r of rows) { + if (r.id === "tr_bad") continue; + const row = db!.repos.traces.getById(r.id as never); + expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS); + expect(row?.vecAction?.length).toBe(TEST_EMBED_DIMENSIONS); + } + // The poison row: `vec_summary` never applied, `vec_action` should + // still be populated (agentText has no poison). + const badRow = db!.repos.traces.getById("tr_bad" as never); + expect(badRow?.vecSummary).toBeNull(); + expect(badRow?.vecAction?.length).toBe(TEST_EMBED_DIMENSIONS); + }); + + /** + * Companion to the isolation test above: transient/systemic provider + * failures (network down, 5xx, 429 — thrown by the fetcher as + * `MemosError` AFTER its internal retries are exhausted) must NOT + * trigger the divide-and-conquer split. Splitting a systemic failure + * multiplies provider calls by O(log N) with zero chance of success. + * The whole batch fails in a single call; the next run retries it. + */ + it("rebuildEmbeddings does not split the batch on transient provider errors", async () => { + let embedCalls = 0; + const deps = buildDeps(db!); + deps.embedder = { + ...deps.embedder!, + async embedMany() { + embedCalls++; + // Shape thrown by `httpPostJson` for a 503 after retry exhaustion. + throw new MemosError( + "embedding_unavailable", + "HTTP 503 from openai_compatible", + { provider: "openai_compatible", url: "https://x", status: 503 }, + ); + }, + } as typeof deps.embedder; + + pipeline = createPipeline(deps); + core = createMemoryCore( + pipeline, + resolveHome("openclaw", "/tmp/memos-mc-test"), + "test", + ); + await core.init(); + + await core.importBundle({ + version: 1, + traces: Array.from({ length: 4 }, (_, i) => ({ + id: `tr_tra_${i}`, + episodeId: `ep_tra_${i}`, + sessionId: `se_tra_${i}`, + ts: 1_700_000_000_000 + i, + userText: `user text ${i}`, + agentText: `agent text ${i}`, + summary: `summary ${i}`, + toolCalls: [], + value: 0, + alpha: 0, + priority: 0, + turnId: 1_700_000_000_000 + i, + })), + }); + + const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 }); + // 4 rows × 2 slots — all failed, in exactly ONE embedMany call + // (no halving cascade: 8 slots would otherwise cost up to 15 calls). + expect(result.updated).toBe(0); + expect(result.failed).toBe(8); + expect(result.error).toMatch(/HTTP 503/); + expect(embedCalls).toBe(1); + }); + + /** + * Contract-violation diagnostics: when `embedMany` resolves with + * fewer vectors than inputs, the missing slots must fail WITH a + * `firstError` describing which slot got no vector — previously this + * mode was silent and indistinguishable from an update failure. + */ + it("rebuildEmbeddings surfaces an error when embedMany returns short", async () => { + const deps = buildDeps(db!); + // Phase flag: import-time embedding must fail wholesale so both + // slots stay missing; only the rebuild call returns short. + let shortMode = false; + deps.embedder = { + ...deps.embedder!, + async embedMany(inputs) { + if (!shortMode) throw new Error("import-phase embedding disabled"); + // Drop the last vector — a contract violation. + return inputs.slice(0, -1).map(() => { + const v = new Float32Array(TEST_EMBED_DIMENSIONS); + v[0] = 1; + return v; + }); + }, + } as typeof deps.embedder; + + pipeline = createPipeline(deps); + core = createMemoryCore( + pipeline, + resolveHome("openclaw", "/tmp/memos-mc-test"), + "test", + ); + await core.init(); + + await core.importBundle({ + version: 1, + traces: [{ + id: "tr_short", + episodeId: "ep_short", + sessionId: "se_short", + ts: 1_700_000_000_000, + userText: "user text long enough to embed", + agentText: "agent text long enough to embed", + summary: "summary of the short-return regression trace", + toolCalls: [], + value: 0, + alpha: 0, + priority: 0, + turnId: 1_700_000_000_000, + }], + }); + + const before = await core.embeddingMaintenanceStats(); + expect(before.byKind.trace.missing).toBe(2); + + shortMode = true; + const result = await core.rebuildEmbeddings({ mode: "repair", limit: 100 }); + // 1 row × 2 slots; the dropped vector fails its slot with a message. + expect(result.updated).toBe(1); + expect(result.failed).toBe(1); + expect(result.error).toMatch(/returned no vector for slot/); + }); + it("does not require action vectors for lightweight memory traces", async () => { pipeline = createPipeline(buildDeps(db!, configWithLightweightMemory(true))); core = createMemoryCore(