diff --git a/apps/memos-local-plugin/adapters/openclaw/runtime-lock.ts b/apps/memos-local-plugin/adapters/openclaw/runtime-lock.ts index e3cdbd3ea..26f23394e 100644 --- a/apps/memos-local-plugin/adapters/openclaw/runtime-lock.ts +++ b/apps/memos-local-plugin/adapters/openclaw/runtime-lock.ts @@ -7,6 +7,7 @@ import type { ResolvedHome } from "../../core/config/index.js"; const LOCK_DIRNAME = "openclaw-runtime.lock"; const OWNER_FILENAME = "owner.json"; const UNWRITTEN_OWNER_STALE_MS = 30_000; +const heldRuntimeLockDirs = new Set(); export interface OpenClawRuntimeLockOwner { pluginId: string; @@ -101,7 +102,12 @@ export function acquireOpenClawRuntimeLock( if (e.code !== "EEXIST") throw err; const owner = readOwner(ownerFile); - if (owner && pidIsAlive(owner.pid)) { + if (heldRuntimeLockDirs.has(lockDir)) { + throw new DuplicateOpenClawRuntimeError(lockDir, owner); + } + // A live self PID without a locally held lock can only be a stale owner + // from an earlier process lifecycle whose PID was recycled by the OS. + if (owner && owner.pid !== pid && pidIsAlive(owner.pid)) { throw new DuplicateOpenClawRuntimeError(lockDir, owner); } if (!owner && !lockLooksStale(lockDir, now(), unwrittenOwnerStaleMs)) { @@ -128,11 +134,13 @@ export function acquireOpenClawRuntimeLock( fs.rmSync(lockDir, { recursive: true, force: true }); throw err; } + heldRuntimeLockDirs.add(lockDir); let released = false; const releaseSync = () => { if (released) return; released = true; + heldRuntimeLockDirs.delete(lockDir); const current = readOwner(ownerFile); if (current?.token !== owner.token) return; fs.rmSync(lockDir, { recursive: true, force: true }); diff --git a/apps/memos-local-plugin/core/capture/embedder.ts b/apps/memos-local-plugin/core/capture/embedder.ts index 65fa71cfb..be1deff47 100644 --- a/apps/memos-local-plugin/core/capture/embedder.ts +++ b/apps/memos-local-plugin/core/capture/embedder.ts @@ -11,7 +11,7 @@ */ import { MemosError } from "../../agent-contract/errors.js"; -import type { Embedder } from "../embedding/index.js"; +import type { Embedder, EmbeddingSettledResult } from "../embedding/index.js"; import { rootLogger } from "../logger/index.js"; import type { EmbeddingVector } from "../types.js"; import type { NormalizedStep } from "./types.js"; @@ -38,6 +38,20 @@ export async function embedSteps( const log = rootLogger.child({ channel: "core.capture.embed" }); if (steps.length === 0) return []; + const warnPartialFailures = ( + settled: readonly EmbeddingSettledResult[], + inputCount: number, + ): void => { + let failedCount = 0; + for (let i = 0; i < inputCount; i++) { + if (!settled[i]?.ok) failedCount++; + } + if (failedCount > 0) { + const event = failedCount === inputCount ? "embed.failed_all" : "embed.partial_failed"; + log.warn(event, { failedCount, inputCount, stepCount: steps.length }); + } + }; + const summaryTexts = steps.map((s, i) => { const override = summaryOverrides?.[i]?.trim(); if (override) return override; @@ -46,9 +60,19 @@ export async function embedSteps( const actionTexts = steps.map(actionText); if (opts.summaryOnly) { try { - const vecs = await embedder.embedMany( - summaryTexts.map((t) => ({ text: t || "(empty)", role: "document" as const })), - ); + const inputs = summaryTexts.map((t) => ({ + text: t || "(empty)", + role: "document" as const, + })); + if (embedder.embedManySettled) { + const settled = await embedder.embedManySettled(inputs); + warnPartialFailures(settled, inputs.length); + return steps.map((_, i) => ({ + summary: settled[i]?.ok ? settled[i].vector : null, + action: null, + })); + } + const vecs = await embedder.embedMany(inputs); return steps.map((_, i) => ({ summary: vecs[i] ?? null, action: null })); } catch (err) { log.warn("embed.failed_all", { err: errDetail(err), stepCount: steps.length }); @@ -63,6 +87,20 @@ export async function embedSteps( ]; try { + if (embedder.embedManySettled) { + const settled = await embedder.embedManySettled(inputs); + warnPartialFailures(settled, inputs.length); + const out: VecPair[] = new Array(steps.length); + for (let i = 0; i < steps.length; i++) { + const summary = settled[i]; + const action = settled[i + steps.length]; + out[i] = { + summary: summary?.ok ? summary.vector : null, + action: action?.ok ? action.vector : null, + }; + } + return out; + } const vecs = await embedder.embedMany(inputs); const out: VecPair[] = new Array(steps.length); for (let i = 0; i < steps.length; i++) { diff --git a/apps/memos-local-plugin/core/config/defaults.ts b/apps/memos-local-plugin/core/config/defaults.ts index 6f06210d0..9360b8802 100644 --- a/apps/memos-local-plugin/core/config/defaults.ts +++ b/apps/memos-local-plugin/core/config/defaults.ts @@ -40,6 +40,8 @@ export const DEFAULT_CONFIG: ResolvedConfig = { providerIgnore: [], providerOrder: [], openRouter: false, + maxInputTokens: 1_024, + batchSize: 32, cache: { enabled: true, maxItems: 20_000, diff --git a/apps/memos-local-plugin/core/config/index.ts b/apps/memos-local-plugin/core/config/index.ts index 6d529d960..9e7ccfd84 100644 --- a/apps/memos-local-plugin/core/config/index.ts +++ b/apps/memos-local-plugin/core/config/index.ts @@ -65,6 +65,11 @@ export async function loadConfig(home: ResolvedHome, agent?: string): Promise { return typeof v === "object" && v !== null && !Array.isArray(v); } +function withLegacyEmbeddingInputLimit(raw: unknown): unknown { + if (!isPlainObject(raw)) return raw; + const embedding = isPlainObject(raw.embedding) ? raw.embedding : {}; + if (Object.hasOwn(embedding, "maxInputTokens")) return raw; + return { + ...raw, + embedding: { + ...embedding, + maxInputTokens: 0, + }, + }; +} + function stripUnsupportedEmbeddingDimensions(merged: Record): void { const embedding = merged.embedding; if (!isPlainObject(embedding)) return; diff --git a/apps/memos-local-plugin/core/config/schema.ts b/apps/memos-local-plugin/core/config/schema.ts index 8566f90f3..56bf27fd8 100644 --- a/apps/memos-local-plugin/core/config/schema.ts +++ b/apps/memos-local-plugin/core/config/schema.ts @@ -14,8 +14,13 @@ import { Type, type Static } from "@sinclair/typebox"; const StringWithDefault = (def = "") => Type.String({ default: def }); const Bool = (def: boolean) => Type.Boolean({ default: def }); -const NumberInRange = (def: number, min?: number, max?: number) => - Type.Number({ default: def, ...(min != null ? { minimum: min } : {}), ...(max != null ? { maximum: max } : {}) }); +const NumberInRange = (def: number, min?: number, max?: number, description?: string) => + Type.Number({ + default: def, + ...(min != null ? { minimum: min } : {}), + ...(max != null ? { maximum: max } : {}), + ...(description ? { description } : {}), + }); // ─── Sub-schemas ──────────────────────────────────────────────────────────── @@ -45,6 +50,18 @@ const EmbeddingSchema = Type.Object({ providerOrder: Type.Optional(Type.Array(Type.String(), { default: [] })), /** Explicitly enable OpenRouter fields for a reverse proxy or CNAME. */ openRouter: Type.Optional(Bool(false)), + /** + * Maximum estimated tokens in one provider input. `0` explicitly disables + * client-side chunking. New installations default to a conservative 1024. + */ + maxInputTokens: NumberInRange( + 1_024, + 0, + 1_000_000, + "Maximum estimated tokens per embedding input. Set to 0 to disable client-side chunking.", + ), + /** Maximum physical texts sent in one embedding-provider HTTP request. */ + batchSize: NumberInRange(32, 1, 256), cache: Type.Object({ enabled: Bool(true), maxItems: NumberInRange(20_000, 0), diff --git a/apps/memos-local-plugin/core/config/writer.ts b/apps/memos-local-plugin/core/config/writer.ts index 34f431f70..36a2acdae 100644 --- a/apps/memos-local-plugin/core/config/writer.ts +++ b/apps/memos-local-plugin/core/config/writer.ts @@ -61,6 +61,14 @@ export async function patchConfig( if (!existingText) { const initialPort = effectiveViewerPort(agent); if (initialPort !== undefined) doc.setIn(["viewer", "port"], initialPort); + } else if ( + doc.getIn(["embedding", "maxInputTokens"]) === undefined && + !patchSetsEmbeddingInputLimit(patch) + ) { + // Editing an older config must not silently opt it into the new-install + // 1024 default. Persist its prior disabled behaviour before applying the + // unrelated patch so subsequent loads remain stable. + applyPatch(doc, { embedding: { maxInputTokens: 0 } }); } applyPatch(doc, patch); if ( @@ -97,6 +105,11 @@ export async function patchConfig( return { config, bytes, source: home.configFile, created }; } +function patchSetsEmbeddingInputLimit(patch: Record): boolean { + const embedding = patch.embedding; + return isPlainObject(embedding) && Object.hasOwn(embedding, "maxInputTokens"); +} + /** * Walk the patch object and apply each leaf to the YAML Document. Deep keys * are created as needed; arrays are replaced wholesale. Comments on existing diff --git a/apps/memos-local-plugin/core/embedding/README.md b/apps/memos-local-plugin/core/embedding/README.md index ab4393395..4b3364fef 100644 --- a/apps/memos-local-plugin/core/embedding/README.md +++ b/apps/memos-local-plugin/core/embedding/README.md @@ -47,15 +47,15 @@ Return type: `Float32Array` of length `config.embedding.dimensions`. inputs ─▶ normalize(input list → role-tagged {text}) │ ▼ - sha256(provider|model|role|text) → cache lookup (LRU) - │ ├── hit ──────────────────────────┐ - ▼ └── miss → batched by role │ - batch k texts ──▶ provider.embed() │ - │ │ │ - ▼ ▼ ▼ - dim-enforce + L2-normalize (Float32Array) ──────▶ interleave in input order + sha256(provider|model|role|full text) → cache lookup (LRU) + │ ├── hit ────────────────────────────┐ + ▼ └── miss → token estimate + sampling │ + physical chunks ──▶ role grouping + provider batchSize │ + │ │ │ + ▼ ▼ ▼ + dim-enforce + normalize → pool chunks ────────▶ logical input order │ - └── cache.set(key, vec) + └── cache.set(full-text key, vec) ``` The cache is indexed by `sha256(provider|model|role|text)` in hex. Duplicate @@ -69,6 +69,19 @@ are grouped by `role` first, then chunked into batches. That way a mixed list (some `query`, some `document`) still yields two role-correct round trips instead of one role-ambiguous one. +`maxInputTokens` (default 1024; set 0 to disable) is an operator-supplied +per-input model limit for providers that do not expose capability metadata +(especially custom `openai_compatible` endpoints). When enabled, the facade uses a conservative +dependency-free token estimate, keeps a safety margin, and represents an +over-limit logical input with at most four uniformly sampled chunks. Chunk +vectors are token-weighted and pooled back to the one-vector public contract. + +HTTP 400/413/422 failures on a multi-input provider request trigger bounded +divide-and-conquer splitting. `embedManySettled` exposes the resulting +per-logical-input success/error values so a rejected input does not discard +valid neighbours; authentication, rate-limit, server, and network failures are +not recursively split. + ### 2.2 Normalization Providers return raw float arrays of *their* dimensionality. We enforce the diff --git a/apps/memos-local-plugin/core/embedding/embedder.ts b/apps/memos-local-plugin/core/embedding/embedder.ts index 5d035b6ce..42367d1fb 100644 --- a/apps/memos-local-plugin/core/embedding/embedder.ts +++ b/apps/memos-local-plugin/core/embedding/embedder.ts @@ -25,7 +25,7 @@ import { makeCacheKey, type EmbedCache, } from "./cache.js"; -import { postProcess } from "./normalize.js"; +import { l2Normalize, postProcess } from "./normalize.js"; import { CohereEmbeddingProvider } from "./providers/cohere.js"; import { GeminiEmbeddingProvider } from "./providers/gemini.js"; import { LocalEmbeddingProvider } from "./providers/local.js"; @@ -41,6 +41,7 @@ import type { EmbeddingConfig, EmbeddingProvider, EmbeddingProviderName, + EmbeddingSettledResult, ProviderCallCtx, ProviderLogger, } from "./types.js"; @@ -106,19 +107,33 @@ export function createEmbedderWithProvider( input: string | EmbedInput, options?: EmbedCallOptions, ): Promise { - const vecs = await embedMany([input], options); - return vecs[0]!; + const result = (await embedManySettled([input], options))[0]!; + if (!result.ok) throw result.error; + return result.vector; } async function embedMany( inputs: Array, options?: EmbedCallOptions, ): Promise { + const settled = await embedManySettled(inputs, options); + const failed = settled.find((result) => !result.ok); + if (failed && !failed.ok) throw failed.error; + return settled.map((result) => { + if (!result.ok) throw result.error; + return result.vector; + }); + } + + async function embedManySettled( + inputs: Array, + options?: EmbedCallOptions, + ): Promise { requests += inputs.length; if (inputs.length === 0) return []; const normalized = inputs.map(toInput); - const results = new Array(normalized.length).fill(null); + const results = new Array(normalized.length).fill(null); const dedupEnabled = config.cache.enabled; const keys = normalized.map((inp, i) => { const base = makeCacheKey({ @@ -141,7 +156,7 @@ export function createEmbedderWithProvider( const key = keys[i]!; const cached = cache.get(key); if (cached !== undefined) { - results[i] = cached; + results[i] = { ok: true, vector: cached }; hits++; continue; } @@ -160,137 +175,85 @@ export function createEmbedderWithProvider( if (missByKey.size === 0) { cacheLog.trace("all-hit", { n: inputs.length }); - return results as EmbeddingVector[]; + return results as EmbeddingSettledResult[]; } const missEntries = Array.from(missByKey.entries()); const batchSize = Math.max(1, config.batchSize ?? 32); + type LogicalWork = { + key: string; + role: EmbedRole; + indices: number[]; + chunks: string[]; + chunkVectors: Array; + error: MemosError | null; + }; + type PhysicalWork = { + logical: LogicalWork; + chunkIndex: number; + text: string; + }; + + const logicalWorks: LogicalWork[] = missEntries.map(([key, entry]) => { + const chunks = splitEmbeddingInput(entry.text, config.maxInputTokens ?? 1_024); + if (chunks.length > 1) { + logger.warn("input.chunked", { + provider: provider.name, + model: config.model, + estimatedTokens: estimateEmbeddingTokens(entry.text), + maxInputTokens: config.maxInputTokens, + chunks: chunks.length, + }); + } + return { + key, + role: entry.role, + indices: entry.indices, + chunks, + chunkVectors: new Array(chunks.length).fill(null), + error: null, + }; + }); // Preserve role grouping — provider semantics (e.g. cohere query vs doc) // differ per role so we batch per (role) within each round trip. - const byRole = new Map< - EmbedRole, - Array<{ key: string; text: string; indices: number[] }> - >(); - for (const [key, entry] of missEntries) { - const list = byRole.get(entry.role) ?? []; - list.push({ key, text: entry.text, indices: entry.indices }); - byRole.set(entry.role, list); + const byRole = new Map(); + for (const logical of logicalWorks) { + const list = byRole.get(logical.role) ?? []; + for (let chunkIndex = 0; chunkIndex < logical.chunks.length; chunkIndex++) { + list.push({ + logical, + chunkIndex, + text: logical.chunks[chunkIndex]!, + }); + } + byRole.set(logical.role, list); } for (const [role, list] of byRole.entries()) { for (let start = 0; start < list.length; start += batchSize) { const slice = list.slice(start, start + batchSize); - const texts = slice.map((s) => s.text); - roundTrips++; - let raw: number[][]; - const startedAt = Date.now(); - try { - const ctx: ProviderCallCtx = { - config, - log: providerCtxLog, - signal: options?.signal, - deadlineAt: options?.deadlineAt, - }; - raw = await provider.embed(texts, role, ctx); - // Record success but DO NOT clear `lastError` — the viewer - // compares `lastError.at` against `lastOkAt` to decide the - // overview card colour. Clearing here would let one cache- - // friendly success silently mask a still-real provider - // outage that just produced a `system_error` log row. - lastOkAt = Date.now(); - notifyStatus({ - status: "ok", - provider: provider.name, - model: config.model, - at: lastOkAt, - durationMs: lastOkAt - startedAt, - }); - } catch (err) { - failures++; - const errAt = Date.now(); - const errMessage = - err instanceof MemosError - ? `${err.code}: ${err.message}` - : err instanceof Error - ? err.message - : String(err); - lastError = { at: errAt, message: errMessage }; - logger.warn("provider.failed", { - provider: provider.name, - model: config.model, - role, - count: texts.length, - err: toErrDetail(err), - }); - // Notify the bootstrap-supplied error sink (if any). Wrapped in - // its own try/catch so a buggy sink never masks the original - // failure for the caller. - if (config.onError) { - try { - config.onError({ - kind: "embedding", - provider: provider.name, - model: config.model, - message: errMessage, - code: err instanceof MemosError ? err.code : undefined, - at: errAt, - ...extractRetryDiagnostics(err instanceof MemosError ? err.details : undefined), - }); - } catch { - /* sink errors are non-fatal */ - } - } - notifyStatus({ - status: "error", - provider: provider.name, - model: config.model, - message: errMessage, - code: err instanceof MemosError ? err.code : undefined, - at: errAt, - durationMs: errAt - startedAt, - ...extractRetryDiagnostics(err instanceof MemosError ? err.details : undefined), - }); - throw err instanceof MemosError - ? err - : new MemosError( - ERROR_CODES.EMBEDDING_UNAVAILABLE, - `${provider.name} failed: ${(err as Error).message ?? String(err)}`, - { provider: provider.name }, - ); - } - if (raw.length !== texts.length) { - throw new MemosError( - ERROR_CODES.EMBEDDING_UNAVAILABLE, - `${provider.name} returned ${raw.length} vectors for ${texts.length} inputs`, - { provider: provider.name }, - ); - } - const normalize = config.normalize ?? true; - const processed = postProcess(raw, { - dimensions: actualDimensions, - provider: provider.name, - model: config.model, - normalize, - }); - if (actualDimensions <= 0 && processed[0]) { - actualDimensions = processed[0].length; - logger.info("dimensions.inferred", { - provider: provider.name, - model: config.model, - dimensions: actualDimensions, - }); - } + const physicalResults = await embedPhysicalBatch(slice, role, options); for (let j = 0; j < slice.length; j++) { - const vec = processed[j]!; const entry = slice[j]!; - cache.set(entry.key, vec); - for (const idx of entry.indices) results[idx] = vec; + const result = physicalResults[j]!; + if (result.ok) entry.logical.chunkVectors[entry.chunkIndex] = result.vector; + else entry.logical.error ??= result.error; } } } - // Final assertion — everything should be filled by now. + for (const logical of logicalWorks) { + const result: EmbeddingSettledResult = logical.error + ? { ok: false, error: logical.error } + : { + ok: true, + vector: poolChunkVectors(logical.chunks, logical.chunkVectors, config.normalize ?? true), + }; + if (result.ok) cache.set(logical.key, result.vector); + for (const idx of logical.indices) results[idx] = result; + } + for (let i = 0; i < results.length; i++) { if (results[i] === null) { throw new MemosError( @@ -300,7 +263,119 @@ export function createEmbedderWithProvider( ); } } - return results as EmbeddingVector[]; + return results as EmbeddingSettledResult[]; + } + + async function embedPhysicalBatch( + entries: Array<{ text: string }>, + role: EmbedRole, + options?: EmbedCallOptions, + ): Promise { + const texts = entries.map((entry) => entry.text); + roundTrips++; + const startedAt = Date.now(); + try { + const ctx: ProviderCallCtx = { + config, + log: providerCtxLog, + signal: options?.signal, + deadlineAt: options?.deadlineAt, + }; + const raw = await provider.embed(texts, role, ctx); + if (raw.length !== texts.length) { + throw new MemosError( + ERROR_CODES.EMBEDDING_UNAVAILABLE, + `${provider.name} returned ${raw.length} vectors for ${texts.length} inputs`, + { provider: provider.name, reason: "response_count_mismatch" }, + ); + } + const processed = postProcess(raw, { + dimensions: actualDimensions, + provider: provider.name, + model: config.model, + normalize: config.normalize ?? true, + }); + if (actualDimensions <= 0 && processed[0]) { + actualDimensions = processed[0].length; + logger.info("dimensions.inferred", { + provider: provider.name, + model: config.model, + dimensions: actualDimensions, + }); + } + // Record success but DO NOT clear `lastError` — consumers compare + // timestamps to determine whether the latest provider event recovered. + lastOkAt = Date.now(); + notifyStatus({ + status: "ok", + provider: provider.name, + model: config.model, + at: lastOkAt, + durationMs: lastOkAt - startedAt, + }); + return processed.map((vector) => ({ ok: true, vector })); + } catch (err) { + failures++; + const wrapped = asEmbeddingError(err, provider.name); + if (entries.length > 1 && shouldSplitProviderBatch(wrapped)) { + const mid = entries.length >> 1; + logger.warn("provider.batch_split", { + provider: provider.name, + model: config.model, + role, + count: entries.length, + status: providerStatus(wrapped), + }); + const left = await embedPhysicalBatch(entries.slice(0, mid), role, options); + const right = await embedPhysicalBatch(entries.slice(mid), role, options); + return [...left, ...right]; + } + recordTerminalProviderFailure(wrapped, role, texts.length, startedAt); + return entries.map(() => ({ ok: false, error: wrapped })); + } + } + + function recordTerminalProviderFailure( + err: MemosError, + role: EmbedRole, + count: number, + startedAt: number, + ): void { + const errAt = Date.now(); + const errMessage = `${err.code}: ${err.message}`; + lastError = { at: errAt, message: errMessage }; + logger.warn("provider.failed", { + provider: provider.name, + model: config.model, + role, + count, + err: toErrDetail(err), + }); + if (config.onError) { + try { + config.onError({ + kind: "embedding", + provider: provider.name, + model: config.model, + message: errMessage, + code: err.code, + at: errAt, + ...extractRetryDiagnostics(err.details), + }); + } catch { + /* sink errors are non-fatal */ + } + } + notifyStatus({ + status: "error", + provider: provider.name, + model: config.model, + message: errMessage, + code: err.code, + at: errAt, + durationMs: errAt - startedAt, + ...extractRetryDiagnostics(err.details), + }); } const api: Embedder = { @@ -311,6 +386,7 @@ export function createEmbedderWithProvider( }, embedOne, embedMany, + embedManySettled, stats(): EmbedStats { return { hits, misses, requests, roundTrips, failures, lastOkAt, lastError }; }, @@ -339,11 +415,116 @@ export function createEmbedderWithProvider( dimensions: actualDimensions > 0 ? actualDimensions : "auto", cacheEnabled: config.cache.enabled, batchSize: config.batchSize ?? 32, + maxInputTokens: config.maxInputTokens ?? 1_024, }); return api; } +const MAX_CHUNKS_PER_LOGICAL_INPUT = 4; +const INPUT_TOKEN_SAFETY_RATIO = 0.9; + +/** + * Dependency-free token estimate for providers that do not expose a tokenizer. + * It intentionally leans conservative for CJK, emoji, code, and punctuation; + * the configured value remains a provider limit rather than a character cap. + */ +export function estimateEmbeddingTokens(text: string): number { + let estimate = 0; + for (const char of text) estimate += estimatedTokenWeight(char); + return Math.ceil(estimate); +} + +function estimatedTokenWeight(char: string): number { + const codePoint = char.codePointAt(0) ?? 0; + if (codePoint > 0xffff) return 2; + if (codePoint > 0x7f) return 1; + if (/\s/.test(char)) return 0.25; + if (/[A-Za-z0-9]/.test(char)) return 0.5; + return 1; +} + +function splitEmbeddingInput(text: string, configuredLimit: number): string[] { + if (!Number.isFinite(configuredLimit) || configuredLimit <= 0) return [text]; + const maxTokens = Math.max(1, Math.floor(configuredLimit * INPUT_TOKEN_SAFETY_RATIO)); + if (estimateEmbeddingTokens(text) <= maxTokens) return [text]; + + const chunks: string[] = []; + let current = ""; + let currentTokens = 0; + for (const char of text) { + const weight = estimatedTokenWeight(char); + if (current && currentTokens + weight > maxTokens) { + chunks.push(current); + current = ""; + currentTokens = 0; + } + current += char; + currentTokens += weight; + } + if (current || chunks.length === 0) chunks.push(current); + if (chunks.length <= MAX_CHUNKS_PER_LOGICAL_INPUT) return chunks; + + // Bound embedding spend for imported transcripts that may be tens of + // thousands of characters. Uniform sampling retains head/middle/tail + // coverage and is deterministic, so the logical cache key stays stable. + const selected: string[] = []; + const seen = new Set(); + for (let i = 0; i < MAX_CHUNKS_PER_LOGICAL_INPUT; i++) { + const index = Math.round((i * (chunks.length - 1)) / (MAX_CHUNKS_PER_LOGICAL_INPUT - 1)); + if (!seen.has(index)) { + selected.push(chunks[index]!); + seen.add(index); + } + } + return selected; +} + +function poolChunkVectors( + chunks: string[], + vectors: Array, + normalize: boolean, +): EmbeddingVector { + const first = vectors.find((vector): vector is EmbeddingVector => vector !== null); + if (!first || vectors.some((vector) => vector === null)) { + throw new MemosError( + ERROR_CODES.EMBEDDING_UNAVAILABLE, + "[embedding] internal: missing chunk vector", + ); + } + if (vectors.length === 1) return first; + + const pooled = new Float32Array(first.length); + let totalWeight = 0; + for (let i = 0; i < vectors.length; i++) { + const vector = vectors[i]!; + const weight = Math.max(1, estimateEmbeddingTokens(chunks[i]!)); + totalWeight += weight; + for (let j = 0; j < pooled.length; j++) pooled[j]! += vector[j]! * weight; + } + for (let j = 0; j < pooled.length; j++) pooled[j]! /= totalWeight; + return normalize ? l2Normalize(pooled) : pooled; +} + +function asEmbeddingError(err: unknown, provider: EmbeddingProviderName): MemosError { + if (err instanceof MemosError) return err; + return new MemosError( + ERROR_CODES.EMBEDDING_UNAVAILABLE, + `${provider} failed: ${err instanceof Error ? err.message : String(err)}`, + { provider }, + ); +} + +function providerStatus(err: MemosError): number | null { + const status = (err.details as { status?: unknown } | undefined)?.status; + return typeof status === "number" ? status : null; +} + +function shouldSplitProviderBatch(err: MemosError): boolean { + const status = providerStatus(err); + return status === 400 || status === 413 || status === 422; +} + // ─── Provider lookup ───────────────────────────────────────────────────────── export function makeProviderFor(name: EmbeddingProviderName): EmbeddingProvider { diff --git a/apps/memos-local-plugin/core/embedding/index.ts b/apps/memos-local-plugin/core/embedding/index.ts index f6f4b1ed9..499d66dc3 100644 --- a/apps/memos-local-plugin/core/embedding/index.ts +++ b/apps/memos-local-plugin/core/embedding/index.ts @@ -5,6 +5,7 @@ export { createEmbedder, createEmbedderWithProvider, + estimateEmbeddingTokens, makeProviderFor, } from "./embedder.js"; export { @@ -27,6 +28,7 @@ export type { EmbeddingConfig, EmbeddingProvider, EmbeddingProviderName, + EmbeddingSettledResult, ProviderCallCtx, ProviderLogger, } from "./types.js"; diff --git a/apps/memos-local-plugin/core/embedding/types.ts b/apps/memos-local-plugin/core/embedding/types.ts index 95726703c..9e190ea58 100644 --- a/apps/memos-local-plugin/core/embedding/types.ts +++ b/apps/memos-local-plugin/core/embedding/types.ts @@ -5,6 +5,7 @@ * are never imported directly outside of `core/embedding/`. */ +import type { MemosError } from "../../agent-contract/errors.js"; import type { EmbeddingVector } from "../types.js"; import type { RetryDiagnosticDetails } from "../util/retry-after.js"; @@ -44,6 +45,12 @@ export interface EmbeddingConfig { maxRetries?: number; /** Max texts per HTTP round trip. Default: 32. */ batchSize?: number; + /** + * Maximum estimated tokens per provider input. Longer logical inputs are + * represented by at most four sampled chunks and pooled back to one vector. + * `0` disables client-side chunking. Default: 1024. + */ + maxInputTokens?: number; /** Extra headers to tack on outgoing HTTP. */ headers?: Record; /** If true, all output vectors are L2-normalized. Default: true. */ @@ -180,6 +187,15 @@ export interface Embedder { options?: EmbedCallOptions, ): Promise; + /** + * Batch-embed without allowing one rejected input to discard successful + * neighbours. Implementations predating this method may omit it. + */ + embedManySettled?( + inputs: Array, + options?: EmbedCallOptions, + ): Promise; + stats(): EmbedStats; resetCache(): void; @@ -187,6 +203,10 @@ export interface Embedder { close(): Promise; } +export type EmbeddingSettledResult = + | { ok: true; vector: EmbeddingVector } + | { ok: false; error: MemosError }; + export interface EmbedCallOptions { signal?: AbortSignal; /** Absolute end-to-end deadline shared across provider retry attempts. */ diff --git a/apps/memos-local-plugin/core/index.ts b/apps/memos-local-plugin/core/index.ts index c5e4fab9d..884a9b522 100644 --- a/apps/memos-local-plugin/core/index.ts +++ b/apps/memos-local-plugin/core/index.ts @@ -94,6 +94,7 @@ export { export { createEmbedder, createEmbedderWithProvider, + estimateEmbeddingTokens, makeProviderFor, LruEmbedCache, NullEmbedCache, @@ -118,6 +119,7 @@ export { type EmbeddingConfig, type EmbeddingProvider, type EmbeddingProviderName, + type EmbeddingSettledResult, type ProviderCallCtx, type ProviderLogger, } from "./embedding/index.js"; diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index 096e16a1c..4677ece0c 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts @@ -4628,23 +4628,36 @@ export function createMemoryCore( 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 })), - ); + const inputs = batch.map((slot) => ({ + text: slot.sourceText || "(empty)", + role: "document" as const, + })); + const settled = handle.embedder.embedManySettled + ? await handle.embedder.embedManySettled(inputs) + : (await handle.embedder.embedMany(inputs)).map((vector) => ({ + ok: true as const, + vector, + })); + let firstSlotError: string | undefined; for (let i = 0; i < batch.length; i++) { const slot = batch[i]!; - const vec = vecs[i]; - if (!vec) { + const result = settled[i]; + if (!result?.ok) { failed++; + firstSlotError ??= result?.error.message ?? `missing vector for ${slot.id}`; continue; } try { - if (slot.update(vec)) updated++; + if (slot.update(result.vector)) updated++; else failed++; } catch { failed++; } } + // In repair mode successful slots disappear from the next query, so + // partial failures do not block progress. Stop only when a pass makes + // no progress and the remaining slots are terminally rejected. + if (mode === "repair" && updated === 0 && failed > 0) error = firstSlotError; } catch (err) { failed = batch.length; error = err instanceof Error ? err.message : String(err); @@ -4655,7 +4668,7 @@ export function createMemoryCore( const nextOffset = mode === "rebuild" ? offset + batch.length : 0; const done = mode === "rebuild" ? nextOffset >= targetSlots.length || batch.length === 0 - : statsAfter.needsRepair === 0 || batch.length === 0; + : statsAfter.needsRepair === 0 || batch.length === 0 || (updated === 0 && failed > 0); return { mode, processed: batch.length, diff --git a/apps/memos-local-plugin/core/util/foreground-resources.ts b/apps/memos-local-plugin/core/util/foreground-resources.ts index d818d189f..b2518f3ed 100644 --- a/apps/memos-local-plugin/core/util/foreground-resources.ts +++ b/apps/memos-local-plugin/core/util/foreground-resources.ts @@ -1,7 +1,9 @@ +import { ERROR_CODES, MemosError } from "../../agent-contract/errors.js"; import type { EmbedCallOptions, Embedder, EmbedInput, + EmbeddingSettledResult, } from "../embedding/types.js"; import type { EmbeddingVector } from "../types.js"; @@ -255,6 +257,52 @@ export function prioritizeEmbedder( return results; } + async function embedManySettled( + inputs: Array, + options?: EmbedCallOptions, + ): Promise { + const signal = resources.signalFor(options?.signal); + const callOptions = { ...options, signal }; + const run = async (slice: Array): Promise => { + if (inner!.embedManySettled) return await inner!.embedManySettled(slice, callOptions); + try { + return (await inner!.embedMany(slice, callOptions)).map((vector) => ({ + ok: true as const, + vector, + })); + } catch (err) { + const error = err instanceof MemosError + ? err + : new MemosError( + ERROR_CODES.EMBEDDING_UNAVAILABLE, + `legacy embedMany failed: ${err instanceof Error ? err.message : String(err)}`, + ); + return slice.map(() => ({ ok: false as const, error })); + } + }; + if (priority === "foreground" || inputs.length <= backgroundChunkSize) { + if (priority === "background") await resources.waitForBackground(signal); + const release = await resources.acquireEmbedding(priority, signal); + try { + return await run(inputs); + } finally { + release(); + } + } + + const results: EmbeddingSettledResult[] = []; + for (let start = 0; start < inputs.length; start += backgroundChunkSize) { + await resources.waitForBackground(signal); + const release = await resources.acquireEmbedding(priority, signal); + try { + results.push(...await run(inputs.slice(start, start + backgroundChunkSize))); + } finally { + release(); + } + } + return results; + } + return { get dimensions() { return inner.dimensions; @@ -267,6 +315,7 @@ export function prioritizeEmbedder( }, embedOne, embedMany, + embedManySettled, stats: () => inner.stats(), resetCache: () => inner.resetCache(), close: () => inner.close(), diff --git a/apps/memos-local-plugin/install.ps1 b/apps/memos-local-plugin/install.ps1 index 2acfb0713..7afbabb65 100644 --- a/apps/memos-local-plugin/install.ps1 +++ b/apps/memos-local-plugin/install.ps1 @@ -292,8 +292,10 @@ function Prepare-StagedPackage { Push-Location $StagedPrefix try { $env:MEMOS_SKIP_SETUP = "1" + # npm 11 still resolves omitted DSH development peer trees unless + # legacy peer resolution is requested for the packed runtime. Invoke-NativeChecked -Command $NpmCommand -Arguments @( - "install", "--omit=dev", "--no-fund", "--no-audit", "--loglevel=error" + "install", "--omit=dev", "--legacy-peer-deps", "--no-fund", "--no-audit", "--loglevel=error" ) -FailureMessage "npm install failed" } finally { if ($null -eq $PreviousSkipSetup) { diff --git a/apps/memos-local-plugin/install.sh b/apps/memos-local-plugin/install.sh index db36c1e5f..bd972c8f1 100755 --- a/apps/memos-local-plugin/install.sh +++ b/apps/memos-local-plugin/install.sh @@ -394,7 +394,10 @@ deploy_tarball_to_prefix() { node_dir="$(dirname "${node_bin}")" node_version="$("${node_bin}" -v 2>/dev/null || echo "unknown")" printf "%s\n" "${node_bin}" > "${prefix}/.memos-node-bin" - ( cd "${prefix}" && PATH="${node_dir}:${PATH}" MEMOS_SKIP_SETUP=1 npm install --omit=dev --no-fund --no-audit --loglevel=error >/dev/null 2>&1 ) + # npm 11 still resolves omitted devDependency peer trees unless legacy peer + # resolution is requested. DSH peers are intentionally absent from the + # standalone OpenClaw/Hermes runtime. + ( cd "${prefix}" && PATH="${node_dir}:${PATH}" MEMOS_SKIP_SETUP=1 npm install --omit=dev --legacy-peer-deps --no-fund --no-audit --loglevel=error >/dev/null 2>&1 ) [[ -d "${prefix}/node_modules" ]] || die "npm install failed in ${prefix}" if [[ -d "${prefix}/node_modules/better-sqlite3" ]]; then diff --git a/apps/memos-local-plugin/templates/config.hermes.yaml b/apps/memos-local-plugin/templates/config.hermes.yaml index d08b320db..4c04daba3 100644 --- a/apps/memos-local-plugin/templates/config.hermes.yaml +++ b/apps/memos-local-plugin/templates/config.hermes.yaml @@ -20,6 +20,8 @@ viewer: embedding: provider: local # local | openai_compatible | gemini | cohere | voyage | mistral apiKey: "" + maxInputTokens: 1024 # conservative per-input limit; 0 disables client-side chunking + batchSize: 32 # maximum texts sent in one embedding-provider request llm: # Hermes has no "host LLM", so pick a real provider. diff --git a/apps/memos-local-plugin/templates/config.openclaw.yaml b/apps/memos-local-plugin/templates/config.openclaw.yaml index 1a84a55dc..743b78fd2 100644 --- a/apps/memos-local-plugin/templates/config.openclaw.yaml +++ b/apps/memos-local-plugin/templates/config.openclaw.yaml @@ -20,6 +20,8 @@ viewer: embedding: provider: local # local | openai_compatible | gemini | cohere | voyage | mistral apiKey: "" # required for cloud providers + maxInputTokens: 1024 # conservative per-input limit; 0 disables client-side chunking + batchSize: 32 # maximum texts sent in one embedding-provider request llm: provider: host # host | local_only | openai_compatible | anthropic | gemini | bedrock diff --git a/apps/memos-local-plugin/tests/unit/adapters/openclaw-runtime-lock.test.ts b/apps/memos-local-plugin/tests/unit/adapters/openclaw-runtime-lock.test.ts index b80000c02..32d3587cb 100644 --- a/apps/memos-local-plugin/tests/unit/adapters/openclaw-runtime-lock.test.ts +++ b/apps/memos-local-plugin/tests/unit/adapters/openclaw-runtime-lock.test.ts @@ -63,6 +63,9 @@ describe("OpenClaw runtime lock", () => { lock.release(); expect(fs.existsSync(lock.lockDir)).toBe(false); + + const reacquired = acquire(home); + reacquired.release(); }); it("rejects a second live owner before another runtime can bootstrap", () => { @@ -100,6 +103,57 @@ describe("OpenClaw runtime lock", () => { lock.release(); }); + it("reclaims a stale owner whose PID was recycled to the current process", () => { + const home = tmpHome(); + const lockDir = openClawRuntimeLockDir(home); + fs.mkdirSync(lockDir, { recursive: true }); + fs.writeFileSync( + path.join(lockDir, "owner.json"), + JSON.stringify({ + pluginId: "memos-local-plugin", + version: "old", + pid: process.pid, + token: "stale-self-token", + startedAt: 1, + dbFile: home.dbFile, + viewerPort: 18799, + }), + "utf8", + ); + + const lock = acquire(home); + expect(lock.owner.pid).toBe(process.pid); + expect(lock.owner.token).not.toBe("stale-self-token"); + + lock.release(); + }); + + it("reclaims a stale owner whose PID matches the configured owner PID", () => { + const home = tmpHome(); + const lockDir = openClawRuntimeLockDir(home); + const ownerPid = process.ppid; + fs.mkdirSync(lockDir, { recursive: true }); + fs.writeFileSync( + path.join(lockDir, "owner.json"), + JSON.stringify({ + pluginId: "memos-local-plugin", + version: "old", + pid: ownerPid, + token: "stale-configured-pid-token", + startedAt: 1, + dbFile: home.dbFile, + viewerPort: 18799, + }), + "utf8", + ); + + const lock = acquire(home, ownerPid); + expect(lock.owner.pid).toBe(ownerPid); + expect(lock.owner.token).not.toBe("stale-configured-pid-token"); + + lock.release(); + }); + it("allows diagnostic mode to skip lock when gateway is running", () => { const home = tmpHome(); const gatewayLock = acquire(home, process.pid, false); diff --git a/apps/memos-local-plugin/tests/unit/capture/embedder.test.ts b/apps/memos-local-plugin/tests/unit/capture/embedder.test.ts index 5df2be9a5..1fc7b2474 100644 --- a/apps/memos-local-plugin/tests/unit/capture/embedder.test.ts +++ b/apps/memos-local-plugin/tests/unit/capture/embedder.test.ts @@ -1,8 +1,10 @@ import { beforeAll, describe, expect, it } from "vitest"; +import { MemosError } from "../../../agent-contract/errors.js"; import { embedSteps } from "../../../core/capture/embedder.js"; import type { NormalizedStep } from "../../../core/capture/types.js"; -import { initTestLogger } from "../../../core/logger/index.js"; +import type { Embedder } from "../../../core/embedding/types.js"; +import { initTestLogger, memoryBuffer } from "../../../core/logger/index.js"; import { fakeEmbedder } from "../../helpers/fake-embedder.js"; function step(partial: Partial): NormalizedStep { @@ -111,6 +113,84 @@ describe("capture/embedder", () => { expect(out[0]!.action).toBeNull(); }); + it("preserves successful vectors when a neighboring input is rejected", async () => { + const base = fakeEmbedder({ dimensions: 3 }); + const e: Embedder = { + ...base, + async embedManySettled(inputs) { + return inputs.map((_, index) => index === 1 + ? { ok: false, error: new MemosError("embedding_unavailable", "bad action") } + : { ok: true, vector: new Float32Array([1, 2, 3]) }); + }, + }; + + const out = await embedSteps(e, [step({ userText: "good summary", agentText: "bad action" })]); + + expect(out[0]!.summary).toEqual(new Float32Array([1, 2, 3])); + expect(out[0]!.action).toBeNull(); + expect(memoryBuffer().tail({ channel: "core.capture.embed", limit: 20 })).toContainEqual( + expect.objectContaining({ + level: "warn", + msg: "embed.partial_failed", + data: expect.objectContaining({ failedCount: 1, inputCount: 2, stepCount: 1 }), + }), + ); + }); + + it("logs aggregate partial failures in summary-only mode", async () => { + const base = fakeEmbedder({ dimensions: 3 }); + const e: Embedder = { + ...base, + async embedManySettled(inputs) { + return inputs.map((_, index) => index === 0 + ? { ok: false, error: new MemosError("embedding_unavailable", "bad summary") } + : { ok: true, vector: new Float32Array([1, 2, 3]) }); + }, + }; + + const out = await embedSteps( + e, + [step({ userText: "bad" }), step({ userText: "good", key: "k2" })], + undefined, + { summaryOnly: true }, + ); + + expect(out[0]!.summary).toBeNull(); + expect(out[1]!.summary).toEqual(new Float32Array([1, 2, 3])); + expect(memoryBuffer().tail({ channel: "core.capture.embed", limit: 20 })).toContainEqual( + expect.objectContaining({ + level: "warn", + msg: "embed.partial_failed", + data: expect.objectContaining({ failedCount: 1, inputCount: 2, stepCount: 2 }), + }), + ); + }); + + it("reports an all-failed settled batch without calling it a partial failure", async () => { + const base = fakeEmbedder({ dimensions: 3 }); + const e: Embedder = { + ...base, + async embedManySettled(inputs) { + return inputs.map(() => ({ + ok: false, + error: new MemosError("embedding_unavailable", "provider unavailable"), + })); + }, + }; + + const out = await embedSteps(e, [step({ userText: "summary", agentText: "action" })]); + + expect(out[0]!.summary).toBeNull(); + expect(out[0]!.action).toBeNull(); + expect(memoryBuffer().tail({ channel: "core.capture.embed", limit: 20 })).toContainEqual( + expect.objectContaining({ + level: "warn", + msg: "embed.failed_all", + data: expect.objectContaining({ failedCount: 2, inputCount: 2, stepCount: 1 }), + }), + ); + }); + it("empty text step still produces a vector (uses '(empty)' fallback)", async () => { const e = fakeEmbedder(); const out = await embedSteps(e, [step({ userText: "", agentText: "" })]); diff --git a/apps/memos-local-plugin/tests/unit/config/load.test.ts b/apps/memos-local-plugin/tests/unit/config/load.test.ts index 5d85fd6e0..6bfd158db 100644 --- a/apps/memos-local-plugin/tests/unit/config/load.test.ts +++ b/apps/memos-local-plugin/tests/unit/config/load.test.ts @@ -1,9 +1,11 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { promises as fs } from "node:fs"; import { join } from "node:path"; +import { parse } from "yaml"; import { MemosError } from "../../../agent-contract/errors.js"; import { DEFAULT_CONFIG, loadConfig, resolveConfig, resolveHome } from "../../../core/config/index.js"; +import { ConfigSchema } from "../../../core/config/schema.js"; import { makeTmpHome } from "../../helpers/tmp-home.js"; describe("config/loadConfig", () => { @@ -54,6 +56,58 @@ describe("config/loadConfig", () => { expect(cfg.skillEvolver.openRouter).toBe(false); expect(cfg.l3Llm.openRouter).toBe(false); expect(cfg.embedding.openRouter).toBe(false); + expect(cfg.embedding.maxInputTokens).toBe(1_024); + expect(cfg.embedding.batchSize).toBe(32); + }); + + it("accepts operator-controlled embedding input and provider batch limits", () => { + const cfg = resolveConfig({ + embedding: { + maxInputTokens: 3_072, + batchSize: 4, + }, + }); + + expect(cfg.embedding.maxInputTokens).toBe(3_072); + expect(cfg.embedding.batchSize).toBe(4); + }); + + it("documents the zero-value maxInputTokens sentinel in JSON Schema", () => { + const schema = ConfigSchema as unknown as { + properties: { + embedding: { + properties: { maxInputTokens: { description?: string } }; + }; + }; + }; + + expect(schema.properties.embedding.properties.maxInputTokens.description).toContain("0"); + expect(schema.properties.embedding.properties.maxInputTokens.description).toContain( + "client-side chunking", + ); + }); + + it("ships new agent installations with a safe embedding input limit", async () => { + for (const template of ["config.openclaw.yaml", "config.hermes.yaml"]) { + const raw = await fs.readFile(join(__dirname, "../../../templates", template), "utf8"); + const cfg = resolveConfig(parse(raw)); + expect(cfg.embedding.maxInputTokens, template).toBe(1_024); + } + }); + + it("keeps long-input chunking disabled for existing configs that predate the setting", async () => { + const ctx = await makeTmpHome({ + agent: "openclaw", + configYaml: "version: 1\nembedding:\n provider: local\n", + }); + cleanup = ctx.cleanup; + + expect(ctx.config.embedding.maxInputTokens).toBe(0); + }); + + it("rejects invalid embedding input and provider batch limits", () => { + expect(() => resolveConfig({ embedding: { maxInputTokens: -1 } })).toThrow(MemosError); + expect(() => resolveConfig({ embedding: { batchSize: 0 } })).toThrow(MemosError); }); it("merges YAML over defaults and preserves unspecified branches", async () => { diff --git a/apps/memos-local-plugin/tests/unit/config/writer.test.ts b/apps/memos-local-plugin/tests/unit/config/writer.test.ts index e230acc14..9e7c8289c 100644 --- a/apps/memos-local-plugin/tests/unit/config/writer.test.ts +++ b/apps/memos-local-plugin/tests/unit/config/writer.test.ts @@ -94,6 +94,31 @@ llm: expect(idxViewer).toBeLessThan(idxLlm); }); + it("materializes the legacy disabled input limit when patching an older config", async () => { + const ctx = await makeTmpHome({ + agent: "openclaw", + configYaml: "version: 1\nembedding:\n provider: local\nllm:\n temperature: 0\n", + }); + cleanup = ctx.cleanup; + + const result = await patchConfig(ctx.home, { llm: { temperature: 0.2 } }); + + expect(result.config.embedding.maxInputTokens).toBe(0); + expect(await fs.readFile(ctx.home.configFile, "utf8")).toMatch(/maxInputTokens:\s*0/); + }); + + it("materializes the legacy input limit when the old embedding block is null", async () => { + const ctx = await makeTmpHome({ + agent: "openclaw", + configYaml: "version: 1\nembedding:\nllm:\n temperature: 0\n", + }); + cleanup = ctx.cleanup; + + const result = await patchConfig(ctx.home, { llm: { temperature: 0.2 } }); + + expect(result.config.embedding.maxInputTokens).toBe(0); + }); + it("validates after merge — invalid patches are rejected", async () => { const ctx = await makeTmpHome({ agent: "openclaw" }); cleanup = ctx.cleanup; 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..c4ced2d42 100644 --- a/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts +++ b/apps/memos-local-plugin/tests/unit/embedding/embedder.test.ts @@ -1,7 +1,10 @@ import { beforeAll, describe, expect, it } from "vitest"; import { MemosError } from "../../../agent-contract/errors.js"; -import { createEmbedderWithProvider } from "../../../core/embedding/embedder.js"; +import { + createEmbedderWithProvider, + estimateEmbeddingTokens, +} from "../../../core/embedding/embedder.js"; import { initTestLogger } from "../../../core/logger/index.js"; import type { EmbedRole, @@ -124,6 +127,83 @@ describe("embedder facade", () => { expect(p.calls.map((c) => c.texts)).toEqual([["a", "bb"], ["ccc", "dddd"]]); }); + it("chunks an over-limit input before calling the provider and pools one logical vector", async () => { + const p = new FakeProvider(); + const e = createEmbedderWithProvider(cfg({ maxInputTokens: 3, batchSize: 10 }), p); + + const out = await e.embedMany(["甲乙丙丁戊己庚辛"]); + + expect(out).toHaveLength(1); + expect(out[0]).toBeInstanceOf(Float32Array); + const chunks = p.calls.flatMap((call) => call.texts); + expect(chunks.length).toBeGreaterThan(1); + expect(chunks.length).toBeLessThanOrEqual(4); + expect(chunks.every((text) => estimateEmbeddingTokens(text) <= 3)).toBe(true); + }); + + it("automatically splits a provider batch rejected for request size", async () => { + const calls: string[][] = []; + const provider: EmbeddingProvider = { + name: "openai_compatible", + async embed(texts) { + calls.push([...texts]); + if (texts.length > 2) { + throw new MemosError("embedding_unavailable", "batch too large", { status: 400 }); + } + return texts.map((text) => [text.length, text.charCodeAt(0), 0]); + }, + }; + const e = createEmbedderWithProvider(cfg({ batchSize: 4 }), provider); + + const out = await e.embedMany(["a", "bb", "ccc", "dddd"]); + + expect(out).toHaveLength(4); + expect(calls).toEqual([ + ["a", "bb", "ccc", "dddd"], + ["a", "bb"], + ["ccc", "dddd"], + ]); + }); + + it("isolates a permanently rejected input while preserving valid neighbors", async () => { + const provider: EmbeddingProvider = { + name: "openai_compatible", + async embed(texts) { + if (texts.includes("bad")) { + throw new MemosError("embedding_unavailable", "invalid input", { status: 400 }); + } + return texts.map((text) => [text.length, text.charCodeAt(0), 0]); + }, + }; + const e = createEmbedderWithProvider(cfg({ batchSize: 3 }), provider); + + const settled = await e.embedManySettled?.(["good-a", "bad", "good-b"]); + + expect(settled).toBeDefined(); + expect(settled?.[0]?.ok).toBe(true); + expect(settled?.[1]?.ok).toBe(false); + expect(settled?.[2]?.ok).toBe(true); + if (settled?.[0]?.ok) expect(Array.from(settled[0].vector)).toEqual([6, 103, 0]); + if (settled?.[2]?.ok) expect(Array.from(settled[2].vector)).toEqual([6, 103, 0]); + }); + + it("does not recursively split authentication failures", async () => { + let calls = 0; + const provider: EmbeddingProvider = { + name: "openai_compatible", + async embed() { + calls++; + throw new MemosError("embedding_unavailable", "unauthorized", { status: 401 }); + }, + }; + const e = createEmbedderWithProvider(cfg({ batchSize: 3 }), provider); + + const settled = await e.embedManySettled?.(["a", "b", "c"]); + + expect(calls).toBe(1); + expect(settled?.every((result) => !result.ok)).toBe(true); + }); + it("splits by role before batching", async () => { const p = new FakeProvider(); const e = createEmbedderWithProvider(cfg({ batchSize: 10 }), p); diff --git a/apps/memos-local-plugin/tests/unit/install/hermes-provider-link.test.ts b/apps/memos-local-plugin/tests/unit/install/hermes-provider-link.test.ts index d0de2f5bc..6943c4807 100644 --- a/apps/memos-local-plugin/tests/unit/install/hermes-provider-link.test.ts +++ b/apps/memos-local-plugin/tests/unit/install/hermes-provider-link.test.ts @@ -97,6 +97,16 @@ describe("Hermes provider install links", () => { expect(preserveLine).not.toContain('"node_modules"'); }); + it("legacy agent installers omit DSH peers without resolving their development peer tree", () => { + const unixSource = readFileSync(path.join(repoRoot, "install.sh"), "utf8"); + const windowsSource = readFileSync(path.join(repoRoot, "install.ps1"), "utf8"); + + expect(unixSource).toContain("npm install --omit=dev --legacy-peer-deps"); + expect(windowsSource).toMatch( + /"install", "--omit=dev", "--legacy-peer-deps", "--no-fund"/, + ); + }); + it("PowerShell installer stops Hermes only after staging succeeds", () => { const source = readFileSync(path.join(repoRoot, "install.ps1"), "utf8"); const deployStart = source.indexOf("function Deploy-Tarball"); diff --git a/apps/memos-local-plugin/tests/unit/install/production-peer-deps.test.ts b/apps/memos-local-plugin/tests/unit/install/production-peer-deps.test.ts new file mode 100644 index 000000000..efee481b0 --- /dev/null +++ b/apps/memos-local-plugin/tests/unit/install/production-peer-deps.test.ts @@ -0,0 +1,28 @@ +import { readFileSync } from "node:fs"; +import path from "node:path"; + +import { describe, expect, it } from "vitest"; + +const repoRoot = path.resolve(__dirname, "../../.."); + +describe("production dependency staging", () => { + it("ignores development-only peer conflicts in the Unix installer", () => { + const source = readFileSync(path.join(repoRoot, "install.sh"), "utf8"); + const installCommand = source + .split("\n") + .find((line) => line.includes("npm install --omit=dev")); + + expect(installCommand).toBeDefined(); + expect(installCommand).toContain("--legacy-peer-deps"); + }); + + it("ignores development-only peer conflicts in the Windows installer", () => { + const source = readFileSync(path.join(repoRoot, "install.ps1"), "utf8"); + const installArguments = source + .split("\n") + .find((line) => line.includes('"install", "--omit=dev"')); + + expect(installArguments).toBeDefined(); + expect(installArguments).toContain('"--legacy-peer-deps"'); + }); +}); 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 500d37b6f..b4c095079 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 @@ -31,6 +31,8 @@ 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 as EmbeddingFailure } from "../../../agent-contract/errors.js"; +import type { Embedder } from "../../../core/embedding/types.js"; import type { SkillId, SkillRow, TraceRow } from "../../../core/types.js"; let db: TmpDbHandle | null = null; @@ -60,6 +62,7 @@ function configWithLightweightMemory(enabled: boolean): typeof DEFAULT_CONFIG { function buildDeps( h: TmpDbHandle, config: typeof DEFAULT_CONFIG = configWithLightweightMemory(false), + embedder: Embedder = fakeEmbedder({ dimensions: TEST_EMBED_DIMENSIONS }), ): PipelineDeps { return { agent: "openclaw", @@ -69,7 +72,7 @@ function buildDeps( repos: h.repos, llm: null, reflectLlm: null, - embedder: fakeEmbedder({ dimensions: TEST_EMBED_DIMENSIONS }), + embedder, log: rootLogger.child({ channel: "test.memory-core" }), namespace: { agentKind: "openclaw", profileId: "main" }, now: () => 1_700_000_000_000, @@ -280,6 +283,61 @@ describe("MemoryCore façade", () => { expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS); }); + it("repairs valid embedding slots while isolating a rejected neighbor", async () => { + const base = fakeEmbedder({ dimensions: TEST_EMBED_DIMENSIONS }); + const settledEmbedder: Embedder = { + ...base, + async embedManySettled(inputs) { + return inputs.map((input) => { + const text = typeof input === "string" ? input : input.text; + return text.includes("rejected action") + ? { ok: false, error: new EmbeddingFailure("embedding_unavailable", "bad action") } + : { ok: true, vector: new Float32Array(TEST_EMBED_DIMENSIONS).fill(1) }; + }); + }, + }; + pipeline = createPipeline(buildDeps(db!, configWithLightweightMemory(false), settledEmbedder)); + core = createMemoryCore( + pipeline, + resolveHome("openclaw", "/tmp/memos-mc-test"), + "test", + ); + await core.init(); + await core.importBundle({ + version: 1, + traces: [{ + id: "tr_partial_embedding", + episodeId: "ep_partial_embedding", + sessionId: "se_partial_embedding", + ts: 1_700_000_000_000, + userText: "valid summary source", + agentText: "rejected action source", + summary: "valid summary", + toolCalls: [], + value: 0, + alpha: 0, + priority: 0, + turnId: 1_700_000_000_000, + }], + }); + + const repaired = await core.rebuildEmbeddings({ mode: "repair", limit: 10 }); + + expect(repaired.updated).toBe(1); + expect(repaired.failed).toBe(1); + expect(repaired.done).toBe(false); + expect(repaired.error).toBeUndefined(); + const row = db!.repos.traces.getById("tr_partial_embedding" as never); + expect(row?.vecSummary?.length).toBe(TEST_EMBED_DIMENSIONS); + expect(row?.vecAction).toBeNull(); + + const terminal = await core.rebuildEmbeddings({ mode: "repair", limit: 10 }); + expect(terminal.updated).toBe(0); + expect(terminal.failed).toBe(1); + expect(terminal.done).toBe(true); + expect(terminal.error).toBe("bad action"); + }); + it("does not require action vectors for lightweight memory traces", async () => { pipeline = createPipeline(buildDeps(db!, configWithLightweightMemory(true))); core = createMemoryCore( diff --git a/apps/memos-local-plugin/tests/unit/util/foreground-resources.test.ts b/apps/memos-local-plugin/tests/unit/util/foreground-resources.test.ts index 1380b9a47..2b61e1d98 100644 --- a/apps/memos-local-plugin/tests/unit/util/foreground-resources.test.ts +++ b/apps/memos-local-plugin/tests/unit/util/foreground-resources.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from "vitest"; +import { MemosError } from "../../../agent-contract/errors.js"; +import type { Embedder } from "../../../core/embedding/types.js"; import { createForegroundResources, prioritizeEmbedder, @@ -102,6 +104,48 @@ describe("foreground resources", () => { expect(batchSizes).toEqual([2, 2, 1]); }); + it("preserves settled-result isolation through the priority wrapper", async () => { + const resources = createForegroundResources({ embeddingConcurrency: 1 }); + const base = fakeEmbedder({ dimensions: 4 }); + const inner = { + ...base, + async embedManySettled(inputs: Parameters[0]) { + return inputs.map(() => ({ + ok: true as const, + vector: new Float32Array([1, 2, 3, 4]), + })); + }, + }; + const background = prioritizeEmbedder(inner, resources, "background", 2)!; + + const settled = await background.embedManySettled?.(["a", "b", "c"]); + + expect(settled).toHaveLength(3); + expect(settled?.every((result) => result.ok)).toBe(true); + }); + + it("settles legacy embedMany failures instead of rejecting the wrapper call", async () => { + const resources = createForegroundResources({ embeddingConcurrency: 1 }); + const legacy: Embedder = { ...fakeEmbedder({ dimensions: 4 }) }; + delete legacy.embedManySettled; + legacy.embedMany = async () => { + throw new Error("legacy batch failed"); + }; + const background = prioritizeEmbedder(legacy, resources, "background", 2)!; + + const settled = await background.embedManySettled?.(["a", "b"]); + + expect(settled).toHaveLength(2); + expect(settled?.every((result) => !result.ok)).toBe(true); + for (const result of settled ?? []) { + if (!result.ok) { + expect(result.error).toBeInstanceOf(MemosError); + expect(result.error.code).toBe("embedding_unavailable"); + expect(result.error.message).toContain("legacy batch failed"); + } + } + }); + it("aborts queued and in-flight provider work during shutdown", async () => { const resources = createForegroundResources({ embeddingConcurrency: 1 }); const base = fakeEmbedder({ dimensions: 4 }); diff --git a/apps/memos-local-plugin/viewer/src/stores/i18n.ts b/apps/memos-local-plugin/viewer/src/stores/i18n.ts index bfa86e918..0833819c5 100644 --- a/apps/memos-local-plugin/viewer/src/stores/i18n.ts +++ b/apps/memos-local-plugin/viewer/src/stores/i18n.ts @@ -887,10 +887,12 @@ const en = { "Ready {ready}/{total}; missing {missing}; dimension mismatch {mismatch}; current dim {dim}.", "settings.embedding.maintenance.unavailable": "Configure an embedding provider before repairing or rebuilding vectors.", - "settings.embedding.batchSize.label": "Items per request", - "settings.embedding.batchSize.option": "{n} items per request", - "settings.embedding.batchSize.hint": - "Larger batches usually rebuild faster, but may hit provider limits or timeouts.", + "settings.embedding.maxInputTokens.label": "Maximum input tokens", + "settings.embedding.maxInputTokens.hint": + "Defaults to 1024; use 0 for no client-side limit. Longer inputs are sampled into chunks and pooled; rebuild vectors after changing it.", + "settings.embedding.providerBatchSize.label": "Embedding API batch size", + "settings.embedding.providerBatchSize.hint": + "Maximum texts per provider request. Rejected oversized batches are split automatically.", "settings.embedding.repair": "Repair missing/mismatched", "settings.embedding.rebuild": "Rebuild all vectors", "settings.embedding.rebuild.running": "Rebuilding embeddings…", @@ -1750,9 +1752,10 @@ const zh: Record = { "settings.embedding.maintenance.stats": "可用 {ready}/{total}；缺失 {missing}；维度不匹配 {mismatch}；当前维度 {dim}。", "settings.embedding.maintenance.unavailable": "请先配置嵌入模型，再修复或重建向量。", - "settings.embedding.batchSize.label": "每次请求条数", - "settings.embedding.batchSize.option": "每次请求 {n} 条", - "settings.embedding.batchSize.hint": "每次请求条数越大通常重建越快，但可能触发模型服务限流或超时。", + "settings.embedding.maxInputTokens.label": "单条输入最大 Token 数", + "settings.embedding.maxInputTokens.hint": "默认 1024；设为 0 表示不启用客户端限制。超长输入会分块采样并聚合向量，修改后请重建向量。", + "settings.embedding.providerBatchSize.label": "Embedding API 批量大小", + "settings.embedding.providerBatchSize.hint": "单次模型请求最多发送的文本数；超限失败时会自动拆批。", "settings.embedding.repair": "修复缺失/错维", "settings.embedding.rebuild": "全量重建向量", "settings.embedding.rebuild.running": "正在重建向量…", diff --git a/apps/memos-local-plugin/viewer/src/views/SettingsView.tsx b/apps/memos-local-plugin/viewer/src/views/SettingsView.tsx index 9a7cfd818..d3e62fdd5 100644 --- a/apps/memos-local-plugin/viewer/src/views/SettingsView.tsx +++ b/apps/memos-local-plugin/viewer/src/views/SettingsView.tsx @@ -37,6 +37,8 @@ interface ProviderBlock { model?: string; apiKey?: string; temperature?: number; + maxInputTokens?: number; + batchSize?: number; } interface AlgorithmBlock { @@ -87,10 +89,6 @@ interface EmbeddingMaintenanceRunResult { error?: string; } -const EMBEDDING_REBUILD_BATCH_STORAGE_KEY = "memos.embeddingRebuildBatchSize"; -const EMBEDDING_REBUILD_BATCH_OPTIONS = [10, 20, 50, 100, 200, 500] as const; -type EmbeddingRebuildBatchSize = typeof EMBEDDING_REBUILD_BATCH_OPTIONS[number]; - const SECRET_MASKED = (s: string | undefined | null): boolean => !!s && (s === "__memos_secret__" || /^[\s•]+$/.test(s)); @@ -528,6 +526,53 @@ function ModelCard({ onInput={(e) => onPatch({ apiKey: (e.target as HTMLInputElement).value })} /> + {type === "embedding" && ( + + + onPatch({ + maxInputTokens: Math.max( + 0, + Math.floor(Number((e.target as HTMLInputElement).value) || 0), + ), + })} + /> + + {t("settings.embedding.maxInputTokens.hint")} + + + )} + {type === "embedding" && ( + + + onPatch({ + batchSize: Math.max( + 1, + Math.min( + 256, + Math.floor(Number((e.target as HTMLInputElement).value) || 1), + ), + ), + })} + /> + + {t("settings.embedding.providerBatchSize.hint")} + + + )} {withTemperature && ( (null); const [running, setRunning] = useState<"repair" | "rebuild" | null>(null); const [status, setStatus] = useState<{ kind: "ok" | "error" | "muted"; text: string } | null>(null); - const [batchSize, setBatchSize] = useState(() => loadEmbeddingRebuildBatchSize()); const refresh = async () => { try { @@ -606,7 +650,7 @@ function EmbeddingMaintenancePanel() { for (;;) { const r = await api.post( "/api/v1/embeddings/rebuild", - { mode, offset, limit: batchSize }, + { mode, offset }, ); updated += r.updated; failed += r.failed; @@ -663,33 +707,6 @@ function EmbeddingMaintenancePanel() {
{healthText}
- -
- {t("settings.embedding.batchSize.hint")} -