diff --git a/CHANGELOG.md b/CHANGELOG.md index 6af5f3c33945..c3da626ac5d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- Memory/local embeddings: run local GGUF embeddings in an isolated worker sidecar and degrade to configured fallback or keyword search on worker failure so native embedding crashes do not take down the Gateway. (#85348) Thanks @osolmaz. - Agents/heartbeat: stop heartbeat turns after the first valid `heartbeat_respond` so repeated response loops do not burn tokens. (#86357) Thanks @udaymanish6. - Memory-core: keep REM dreaming focused on live light-staged memories and mark staged entries as considered so old recall history no longer dominates fresh candidates. (#86302) Thanks @SebTardif. - Telegram: propagate forum topic names through the account-scoped topic cache for native command context and topic create/edit actions. (#86299) Thanks @SebTardif. diff --git a/extensions/memory-core/src/memory/index.test.ts b/extensions/memory-core/src/memory/index.test.ts index ee75b9900acc..d6f49b2473ac 100644 --- a/extensions/memory-core/src/memory/index.test.ts +++ b/extensions/memory-core/src/memory/index.test.ts @@ -12,6 +12,7 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } import "./test-runtime-mocks.js"; import type { MemoryIndexManager } from "./index.js"; import { closeAllMemorySearchManagers, getMemorySearchManager } from "./index.js"; +import { LOCAL_EMBEDDING_WORKER_ERROR_CODES } from "./manager-local-worker-errors.js"; import { closeMemoryIndexManagersForAgent, EMBEDDING_PROBE_CACHE_TTL_MS } from "./manager.js"; import { DEFAULT_LOCAL_MODEL, @@ -34,6 +35,14 @@ let providerCloseGate: Promise | null = null; let providerCalls: Array<{ provider?: string; model?: string; outputDimensionality?: number }> = []; let forceNoProvider = false; +function createLocalWorkerExitError(): Error { + return Object.assign(new Error("Local embedding worker exited unexpectedly (exit code 134)"), { + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.exited, + reason: "exit", + exitCode: 134, + }); +} + vi.mock("./embeddings.js", () => { const embedText = (text: string) => { const lower = text.toLowerCase(); @@ -44,6 +53,10 @@ vi.mock("./embeddings.js", () => { return [alpha, beta, image, audio]; }; return { + resolveEmbeddingProviderFallbackModel: (providerId: string, fallbackSourceModel: string) => + providerId === "gemini" || providerId === "fallback-provider" + ? `${providerId}-embed` + : fallbackSourceModel, createEmbeddingProvider: async (options: { provider?: string; model?: string; @@ -61,7 +74,10 @@ vi.mock("./embeddings.js", () => { providerUnavailableReason: "No API key found for provider", }; } - const providerId = options.provider === "gemini" ? "gemini" : "mock"; + const providerId = + options.provider === "gemini" || options.provider === "fallback-provider" + ? options.provider + : "mock"; const model = options.model ?? "mock-embed"; return { requestedProvider: options.provider ?? "openai", @@ -81,7 +97,7 @@ vi.mock("./embeddings.js", () => { embedBatchCalls += 1; return texts.map(embedText); }, - ...(providerId === "gemini" + ...(providerId === "gemini" || providerId === "fallback-provider" ? { embedBatchInputs: async ( inputs: Array<{ @@ -112,12 +128,12 @@ vi.mock("./embeddings.js", () => { } : {}), }, - ...(providerId === "gemini" + ...(providerId === "gemini" || providerId === "fallback-provider" ? { runtime: { - id: "gemini", + id: providerId, cacheKeyData: { - provider: "gemini", + provider: providerId, baseUrl: "https://generativelanguage.googleapis.com/v1beta", model, outputDimensionality: options.outputDimensionality, @@ -242,7 +258,8 @@ describe("memory index", () => { extraPaths?: string[]; sources?: Array<"memory" | "sessions">; sessionMemory?: boolean; - provider?: "openai" | "gemini"; + provider?: "openai" | "gemini" | "fallback-provider"; + fallback?: "none" | "gemini" | "fallback-provider"; model?: string; outputDimensionality?: number; multimodal?: { @@ -263,6 +280,7 @@ describe("memory index", () => { memorySearch: { provider: params.provider ?? "openai", model: params.model ?? "mock-embed", + fallback: params.fallback, outputDimensionality: params.outputDimensionality, store: { path: params.storePath, vector: { enabled: params.vectorEnabled ?? false } }, // Perf: keep test indexes to a single chunk to reduce sqlite work. @@ -577,6 +595,144 @@ describe("memory index", () => { ); }); + it("clears cached embedding probe readiness when local embeddings degrade", async () => { + const cfg = createCfg({ storePath: path.join(workspaceDir, "index-probe-degraded.sqlite") }); + const manager = await getPersistentManager(cfg); + + await expect(manager.probeEmbeddingAvailability()).resolves.toEqual({ ok: true }); + expect(manager.getCachedEmbeddingAvailability()?.ok).toBe(true); + ( + manager as unknown as { + provider: { + id: string; + model: string; + embedQuery: (text: string) => Promise; + embedBatch: (texts: string[]) => Promise; + close: () => Promise; + }; + } + ).provider = { + id: "local", + model: "local-model", + embedQuery: async () => [1, 0], + embedBatch: async (texts: string[]) => texts.map(() => [1, 0]), + close: async () => {}, + }; + + ( + manager as unknown as { + markLocalEmbeddingProviderDegraded: (err: unknown) => void; + } + ).markLocalEmbeddingProviderDegraded(createLocalWorkerExitError()); + + expect(manager.getCachedEmbeddingAvailability()).toBeNull(); + await expect(manager.probeEmbeddingAvailability()).resolves.toMatchObject({ + ok: false, + error: expect.stringContaining("Local embeddings degraded"), + }); + }); + + it("activates configured fallback when local embeddings degrade during search", async () => { + const cfg = createCfg({ + storePath: path.join(workspaceDir, "index-search-degraded-fallback.sqlite"), + fallback: "fallback-provider", + hybrid: { enabled: true, vectorWeight: 0.5, textWeight: 0.5 }, + }); + const manager = await getPersistentManager(cfg); + + await manager.sync({ reason: "test" }); + const callsBeforeSearch = providerCalls.length; + ( + manager as unknown as { + provider: { + id: string; + model: string; + embedQuery: () => Promise; + embedBatch: (texts: string[]) => Promise; + close: () => Promise; + }; + } + ).provider = { + id: "local", + model: "mock-embed", + embedQuery: async () => { + throw createLocalWorkerExitError(); + }, + embedBatch: async (texts: string[]) => texts.map(() => [1, 0, 0, 0]), + close: async () => {}, + }; + + const results = await manager.search("alpha"); + + expect(results.length).toBeGreaterThan(0); + const resultKeys = results.map( + (result) => `${result.source}:${result.path}:${result.startLine}:${result.endLine}`, + ); + expect(new Set(resultKeys).size).toBe(resultKeys.length); + expect(providerCalls.slice(callsBeforeSearch).map((call) => call.provider)).toContain( + "fallback-provider", + ); + expect( + ( + manager as unknown as { + provider: { id: string } | null; + } + ).provider?.id, + ).toBe("fallback-provider"); + }); + + it("activates configured fallback after probe-time local degradation", async () => { + const cfg = createCfg({ + storePath: path.join(workspaceDir, "index-probe-degraded-fallback.sqlite"), + fallback: "fallback-provider", + hybrid: { enabled: true, vectorWeight: 0.5, textWeight: 0.5 }, + }); + const manager = await getPersistentManager(cfg); + + await manager.sync({ reason: "test" }); + ( + manager as unknown as { + provider: { + id: string; + model: string; + embedQuery: () => Promise; + embedBatch: () => Promise; + close: () => Promise; + }; + } + ).provider = { + id: "local", + model: "mock-embed", + embedQuery: async () => { + throw createLocalWorkerExitError(); + }, + embedBatch: async () => { + throw createLocalWorkerExitError(); + }, + close: async () => {}, + }; + const callsBeforeSearch = providerCalls.length; + + await expect(manager.probeEmbeddingAvailability()).resolves.toMatchObject({ + ok: false, + error: expect.stringContaining("Local embedding worker exited"), + }); + + const results = await manager.search("alpha"); + + expect(results.length).toBeGreaterThan(0); + expect(providerCalls.slice(callsBeforeSearch).map((call) => call.provider)).toContain( + "fallback-provider", + ); + expect( + ( + manager as unknown as { + provider: { id: string } | null; + } + ).provider?.id, + ).toBe("fallback-provider"); + }); + it("streams embedding cache rows during safe reindex", async () => { vi.stubEnv("OPENCLAW_TEST_MEMORY_UNSAFE_REINDEX", "0"); type EmbeddingCacheRow = { diff --git a/extensions/memory-core/src/memory/manager-embedding-errors.ts b/extensions/memory-core/src/memory/manager-embedding-errors.ts new file mode 100644 index 000000000000..df586304e233 --- /dev/null +++ b/extensions/memory-core/src/memory/manager-embedding-errors.ts @@ -0,0 +1,37 @@ +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; + +export const MEMORY_EMBEDDING_OPERATION_ERROR_CODE = "MEMORY_EMBEDDING_OPERATION_FAILED"; + +export type MemoryEmbeddingOperationKind = "query" | "batch" | "structured-batch"; + +export type MemoryEmbeddingOperationError = Error & { + code: typeof MEMORY_EMBEDDING_OPERATION_ERROR_CODE; + operation: MemoryEmbeddingOperationKind; + providerId?: string; + cause?: unknown; +}; + +export function createMemoryEmbeddingOperationError(params: { + operation: MemoryEmbeddingOperationKind; + providerId?: string; + cause: unknown; +}): MemoryEmbeddingOperationError { + const message = formatErrorMessage(params.cause); + const error = new Error(message) as MemoryEmbeddingOperationError; + error.code = MEMORY_EMBEDDING_OPERATION_ERROR_CODE; + error.operation = params.operation; + if (params.providerId) { + error.providerId = params.providerId; + } + error.cause = params.cause; + return error; +} + +export function isMemoryEmbeddingOperationError( + err: unknown, +): err is MemoryEmbeddingOperationError { + return ( + err instanceof Error && + (err as { code?: unknown }).code === MEMORY_EMBEDDING_OPERATION_ERROR_CODE + ); +} diff --git a/extensions/memory-core/src/memory/manager-embedding-ops.ts b/extensions/memory-core/src/memory/manager-embedding-ops.ts index eca200d67600..15209a9b7817 100644 --- a/extensions/memory-core/src/memory/manager-embedding-ops.ts +++ b/extensions/memory-core/src/memory/manager-embedding-ops.ts @@ -25,6 +25,7 @@ import { loadMemoryEmbeddingCache, upsertMemoryEmbeddingCache, } from "./manager-embedding-cache.js"; +import { createMemoryEmbeddingOperationError } from "./manager-embedding-errors.js"; import { buildMemoryEmbeddingBatches, buildTextEmbeddingInputs, @@ -144,6 +145,7 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { protected abstract batchFailureLastError?: string; protected abstract batchFailureLastProvider?: string; protected abstract batchFailureLock: Promise; + protected abstract markLocalEmbeddingProviderDegraded(err: unknown): void; protected pruneEmbeddingCacheIfNeeded(): void { if (!this.cache.enabled) { @@ -195,9 +197,13 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { const inputs = buildTextEmbeddingInputs(batch); const hasStructuredInputs = inputs.some((input) => hasNonTextEmbeddingParts(input)); if (hasStructuredInputs && !provider.embedBatchInputs) { - throw new Error( - `Embedding provider "${provider.id}" does not support multimodal memory inputs.`, - ); + throw createMemoryEmbeddingOperationError({ + operation: "structured-batch", + providerId: provider.id, + cause: new Error( + `Embedding provider "${provider.id}" does not support multimodal memory inputs.`, + ), + }); } const batchEmbeddings = hasStructuredInputs ? await this.embedBatchInputsWithRetry(inputs) @@ -324,27 +330,36 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { if (!provider) { throw new Error("Cannot embed batch in FTS-only mode (no embedding provider)"); } - return await runMemoryEmbeddingRetryLoop({ - run: async () => { - const timeoutMs = this.resolveEmbeddingTimeout("batch"); - log.debug("memory embeddings: batch start", { - provider: provider.id, - items: texts.length, - timeoutMs, - }); - return await runEmbeddingOperationWithTimeout({ - timeoutMs, - message: `memory embeddings batch timed out after ${Math.round(timeoutMs / 1000)}s`, - run: async (signal) => await provider.embedBatch(texts, { signal }), - }); - }, - isRetryable: isRetryableMemoryEmbeddingError, - waitForRetry: async (delayMs) => { - await this.waitForEmbeddingRetry(delayMs, "retrying"); - }, - maxAttempts: EMBEDDING_RETRY_MAX_ATTEMPTS, - baseDelayMs: EMBEDDING_RETRY_BASE_DELAY_MS, - }); + try { + return await runMemoryEmbeddingRetryLoop({ + run: async () => { + const timeoutMs = this.resolveEmbeddingTimeout("batch"); + log.debug("memory embeddings: batch start", { + provider: provider.id, + items: texts.length, + timeoutMs, + }); + return await runEmbeddingOperationWithTimeout({ + timeoutMs, + message: `memory embeddings batch timed out after ${Math.round(timeoutMs / 1000)}s`, + run: async (signal) => await provider.embedBatch(texts, { signal }), + }); + }, + isRetryable: isRetryableMemoryEmbeddingError, + waitForRetry: async (delayMs) => { + await this.waitForEmbeddingRetry(delayMs, "retrying"); + }, + maxAttempts: EMBEDDING_RETRY_MAX_ATTEMPTS, + baseDelayMs: EMBEDDING_RETRY_BASE_DELAY_MS, + }); + } catch (err) { + this.markLocalEmbeddingProviderDegraded(err); + throw createMemoryEmbeddingOperationError({ + operation: "batch", + providerId: provider.id, + cause: err, + }); + } } protected async embedBatchInputsWithRetry(inputs: EmbeddingInput[]): Promise { @@ -356,27 +371,36 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { if (!embedBatchInputs) { return await this.embedBatchWithRetry(inputs.map((input) => input.text)); } - return await runMemoryEmbeddingRetryLoop({ - run: async () => { - const timeoutMs = this.resolveEmbeddingTimeout("batch"); - log.debug("memory embeddings: structured batch start", { - provider: provider.id, - items: inputs.length, - timeoutMs, - }); - return await runEmbeddingOperationWithTimeout({ - timeoutMs, - message: `memory embeddings batch timed out after ${Math.round(timeoutMs / 1000)}s`, - run: async (signal) => await embedBatchInputs(inputs, { signal }), - }); - }, - isRetryable: isRetryableMemoryEmbeddingError, - waitForRetry: async (delayMs) => { - await this.waitForEmbeddingRetry(delayMs, "retrying structured batch"); - }, - maxAttempts: EMBEDDING_RETRY_MAX_ATTEMPTS, - baseDelayMs: EMBEDDING_RETRY_BASE_DELAY_MS, - }); + try { + return await runMemoryEmbeddingRetryLoop({ + run: async () => { + const timeoutMs = this.resolveEmbeddingTimeout("batch"); + log.debug("memory embeddings: structured batch start", { + provider: provider.id, + items: inputs.length, + timeoutMs, + }); + return await runEmbeddingOperationWithTimeout({ + timeoutMs, + message: `memory embeddings batch timed out after ${Math.round(timeoutMs / 1000)}s`, + run: async (signal) => await embedBatchInputs(inputs, { signal }), + }); + }, + isRetryable: isRetryableMemoryEmbeddingError, + waitForRetry: async (delayMs) => { + await this.waitForEmbeddingRetry(delayMs, "retrying structured batch"); + }, + maxAttempts: EMBEDDING_RETRY_MAX_ATTEMPTS, + baseDelayMs: EMBEDDING_RETRY_BASE_DELAY_MS, + }); + } catch (err) { + this.markLocalEmbeddingProviderDegraded(err); + throw createMemoryEmbeddingOperationError({ + operation: "structured-batch", + providerId: provider.id, + cause: err, + }); + } } private async waitForEmbeddingRetry(delayMs: number, action: string): Promise { @@ -405,11 +429,20 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } const timeoutMs = this.resolveEmbeddingTimeout("query"); log.debug("memory embeddings: query start", { provider: provider.id, timeoutMs }); - return await runEmbeddingOperationWithTimeout({ - timeoutMs, - message: `memory embeddings query timed out after ${Math.round(timeoutMs / 1000)}s`, - run: async (signal) => await provider.embedQuery(text, { signal }), - }); + try { + return await runEmbeddingOperationWithTimeout({ + timeoutMs, + message: `memory embeddings query timed out after ${Math.round(timeoutMs / 1000)}s`, + run: async (signal) => await provider.embedQuery(text, { signal }), + }); + } catch (err) { + this.markLocalEmbeddingProviderDegraded(err); + throw createMemoryEmbeddingOperationError({ + operation: "query", + providerId: provider.id, + cause: err, + }); + } } protected async withTimeout( diff --git a/extensions/memory-core/src/memory/manager-embedding-timeout.test.ts b/extensions/memory-core/src/memory/manager-embedding-timeout.test.ts index 6e7dfbe30496..4d464e0860fd 100644 --- a/extensions/memory-core/src/memory/manager-embedding-timeout.test.ts +++ b/extensions/memory-core/src/memory/manager-embedding-timeout.test.ts @@ -4,6 +4,10 @@ import { resolveMemoryIndexConcurrency, runEmbeddingOperationWithTimeout, } from "./manager-embedding-ops.js"; +import { + isLocalEmbeddingWorkerFailure, + LOCAL_EMBEDDING_WORKER_ERROR_CODES, +} from "./manager-local-worker-errors.js"; describe("memory embedding timeout resolution", () => { it("uses hosted defaults for inline embedding calls", () => { @@ -38,6 +42,34 @@ describe("memory embedding timeout resolution", () => { }); }); +describe("local embedding worker failure detection", () => { + it("matches structured local worker failure codes", () => { + expect( + isLocalEmbeddingWorkerFailure( + Object.assign(new Error("Local embedding worker exited unexpectedly (exit code 134)"), { + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.exited, + reason: "exit", + }), + ), + ).toBe(true); + expect( + isLocalEmbeddingWorkerFailure( + Object.assign(new Error("Local embedding worker process failed"), { + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.processError, + reason: "process-error", + }), + ), + ).toBe(true); + expect( + isLocalEmbeddingWorkerFailure( + Object.assign(new Error("Local embedding request aborted"), { + code: "ABORT_ERR", + }), + ), + ).toBe(false); + }); +}); + describe("memory embedding timeout abort", () => { it("aborts the provider operation when the timeout fires", async () => { let signalSeen: AbortSignal | undefined; diff --git a/extensions/memory-core/src/memory/manager-local-worker-errors.ts b/extensions/memory-core/src/memory/manager-local-worker-errors.ts new file mode 100644 index 000000000000..d7b4f32ffdcc --- /dev/null +++ b/extensions/memory-core/src/memory/manager-local-worker-errors.ts @@ -0,0 +1,25 @@ +export const LOCAL_EMBEDDING_WORKER_ERROR_CODES = { + exited: "LOCAL_EMBEDDING_WORKER_EXITED", + processError: "LOCAL_EMBEDDING_WORKER_PROCESS_ERROR", + ipcError: "LOCAL_EMBEDDING_WORKER_IPC_ERROR", +} as const; + +export type LocalEmbeddingWorkerFailureCode = + (typeof LOCAL_EMBEDDING_WORKER_ERROR_CODES)[keyof typeof LOCAL_EMBEDDING_WORKER_ERROR_CODES]; + +export type LocalEmbeddingWorkerFailureError = Error & { + code: LocalEmbeddingWorkerFailureCode; +}; + +const LOCAL_EMBEDDING_WORKER_FAILURE_CODES = new Set( + Object.values(LOCAL_EMBEDDING_WORKER_ERROR_CODES), +); + +export function isLocalEmbeddingWorkerFailure( + err: unknown, +): err is LocalEmbeddingWorkerFailureError { + return ( + err instanceof Error && + LOCAL_EMBEDDING_WORKER_FAILURE_CODES.has(String((err as { code?: unknown }).code)) + ); +} diff --git a/extensions/memory-core/src/memory/manager-provider-state.ts b/extensions/memory-core/src/memory/manager-provider-state.ts index ad30efc05df3..fd2668353932 100644 --- a/extensions/memory-core/src/memory/manager-provider-state.ts +++ b/extensions/memory-core/src/memory/manager-provider-state.ts @@ -15,8 +15,96 @@ type MemoryResolvedProviderState = { fallbackReason?: string; providerUnavailableReason?: string; providerRuntime?: EmbeddingProviderRuntime; + lifecycle: MemoryProviderLifecycleState; }; +export type MemoryProviderLifecycleState = + | { + mode: "pending"; + requestedProvider: string; + } + | { + mode: "active"; + providerId: string; + } + | { + mode: "degraded"; + providerId: string; + reason: string; + code?: string; + } + | { + mode: "fallback-active"; + providerId: string; + fallbackFrom: string; + reason: string; + } + | { + mode: "fts-only"; + reason: string; + attemptedProviderId?: string; + }; + +export function createPendingMemoryProviderLifecycle( + requestedProvider: string, +): MemoryProviderLifecycleState { + return { mode: "pending", requestedProvider }; +} + +export function createDegradedMemoryProviderLifecycle(params: { + providerId: string; + reason: string; + code?: string; +}): MemoryProviderLifecycleState { + return { + mode: "degraded", + providerId: params.providerId, + reason: params.reason, + ...(params.code ? { code: params.code } : {}), + }; +} + +function resolveProviderLifecycle( + result: Pick< + EmbeddingProviderResult, + | "provider" + | "fallbackFrom" + | "fallbackReason" + | "providerUnavailableReason" + | "requestedProvider" + >, +): MemoryProviderLifecycleState { + if (result.provider && result.fallbackFrom) { + return { + mode: "fallback-active", + providerId: result.provider.id, + fallbackFrom: result.fallbackFrom, + reason: result.fallbackReason ?? "fallback activated", + }; + } + if (result.provider) { + return { mode: "active", providerId: result.provider.id }; + } + return { + mode: "fts-only", + reason: result.providerUnavailableReason ?? "No embedding provider available", + attemptedProviderId: result.requestedProvider, + }; +} + +export function resolveFallbackCurrentProviderId(params: { + provider: EmbeddingProvider | null; + lifecycle: MemoryProviderLifecycleState; +}): string | null { + if (params.provider) { + return params.provider.id; + } + if (params.lifecycle.mode === "degraded") { + return params.lifecycle.providerId; + } + return null; +} + export function resolveMemoryPrimaryProviderRequest(params: { settings: ResolvedMemorySearchConfig; }): { @@ -46,7 +134,12 @@ export function resolveMemoryPrimaryProviderRequest(params: { export function resolveMemoryProviderState( result: Pick< EmbeddingProviderResult, - "provider" | "fallbackFrom" | "fallbackReason" | "providerUnavailableReason" | "runtime" + | "provider" + | "fallbackFrom" + | "fallbackReason" + | "providerUnavailableReason" + | "runtime" + | "requestedProvider" >, ): MemoryResolvedProviderState { return { @@ -55,6 +148,7 @@ export function resolveMemoryProviderState( fallbackReason: result.fallbackReason, providerUnavailableReason: result.providerUnavailableReason, providerRuntime: result.runtime, + lifecycle: resolveProviderLifecycle(result), }; } @@ -68,8 +162,21 @@ export function applyMemoryFallbackProviderState(params: { ...params.current, fallbackFrom: params.fallbackFrom, fallbackReason: params.reason, + providerUnavailableReason: undefined, provider: params.result.provider, providerRuntime: params.result.runtime, + lifecycle: params.result.provider + ? { + mode: "fallback-active", + providerId: params.result.provider.id, + fallbackFrom: params.fallbackFrom, + reason: params.reason, + } + : { + mode: "fts-only", + reason: params.reason, + attemptedProviderId: params.fallbackFrom, + }, }; } diff --git a/extensions/memory-core/src/memory/manager-sync-ops.archive-delta-bypass.test.ts b/extensions/memory-core/src/memory/manager-sync-ops.archive-delta-bypass.test.ts index 37ac2c448893..36534e3a86f1 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.archive-delta-bypass.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.archive-delta-bypass.test.ts @@ -52,6 +52,8 @@ class SessionDeltaHarness extends MemoryManagerSyncOps { }; protected readonly vector = { enabled: false, available: false }; protected readonly cache = { enabled: false }; + protected providerUnavailableReason?: string; + protected providerLifecycle = { mode: "active" as const, providerId: "test" }; protected db = null as unknown as DatabaseSync; readonly syncCalls: SyncParams[] = []; diff --git a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts index a5338c330ed8..d1396d6ac967 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts @@ -54,6 +54,8 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps { }; protected readonly vector = { enabled: false, available: false }; protected readonly cache = { enabled: false }; + protected providerUnavailableReason?: string; + protected providerLifecycle = { mode: "active" as const, providerId: "test" }; protected db: DatabaseSync; readonly syncCalls: SyncParams[] = []; diff --git a/extensions/memory-core/src/memory/manager-sync-ops.ts b/extensions/memory-core/src/memory/manager-sync-ops.ts index 6763a6fbc1da..a7fc8c8d0fbf 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.ts @@ -42,9 +42,12 @@ import { } from "./embeddings.js"; import { runMemoryAtomicReindex } from "./manager-atomic-reindex.js"; import { closeMemoryDatabase, openMemoryDatabaseAtPath } from "./manager-db.js"; +import { isMemoryEmbeddingOperationError } from "./manager-embedding-errors.js"; import { applyMemoryFallbackProviderState, resolveMemoryFallbackProviderRequest, + resolveFallbackCurrentProviderId, + type MemoryProviderLifecycleState, } from "./manager-provider-state.js"; import { resolveConfiguredScopeHash, @@ -170,6 +173,8 @@ export abstract class MemoryManagerSyncOps { protected abstract readonly settings: ResolvedMemorySearchConfig; protected provider: EmbeddingProvider | null = null; protected fallbackFrom?: EmbeddingProviderId; + protected abstract providerUnavailableReason?: string; + protected abstract providerLifecycle: MemoryProviderLifecycleState; protected providerRuntime?: EmbeddingProviderRuntime; protected abstract batch: { enabled: boolean; @@ -1131,7 +1136,7 @@ export abstract class MemoryManagerSyncOps { syncSessionFiles: async (targetedParams) => { await this.syncSessionFiles(targetedParams); }, - shouldFallbackOnError: (message) => this.shouldFallbackOnError(message), + shouldFallbackOnError: (err) => this.shouldFallbackOnError(err), activateFallbackProvider: async (reason) => await this.activateFallbackProvider(reason), runSafeReindex: async (reindexParams) => { await this.runSafeReindex(reindexParams); @@ -1205,7 +1210,7 @@ export abstract class MemoryManagerSyncOps { } catch (err) { const reason = formatErrorMessage(err); const activated = - this.shouldFallbackOnError(reason) && (await this.activateFallbackProvider(reason)); + this.shouldFallbackOnError(err) && (await this.activateFallbackProvider(reason)); if (activated) { await this.runSafeReindex({ reason: params?.reason ?? "fallback", @@ -1214,12 +1219,21 @@ export abstract class MemoryManagerSyncOps { }); return; } + if (!this.provider && this.fts.enabled && this.shouldFallbackOnError(err)) { + log.warn(`memory embeddings unavailable; rebuilding lexical memory index only: ${reason}`); + await this.runSafeReindex({ + reason: params?.reason ?? "embedding-degraded", + force: true, + progress: progress ?? undefined, + }); + return; + } throw err; } } - private shouldFallbackOnError(message: string): boolean { - return /embedding|embeddings|batch/i.test(message); + protected shouldFallbackOnError(err: unknown): boolean { + return isMemoryEmbeddingOperationError(err); } protected resolveBatchConfig(): { @@ -1240,19 +1254,22 @@ export abstract class MemoryManagerSyncOps { }; } - private async activateFallbackProvider(reason: string): Promise { + protected async activateFallbackProvider(reason: string): Promise { + const currentProviderId = resolveFallbackCurrentProviderId({ + provider: this.provider, + lifecycle: this.providerLifecycle, + }); const fallbackRequest = resolveMemoryFallbackProviderRequest({ cfg: this.cfg, settings: this.settings, - currentProviderId: this.provider?.id ?? null, + currentProviderId, }); - if (!fallbackRequest || !this.provider) { + if (!fallbackRequest || !currentProviderId) { return false; } if (this.fallbackFrom) { return false; } - const fallbackFrom = this.provider.id; const fallbackResult = await createEmbeddingProvider({ config: this.cfg, @@ -1267,8 +1284,9 @@ export abstract class MemoryManagerSyncOps { fallbackReason: this.fallbackReason, providerUnavailableReason: undefined, providerRuntime: this.providerRuntime, + lifecycle: this.providerLifecycle, }, - fallbackFrom, + fallbackFrom: currentProviderId, reason, result: fallbackResult, }); @@ -1276,6 +1294,8 @@ export abstract class MemoryManagerSyncOps { this.fallbackReason = fallbackState.fallbackReason; this.provider = fallbackState.provider; this.providerRuntime = fallbackState.providerRuntime; + this.providerUnavailableReason = fallbackState.providerUnavailableReason; + this.providerLifecycle = fallbackState.lifecycle; this.providerKey = this.computeProviderKey(); this.batch = this.resolveBatchConfig(); log.warn(`memory embeddings: switched to fallback provider (${fallbackRequest.provider})`, { @@ -1284,7 +1304,7 @@ export abstract class MemoryManagerSyncOps { return true; } - private async runSafeReindex(params: { + protected async runSafeReindex(params: { reason?: string; force?: boolean; progress?: MemorySyncProgressState; diff --git a/extensions/memory-core/src/memory/manager-sync-yield.test.ts b/extensions/memory-core/src/memory/manager-sync-yield.test.ts index 57749c7eb94c..fd801741461e 100644 --- a/extensions/memory-core/src/memory/manager-sync-yield.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-yield.test.ts @@ -80,6 +80,8 @@ class SessionSyncYieldHarness extends MemoryManagerSyncOps { }; protected readonly vector = { enabled: false, available: false }; protected readonly cache = { enabled: false }; + protected providerUnavailableReason?: string; + protected providerLifecycle = { mode: "active" as const, providerId: "test" }; protected db = createDbMock(); readonly indexedPaths: string[] = []; diff --git a/extensions/memory-core/src/memory/manager-targeted-sync.ts b/extensions/memory-core/src/memory/manager-targeted-sync.ts index e29d7c6e10d2..0074b6198989 100644 --- a/extensions/memory-core/src/memory/manager-targeted-sync.ts +++ b/extensions/memory-core/src/memory/manager-targeted-sync.ts @@ -34,7 +34,7 @@ export async function runMemoryTargetedSessionSync(params: { targetSessionFiles?: string[]; progress?: TargetedSyncProgress; }) => Promise; - shouldFallbackOnError: (message: string) => boolean; + shouldFallbackOnError: (err: unknown) => boolean; activateFallbackProvider: (reason: string) => Promise; runSafeReindex: (params: { reason?: string; @@ -70,7 +70,7 @@ export async function runMemoryTargetedSessionSync(params: { } catch (err) { const reason = formatErrorMessage(err); const activated = - params.shouldFallbackOnError(reason) && (await params.activateFallbackProvider(reason)); + params.shouldFallbackOnError(err) && (await params.activateFallbackProvider(reason)); if (!activated) { throw err; } diff --git a/extensions/memory-core/src/memory/manager.mistral-provider.test.ts b/extensions/memory-core/src/memory/manager.mistral-provider.test.ts index bee1410793b0..f8c1bb727b44 100644 --- a/extensions/memory-core/src/memory/manager.mistral-provider.test.ts +++ b/extensions/memory-core/src/memory/manager.mistral-provider.test.ts @@ -80,6 +80,7 @@ describe("memory manager mistral provider wiring", () => { const state = resolveMemoryProviderState({ provider: mistralProvider, + requestedProvider: "mistral", runtime: mistralRuntime, fallbackFrom: undefined, fallbackReason: undefined, @@ -102,6 +103,7 @@ describe("memory manager mistral provider wiring", () => { const mistralProvider = createProvider("mistral"); const current = resolveMemoryProviderState({ provider: createProvider("openai"), + requestedProvider: "openai", runtime: openAiRuntime, fallbackFrom: undefined, fallbackReason: undefined, @@ -124,6 +126,30 @@ describe("memory manager mistral provider wiring", () => { expect(fallbackState.providerRuntime).toBe(mistralRuntime); }); + it("clears provider unavailable reason after fallback activation", () => { + const fallbackState = applyMemoryFallbackProviderState({ + current: resolveMemoryProviderState({ + provider: null, + requestedProvider: "local", + fallbackFrom: undefined, + fallbackReason: undefined, + providerUnavailableReason: "Local embeddings degraded: worker crashed", + runtime: undefined, + }), + fallbackFrom: "local", + reason: "worker crashed", + result: { + provider: createProvider("openai"), + runtime: { + id: "openai", + cacheKeyData: { provider: "openai", model: "text-embedding-3-small" }, + }, + }, + }); + + expect(fallbackState.providerUnavailableReason).toBeUndefined(); + }); + it("uses default ollama model when activating ollama fallback", () => { const request = resolveMemoryFallbackProviderRequest({ cfg: {} as OpenClawConfig, diff --git a/extensions/memory-core/src/memory/manager.ts b/extensions/memory-core/src/memory/manager.ts index 684c18661e3d..119a420e5b29 100644 --- a/extensions/memory-core/src/memory/manager.ts +++ b/extensions/memory-core/src/memory/manager.ts @@ -38,9 +38,13 @@ import { } from "./manager-cache.js"; import { closeMemoryDatabase } from "./manager-db.js"; import { MemoryManagerEmbeddingOps } from "./manager-embedding-ops.js"; +import { isLocalEmbeddingWorkerFailure } from "./manager-local-worker-errors.js"; import { + createDegradedMemoryProviderLifecycle, + createPendingMemoryProviderLifecycle, resolveMemoryPrimaryProviderRequest, resolveMemoryProviderState, + type MemoryProviderLifecycleState, } from "./manager-provider-state.js"; import { resolveMemorySearchPreflight } from "./manager-search-preflight.js"; import { searchKeyword, searchVector } from "./manager-search.js"; @@ -124,7 +128,8 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem private providerInitialized = false; protected override fallbackFrom?: EmbeddingProviderId; protected override fallbackReason?: string; - private providerUnavailableReason?: string; + protected providerUnavailableReason?: string; + protected override providerLifecycle: MemoryProviderLifecycleState; protected override providerRuntime?: EmbeddingProviderRuntime; protected batch: { enabled: boolean; @@ -239,6 +244,7 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem this.settings = params.settings; this.provider = null; this.requestedProvider = params.settings.provider; + this.providerLifecycle = createPendingMemoryProviderLifecycle(this.requestedProvider); if (params.providerResult) { this.applyProviderResult(params.providerResult); } @@ -283,6 +289,7 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem this.fallbackFrom = providerState.fallbackFrom; this.fallbackReason = providerState.fallbackReason; this.providerUnavailableReason = providerState.providerUnavailableReason; + this.providerLifecycle = providerState.lifecycle; this.providerRuntime = providerState.providerRuntime; this.providerInitialized = true; } @@ -312,6 +319,35 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem } } + protected markLocalEmbeddingProviderDegraded(err: unknown): void { + if (this.provider?.id !== "local") { + return; + } + if (!isLocalEmbeddingWorkerFailure(err)) { + return; + } + const message = formatErrorMessage(err); + const degradedProvider = this.provider; + this.provider = null; + this.providerRuntime = undefined; + this.providerUnavailableReason = `Local embeddings degraded: ${message}`; + this.providerLifecycle = createDegradedMemoryProviderLifecycle({ + providerId: degradedProvider.id, + reason: message, + code: err.code, + }); + EMBEDDING_PROBE_CACHE.delete(this.cacheKey); + this.providerKey = this.computeProviderKey(); + this.batch = this.resolveBatchConfig(); + this.vector.semanticAvailable = false; + void Promise.resolve(degradedProvider.close?.()).catch((err: unknown) => { + log.debug(`memory embeddings: failed to close degraded local provider: ${String(err)}`); + }); + log.warn("memory embeddings: local provider degraded after worker failure", { + error: message, + }); + } + async warmSession(sessionKey?: string): Promise { if (!this.settings.sync.onSessionStart) { return; @@ -394,6 +430,20 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem Math.max(1, Math.floor(maxResults * hybrid.candidateMultiplier)), ); + if (!this.provider && this.providerLifecycle.mode === "degraded") { + const activatedFallback = await this.activateFallbackProvider( + this.providerLifecycle.reason, + ).catch((fallbackErr: unknown) => { + log.warn( + `memory search: failed to activate fallback provider: ${formatErrorMessage(fallbackErr)}`, + ); + return false; + }); + if (activatedFallback) { + await this.runSafeReindex({ reason: "fallback", force: true }); + } + } + // FTS-only mode: no embedding provider available if (!this.provider) { if (!this.fts.enabled || !this.fts.available) { @@ -461,7 +511,7 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem } // If FTS isn't available, hybrid mode cannot use keyword search; degrade to vector-only. - const keywordResults = + const loadKeywordResults = async () => hybrid.enabled && this.fts.enabled && this.fts.available ? await this.searchKeyword( cleaned, @@ -473,8 +523,32 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem return []; }) : []; + let keywordResults = await loadKeywordResults(); - const queryVec = await this.embedQueryWithTimeout(cleaned); + let queryVec: number[]; + try { + queryVec = await this.embedQueryWithTimeout(cleaned); + } catch (err) { + const message = formatErrorMessage(err); + const activatedFallback = this.shouldFallbackOnError(err) + ? await this.activateFallbackProvider(message).catch((fallbackErr: unknown) => { + log.warn( + `memory search: failed to activate fallback provider: ${formatErrorMessage(fallbackErr)}`, + ); + return false; + }) + : false; + if (activatedFallback) { + await this.runSafeReindex({ reason: "fallback", force: true }); + keywordResults = await loadKeywordResults(); + queryVec = await this.embedQueryWithTimeout(cleaned); + } else if (!this.provider && this.fts.enabled && this.fts.available) { + log.warn(`memory search: embeddings unavailable; using keyword-only results: ${message}`); + return this.selectScoredResults(keywordResults, maxResults, minScore, 0); + } else { + throw err; + } + } const hasVector = queryVec.some((v) => v !== 0); const vectorResults = hasVector ? await this.searchVector(queryVec, candidates, sourceFilterList).catch((err) => { @@ -846,6 +920,7 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem }, custom: { searchMode: providerInfo.searchMode, + providerState: this.providerLifecycle, providerUnavailableReason: this.providerUnavailableReason, readonlyRecovery: { attempts: this.readonlyRecoveryAttempts, diff --git a/extensions/memory-core/src/memory/qmd-manager.ts b/extensions/memory-core/src/memory/qmd-manager.ts index 339aa7cd602d..81ee38dba34a 100644 --- a/extensions/memory-core/src/memory/qmd-manager.ts +++ b/extensions/memory-core/src/memory/qmd-manager.ts @@ -1401,6 +1401,8 @@ export class QmdMemoryManager implements MemorySearchManager { qmd: { collections: this.qmd.collections.length, lastUpdateAt: this.lastUpdateAt, + embedFailures: this.embedFailureCount, + embedBackoffUntil: this.embedBackoffUntil, }, }, }; diff --git a/packages/memory-host-sdk/src/host/embedding-worker-errors.ts b/packages/memory-host-sdk/src/host/embedding-worker-errors.ts new file mode 100644 index 000000000000..fc1762bc1364 --- /dev/null +++ b/packages/memory-host-sdk/src/host/embedding-worker-errors.ts @@ -0,0 +1,47 @@ +export const LOCAL_EMBEDDING_WORKER_ERROR_CODES = { + exited: "LOCAL_EMBEDDING_WORKER_EXITED", + processError: "LOCAL_EMBEDDING_WORKER_PROCESS_ERROR", + ipcError: "LOCAL_EMBEDDING_WORKER_IPC_ERROR", +} as const; + +export type LocalEmbeddingWorkerFailureCode = + (typeof LOCAL_EMBEDDING_WORKER_ERROR_CODES)[keyof typeof LOCAL_EMBEDDING_WORKER_ERROR_CODES]; + +export type LocalEmbeddingWorkerFailureReason = "exit" | "signal" | "process-error" | "ipc"; + +export type LocalEmbeddingWorkerFailureError = Error & { + code: LocalEmbeddingWorkerFailureCode; + reason: LocalEmbeddingWorkerFailureReason; + exitCode?: number | null; + signal?: NodeJS.Signals | null; +}; + +const LOCAL_EMBEDDING_WORKER_FAILURE_CODES = new Set( + Object.values(LOCAL_EMBEDDING_WORKER_ERROR_CODES), +); + +export function createLocalEmbeddingWorkerFailureError(params: { + message: string; + code: LocalEmbeddingWorkerFailureCode; + reason: LocalEmbeddingWorkerFailureReason; + exitCode?: number | null; + signal?: NodeJS.Signals | null; + cause?: unknown; +}): LocalEmbeddingWorkerFailureError { + return Object.assign(new Error(params.message), { + code: params.code, + reason: params.reason, + ...(params.exitCode !== undefined ? { exitCode: params.exitCode } : {}), + ...(params.signal !== undefined ? { signal: params.signal } : {}), + ...(params.cause !== undefined ? { cause: params.cause } : {}), + }); +} + +export function isLocalEmbeddingWorkerFailure( + err: unknown, +): err is LocalEmbeddingWorkerFailureError { + return ( + err instanceof Error && + LOCAL_EMBEDDING_WORKER_FAILURE_CODES.has(String((err as { code?: unknown }).code)) + ); +} diff --git a/packages/memory-host-sdk/src/host/embeddings-worker-child.ts b/packages/memory-host-sdk/src/host/embeddings-worker-child.ts new file mode 100644 index 000000000000..a088f2b4c393 --- /dev/null +++ b/packages/memory-host-sdk/src/host/embeddings-worker-child.ts @@ -0,0 +1,108 @@ +import { createLocalEmbeddingProviderInProcess } from "./embeddings.js"; +import type { EmbeddingProvider, EmbeddingProviderOptions } from "./embeddings.types.js"; + +type LocalEmbeddingWorkerRequest = + | { + id: number; + type: "initialize"; + options: EmbeddingProviderOptions; + } + | { + id: number; + type: "embedQuery"; + options: EmbeddingProviderOptions; + text: string; + } + | { + id: number; + type: "embedBatch"; + options: EmbeddingProviderOptions; + texts: string[]; + } + | { + id: number; + type: "close"; + }; + +type LocalEmbeddingWorkerSerializedError = { + message: string; + code?: string; +}; + +let provider: EmbeddingProvider | null = null; +let providerOptionsKey: string | null = null; +let requestQueue: Promise = Promise.resolve(); + +function send(message: unknown): void { + if (typeof process.send === "function") { + process.send(message); + } +} + +async function getProvider(options: EmbeddingProviderOptions): Promise { + const key = JSON.stringify(options); + if (provider && providerOptionsKey === key) { + return provider; + } + await provider?.close?.(); + provider = await createLocalEmbeddingProviderInProcess(options); + providerOptionsKey = key; + return provider; +} + +async function closeProvider(): Promise { + const current = provider; + provider = null; + providerOptionsKey = null; + await current?.close?.(); +} + +function serializeError(err: unknown): LocalEmbeddingWorkerSerializedError { + if (!(err instanceof Error)) { + return { message: String(err) }; + } + const code = (err as Error & { code?: unknown }).code; + return { + message: err.message, + ...(typeof code === "string" ? { code } : {}), + }; +} + +async function handleRequest(request: LocalEmbeddingWorkerRequest): Promise { + if (request.type === "close") { + await closeProvider(); + send({ id: request.id, ok: true }); + return; + } + + const currentProvider = await getProvider(request.options); + if (request.type === "initialize") { + send({ id: request.id, ok: true }); + return; + } + if (request.type === "embedQuery") { + const value = await currentProvider.embedQuery(request.text); + send({ id: request.id, ok: true, value }); + return; + } + + const value = await currentProvider.embedBatch(request.texts); + send({ id: request.id, ok: true, value }); +} + +process.on("message", (message) => { + const request = message as LocalEmbeddingWorkerRequest; + requestQueue = requestQueue.then(async () => { + try { + await handleRequest(request); + } catch (err) { + send({ id: request.id, ok: false, error: serializeError(err) }); + } + }); +}); + +process.once("disconnect", () => { + void closeProvider().finally(() => { + process.exit(0); + }); +}); diff --git a/packages/memory-host-sdk/src/host/embeddings-worker.ts b/packages/memory-host-sdk/src/host/embeddings-worker.ts new file mode 100644 index 000000000000..0defd7617e4b --- /dev/null +++ b/packages/memory-host-sdk/src/host/embeddings-worker.ts @@ -0,0 +1,364 @@ +import { fork, type ChildProcess } from "node:child_process"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { DEFAULT_LOCAL_MODEL } from "./embedding-defaults.js"; +import { + createLocalEmbeddingWorkerFailureError, + LOCAL_EMBEDDING_WORKER_ERROR_CODES, +} from "./embedding-worker-errors.js"; +import type { LocalEmbeddingProviderRuntimeOptions } from "./embeddings.js"; +import type { + EmbeddingProvider, + EmbeddingProviderCallOptions, + EmbeddingProviderOptions, +} from "./embeddings.types.js"; +import { normalizeOptionalString } from "./string-utils.js"; + +type LocalEmbeddingWorkerRequestPayload = + | { + type: "initialize"; + options: EmbeddingProviderOptions; + } + | { + type: "embedQuery"; + options: EmbeddingProviderOptions; + text: string; + } + | { + type: "embedBatch"; + options: EmbeddingProviderOptions; + texts: string[]; + } + | { + type: "close"; + }; + +type LocalEmbeddingWorkerRequest = LocalEmbeddingWorkerRequestPayload & { id: number }; + +type LocalEmbeddingWorkerResponse = + | { + id: number; + ok: true; + value?: number[] | number[][]; + } + | { + id: number; + ok: false; + error: + | string + | { + message?: string; + code?: string; + }; + }; + +type PendingRequest = { + resolve: (value: number[] | number[][] | undefined) => void; + reject: (err: unknown) => void; + abort?: () => void; +}; + +function resolveDefaultWorkerScriptPath(): string { + const currentPath = fileURLToPath(import.meta.url); + const extension = path.extname(currentPath); + const currentName = path.basename(currentPath); + const sibling = + extension === ".ts" + ? "embeddings-worker-child.ts" + : currentName.startsWith("embeddings-worker.") + ? "embeddings-worker-child.js" + : "memory-core-local-embedding-worker.js"; + return path.join(path.dirname(currentPath), sibling); +} + +function serializeLocalEmbeddingOptions( + options: EmbeddingProviderOptions, +): EmbeddingProviderOptions { + return { + config: {}, + provider: "local", + model: options.model, + fallback: "none", + local: options.local, + }; +} + +function createWorkerExitError(code: number | null, signal: NodeJS.Signals | null): Error { + const detail = signal ? `signal ${signal}` : `exit code ${code ?? "unknown"}`; + return createLocalEmbeddingWorkerFailureError({ + message: `Local embedding worker exited unexpectedly (${detail})`, + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.exited, + reason: signal ? "signal" : "exit", + exitCode: code, + signal, + }); +} + +function createWorkerResponseError(error: LocalEmbeddingWorkerResponse & { ok: false }): Error { + if (typeof error.error === "object" && error.error) { + const message = error.error.message || "Local embedding worker failed"; + const workerError = new Error(message) as Error & { code?: string }; + if (error.error.code) { + workerError.code = error.error.code; + } + return workerError; + } + return new Error(error.error || "Local embedding worker failed"); +} + +const WORKER_UNSAFE_EXEC_ARGV_FLAGS = new Set(["--inspect", "--inspect-brk"]); + +const WORKER_UNSAFE_EXEC_ARGV_FLAGS_WITH_VALUE = new Set([ + "--eval", + "-e", + "--print", + "-p", + "--input-type", + "--inspect-port", +]); + +const WORKER_UNSAFE_EXEC_ARGV_OPTION_PREFIXES = [ + "--eval=", + "--print=", + "--input-type=", + "--inspect=", + "--inspect-brk=", + "--inspect-port=", +]; + +const WORKER_CLOSE_GRACE_MS = 250; + +function resolveWorkerExecArgv(): string[] { + const args: string[] = []; + let skipNext = false; + for (const arg of process.execArgv) { + if (skipNext) { + skipNext = false; + continue; + } + if (WORKER_UNSAFE_EXEC_ARGV_FLAGS.has(arg)) { + continue; + } + if (WORKER_UNSAFE_EXEC_ARGV_FLAGS_WITH_VALUE.has(arg)) { + skipNext = true; + continue; + } + if (WORKER_UNSAFE_EXEC_ARGV_OPTION_PREFIXES.some((prefix) => arg.startsWith(prefix))) { + continue; + } + args.push(arg); + } + return args; +} + +class LocalEmbeddingWorkerClient { + private child: ChildProcess | null = null; + private nextRequestId = 1; + private pending = new Map(); + + constructor(private readonly scriptPath: string) {} + + async initialize(options: EmbeddingProviderOptions): Promise { + await this.send({ type: "initialize", options }); + } + + async embedQuery( + options: EmbeddingProviderOptions, + text: string, + callOptions?: EmbeddingProviderCallOptions, + ): Promise { + const result = await this.send({ type: "embedQuery", options, text }, callOptions); + return Array.isArray(result) ? (result as number[]) : []; + } + + async embedBatch( + options: EmbeddingProviderOptions, + texts: string[], + callOptions?: EmbeddingProviderCallOptions, + ): Promise { + const result = await this.send({ type: "embedBatch", options, texts }, callOptions); + return Array.isArray(result) ? (result as number[][]) : []; + } + + async close(): Promise { + const child = this.child; + if (!child) { + return; + } + let timeout: NodeJS.Timeout | undefined; + const closeRequest = this.send({ type: "close" }).then(() => "closed" as const); + const closeTimeout = new Promise<"timeout">((resolve) => { + timeout = setTimeout(() => resolve("timeout"), WORKER_CLOSE_GRACE_MS); + timeout.unref?.(); + }); + try { + const result = await Promise.race([closeRequest, closeTimeout]); + if (result === "timeout") { + closeRequest.catch(() => {}); + } + } finally { + if (timeout) { + clearTimeout(timeout); + } + this.shutdownChild(); + } + } + + private ensureChild(): ChildProcess { + if (this.child?.connected) { + return this.child; + } + + const child = fork(this.scriptPath, [], { + execArgv: resolveWorkerExecArgv(), + serialization: "json", + stdio: ["ignore", "ignore", "ignore", "ipc"], + }); + child.on("message", (message) => this.handleMessage(message)); + child.on("exit", (code, signal) => { + if (this.child === child) { + this.child = null; + } + this.rejectPending(createWorkerExitError(code, signal)); + }); + child.on("error", (err) => { + if (this.child === child) { + this.child = null; + } + this.rejectPending( + createLocalEmbeddingWorkerFailureError({ + message: `Local embedding worker process failed: ${err.message}`, + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.processError, + reason: "process-error", + cause: err, + }), + ); + }); + this.child = child; + return child; + } + + private async send( + request: LocalEmbeddingWorkerRequestPayload, + options?: EmbeddingProviderCallOptions, + ): Promise { + options?.signal?.throwIfAborted(); + const child = this.ensureChild(); + const id = this.nextRequestId++; + const payload = { ...request, id } as LocalEmbeddingWorkerRequest; + return await new Promise((resolve, reject) => { + const pending: PendingRequest = { resolve, reject }; + if (options?.signal) { + const abort = () => { + this.pending.delete(id); + this.shutdownChild(); + reject(options.signal?.reason ?? new Error("Local embedding request aborted")); + }; + options.signal.addEventListener("abort", abort, { once: true }); + pending.abort = () => options.signal?.removeEventListener("abort", abort); + } + this.pending.set(id, pending); + child.send(payload, (err) => { + if (err) { + this.pending.delete(id); + pending.abort?.(); + reject( + createLocalEmbeddingWorkerFailureError({ + message: `Local embedding worker IPC failed: ${err.message}`, + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.ipcError, + reason: "ipc", + cause: err, + }), + ); + } + }); + }); + } + + private handleMessage(message: unknown): void { + const response = message as Partial; + if (typeof response.id !== "number") { + return; + } + const pending = this.pending.get(response.id); + if (!pending) { + return; + } + this.pending.delete(response.id); + pending.abort?.(); + if (response.ok) { + pending.resolve(response.value); + return; + } + pending.reject( + createWorkerResponseError(response as LocalEmbeddingWorkerResponse & { ok: false }), + ); + } + + private shutdownChild(): void { + const child = this.child; + this.child = null; + if (!child) { + return; + } + if (child.connected) { + child.disconnect(); + } + if (!child.killed) { + child.kill(); + } + } + + private rejectPending(err: unknown): void { + const pending = [...this.pending.values()]; + this.pending.clear(); + for (const entry of pending) { + entry.abort?.(); + entry.reject(err); + } + } +} + +export async function createLocalEmbeddingWorkerProvider( + options: EmbeddingProviderOptions, + runtimeOptions?: LocalEmbeddingProviderRuntimeOptions, +): Promise { + const modelPath = normalizeOptionalString(options.local?.modelPath) || DEFAULT_LOCAL_MODEL; + const workerOptions = serializeLocalEmbeddingOptions(options); + const client = new LocalEmbeddingWorkerClient( + runtimeOptions?.workerScriptPath ?? resolveDefaultWorkerScriptPath(), + ); + try { + await client.initialize(workerOptions); + } catch (err) { + await client.close().catch(() => {}); + throw err; + } + let closed = false; + + const throwIfClosed = () => { + if (closed) { + throw new Error("Local embedding provider has been closed"); + } + }; + + return { + id: "local", + model: modelPath, + embedQuery: async (text, callOptions) => { + throwIfClosed(); + return await client.embedQuery(workerOptions, text, callOptions); + }, + embedBatch: async (texts, callOptions) => { + throwIfClosed(); + return await client.embedBatch(workerOptions, texts, callOptions); + }, + close: async () => { + if (closed) { + return; + } + closed = true; + await client.close(); + }, + }; +} diff --git a/packages/memory-host-sdk/src/host/embeddings.test.ts b/packages/memory-host-sdk/src/host/embeddings.test.ts index c01343401f17..996790d53d84 100644 --- a/packages/memory-host-sdk/src/host/embeddings.test.ts +++ b/packages/memory-host-sdk/src/host/embeddings.test.ts @@ -1,5 +1,14 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { createLocalEmbeddingProvider, DEFAULT_LOCAL_MODEL } from "./embeddings.js"; +import { LOCAL_EMBEDDING_WORKER_ERROR_CODES } from "./embedding-worker-errors.js"; +import { createLocalEmbeddingWorkerProvider } from "./embeddings-worker.js"; +import { + createLocalEmbeddingProvider, + createLocalEmbeddingProviderInProcess, + DEFAULT_LOCAL_MODEL, +} from "./embeddings.js"; const nodeLlamaMock = vi.hoisted(() => ({ importNodeLlamaCpp: vi.fn(), @@ -39,10 +48,11 @@ function mockLocalEmbeddingRuntime(vector = new Float32Array([2.35, 3.45, 0.63, .fn() .mockResolvedValue({ getEmbeddingFor, dispose: disposeContext }); const loadModel = vi.fn().mockResolvedValue({ createEmbeddingContext, dispose: disposeModel }); + const getLlama = vi.fn(async () => ({ loadModel, dispose: disposeLlama })); const resolveModelFile = vi.fn(async (modelPath: string) => `/resolved/${modelPath}`); nodeLlamaMock.importNodeLlamaCpp.mockResolvedValue({ - getLlama: async () => ({ loadModel, dispose: disposeLlama }), + getLlama, resolveModelFile, LlamaLogLevel: { error: 0 }, } as never); @@ -52,6 +62,7 @@ function mockLocalEmbeddingRuntime(vector = new Float32Array([2.35, 3.45, 0.63, disposeContext, disposeLlama, disposeModel, + getLlama, getEmbeddingFor, loadModel, resolveModelFile, @@ -62,7 +73,7 @@ describe("local embedding provider", () => { it("normalizes local embeddings and resolves the default local model", async () => { const runtime = mockLocalEmbeddingRuntime(); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -92,7 +103,7 @@ describe("local embedding provider", () => { it("passes default contextSize (4096) to createEmbeddingContext when not configured", async () => { const runtime = mockLocalEmbeddingRuntime(); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -109,7 +120,7 @@ describe("local embedding provider", () => { it("passes configured contextSize to createEmbeddingContext", async () => { const runtime = mockLocalEmbeddingRuntime(); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -127,7 +138,7 @@ describe("local embedding provider", () => { it('passes "auto" contextSize to createEmbeddingContext when explicitly set', async () => { const runtime = mockLocalEmbeddingRuntime(); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -142,10 +153,43 @@ describe("local embedding provider", () => { ); }); + it("runs local batch embeddings sequentially", async () => { + const calls: string[] = []; + const firstGate = createDeferred<{ vector: Float32Array }>(); + const secondGate = createDeferred<{ vector: Float32Array }>(); + const getEmbeddingFor = vi.fn((text: string) => { + calls.push(text); + return text === "first" ? firstGate.promise : secondGate.promise; + }); + nodeLlamaMock.importNodeLlamaCpp.mockResolvedValue({ + getLlama: vi.fn(async () => ({ + loadModel: vi.fn(async () => ({ + createEmbeddingContext: vi.fn(async () => ({ getEmbeddingFor })), + })), + })), + resolveModelFile: vi.fn(async () => "/resolved/model.gguf"), + LlamaLogLevel: { error: 0 }, + } as never); + const provider = await createLocalEmbeddingProviderInProcess({ + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }); + + const batchPromise = provider.embedBatch(["first", "second"]); + await expect.poll(() => calls.join(",")).toBe("first"); + firstGate.resolve({ vector: new Float32Array([1, 0]) }); + await expect.poll(() => calls.join(",")).toBe("first,second"); + secondGate.resolve({ vector: new Float32Array([0, 1]) }); + + await expect(batchPromise).resolves.toHaveLength(2); + }); + it("trims explicit local model paths and cache directories", async () => { const runtime = mockLocalEmbeddingRuntime(new Float32Array([1, 0])); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -172,7 +216,7 @@ describe("local embedding provider", () => { it("disposes cached local llama resources when closed", async () => { const runtime = mockLocalEmbeddingRuntime(); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -199,7 +243,7 @@ describe("local embedding provider", () => { resolveModelFile: vi.fn(async (modelPath: string) => `/resolved/${modelPath}`), LlamaLogLevel: { error: 0 }, } as never); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -235,7 +279,7 @@ describe("local embedding provider", () => { }), LlamaLogLevel: { error: 0 }, } as never); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -274,7 +318,7 @@ describe("local embedding provider", () => { resolveModelFile: vi.fn(async () => "/resolved/model.gguf"), LlamaLogLevel: { error: 0 }, } as never); - const provider = await createLocalEmbeddingProvider({ + const provider = await createLocalEmbeddingProviderInProcess({ config: {} as never, provider: "local", model: "", @@ -291,4 +335,226 @@ describe("local embedding provider", () => { createContextGate.reject(new Error("context create aborted")); await expect(embedPromise).rejects.toThrow("context create aborted"); }); + + it("uses a worker process for the public local provider", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-local-embedding-worker-")); + const workerScript = path.join(tempDir, "worker.cjs"); + await fs.writeFile( + workerScript, + ` +process.on("message", (message) => { + if (message.type === "initialize") { + process.send({ id: message.id, ok: true }); + return; + } + if (message.type === "embedQuery") { + process.send({ id: message.id, ok: true, value: [1, 0] }); + return; + } + if (message.type === "embedBatch") { + process.send({ id: message.id, ok: true, value: message.texts.map(() => [0, 1]) }); + return; + } + process.send({ id: message.id, ok: true }); +}); +`, + "utf8", + ); + const provider = await createLocalEmbeddingWorkerProvider( + { + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }, + { workerScriptPath: workerScript }, + ); + + await expect(provider.embedQuery("hello")).resolves.toEqual([1, 0]); + await expect(provider.embedBatch(["a", "b"])).resolves.toEqual([ + [0, 1], + [0, 1], + ]); + await expect(provider.close?.()).resolves.toBeUndefined(); + }); + + it("terminates the worker when close runs behind a pending request", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-local-embedding-worker-")); + const workerScript = path.join(tempDir, "worker.cjs"); + const embedStartedPath = path.join(tempDir, "embed-started"); + await fs.writeFile( + workerScript, + ` +const fs = require("node:fs"); +const embedStartedPath = ${JSON.stringify(embedStartedPath)}; +let busy = false; + +process.on("message", (message) => { + if (busy) { + return; + } + if (message.type === "initialize") { + process.send({ id: message.id, ok: true }); + return; + } + if (message.type === "embedQuery") { + busy = true; + fs.writeFileSync(embedStartedPath, "1"); + } +}); +`, + "utf8", + ); + const provider = await createLocalEmbeddingWorkerProvider( + { + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }, + { workerScriptPath: workerScript }, + ); + + const embedPromise = provider.embedQuery("stuck"); + const embedError = embedPromise.then( + () => undefined, + (err) => err, + ); + await expect + .poll(async () => { + try { + await fs.access(embedStartedPath); + return true; + } catch { + return false; + } + }) + .toBe(true); + + const closePromise = provider.close?.() ?? Promise.resolve(); + const closeResult = await Promise.race([ + closePromise.then(() => "closed" as const), + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 1_000)), + ]); + + expect(closeResult).toBe("closed"); + await expect(embedError).resolves.toMatchObject({ + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.exited, + }); + }); + + it("does not pass inline-source or inspector exec args to the file-backed worker", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-local-embedding-worker-")); + const workerScript = path.join(tempDir, "worker.cjs"); + await fs.writeFile( + workerScript, + ` +process.on("message", (message) => { + if (message.type === "initialize" || message.type === "close") { + process.send({ id: message.id, ok: true }); + return; + } + process.send({ id: message.id, ok: true, value: [process.execArgv.length] }); +}); +`, + "utf8", + ); + const originalExecArgv = [...process.execArgv]; + let provider: Awaited> | undefined; + try { + process.execArgv.splice( + 0, + process.execArgv.length, + "--eval", + "setInterval(() => {}, 1000)", + "--print", + "1 + 1", + "--input-type=module", + "--inspect-brk=127.0.0.1:0", + "--inspect-port", + "0", + ); + provider = await createLocalEmbeddingWorkerProvider( + { + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }, + { workerScriptPath: workerScript }, + ); + await expect(provider.embedQuery("hello")).resolves.toEqual([0]); + } finally { + process.execArgv.splice(0, process.execArgv.length, ...originalExecArgv); + await provider?.close?.(); + } + }); + + it("reports worker initialization failures during provider creation", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-local-embedding-worker-")); + const workerScript = path.join(tempDir, "worker.cjs"); + await fs.writeFile( + workerScript, + ` +process.on("message", (message) => { + process.send({ + id: message.id, + ok: false, + error: { message: "Cannot find package 'node-llama-cpp'", code: "ERR_MODULE_NOT_FOUND" }, + }); +}); +`, + "utf8", + ); + + try { + await createLocalEmbeddingWorkerProvider( + { + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }, + { workerScriptPath: workerScript }, + ); + throw new Error("expected local embedding provider creation to fail"); + } catch (err) { + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toBe("Cannot find package 'node-llama-cpp'"); + expect((err as Error & { code?: string }).code).toBe("ERR_MODULE_NOT_FOUND"); + } + }); + + it("reports worker exits with structured failure codes", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-local-embedding-worker-")); + const workerScript = path.join(tempDir, "worker.cjs"); + await fs.writeFile( + workerScript, + ` +process.on("message", (message) => { + if (message.type === "initialize") { + process.send({ id: message.id, ok: true }); + return; + } + process.exit(134); +}); +`, + "utf8", + ); + const provider = await createLocalEmbeddingWorkerProvider( + { + config: {} as never, + provider: "local", + model: "", + fallback: "none", + }, + { workerScriptPath: workerScript }, + ); + + await expect(provider.embedQuery("hello")).rejects.toMatchObject({ + code: LOCAL_EMBEDDING_WORKER_ERROR_CODES.exited, + reason: "exit", + exitCode: 134, + }); + }); }); diff --git a/packages/memory-host-sdk/src/host/embeddings.ts b/packages/memory-host-sdk/src/host/embeddings.ts index e5449dda904a..c375249b17c8 100644 --- a/packages/memory-host-sdk/src/host/embeddings.ts +++ b/packages/memory-host-sdk/src/host/embeddings.ts @@ -1,5 +1,6 @@ import { DEFAULT_LOCAL_MODEL } from "./embedding-defaults.js"; import { sanitizeAndNormalizeEmbedding } from "./embedding-vectors.js"; +import { createLocalEmbeddingWorkerProvider } from "./embeddings-worker.js"; import type { EmbeddingProvider, EmbeddingProviderOptions } from "./embeddings.types.js"; import { importNodeLlamaCpp, @@ -24,6 +25,10 @@ export type { export { DEFAULT_LOCAL_MODEL } from "./embedding-defaults.js"; +export type LocalEmbeddingProviderRuntimeOptions = { + workerScriptPath?: string; +}; + async function disposeResources( resources: Array, ): Promise { @@ -42,6 +47,12 @@ async function disposeResources( export async function createLocalEmbeddingProvider( options: EmbeddingProviderOptions, +): Promise { + return await createLocalEmbeddingWorkerProvider(options); +} + +export async function createLocalEmbeddingProviderInProcess( + options: EmbeddingProviderOptions, ): Promise { const modelPath = normalizeOptionalString(options.local?.modelPath) || DEFAULT_LOCAL_MODEL; const modelCacheDir = normalizeOptionalString(options.local?.modelCacheDir); @@ -85,7 +96,9 @@ export async function createLocalEmbeddingProvider( initAbortController = abortController; try { if (!llama) { - const nextLlama = await getLlama({ logLevel: LlamaLogLevel.error }); + const nextLlama = await getLlama({ + logLevel: LlamaLogLevel.error, + }); llama = await disposeAndThrowIfClosed(nextLlama); } if (!embeddingModel) { @@ -138,14 +151,13 @@ export async function createLocalEmbeddingProvider( const ctx = await ensureContext(); throwIfClosed(); options?.signal?.throwIfAborted(); - const embeddings = await Promise.all( - texts.map(async (text) => { - throwIfClosed(); - options?.signal?.throwIfAborted(); - const embedding = await ctx.getEmbeddingFor(text); - return sanitizeAndNormalizeEmbedding(Array.from(embedding.vector)); - }), - ); + const embeddings: number[][] = []; + for (const text of texts) { + throwIfClosed(); + options?.signal?.throwIfAborted(); + const embedding = await ctx.getEmbeddingFor(text); + embeddings.push(sanitizeAndNormalizeEmbedding(Array.from(embedding.vector))); + } return embeddings; }, close: async () => { diff --git a/src/config/config.schema-regressions.test.ts b/src/config/config.schema-regressions.test.ts index 163c18a1ad98..2f2b551ff876 100644 --- a/src/config/config.schema-regressions.test.ts +++ b/src/config/config.schema-regressions.test.ts @@ -56,6 +56,23 @@ describe("config schema regressions", () => { expect(res.ok).toBe(true); }); + it("rejects local memorySearch GPU policy", () => { + const res = validateConfigObject({ + agents: { + defaults: { + memorySearch: { + provider: "local", + local: { + gpu: "cpu", + }, + }, + }, + }, + }); + + expect(res.ok).toBe(false); + }); + it("accepts memorySearch.qmd.extraCollections", () => { const res = validateConfigObject({ agents: { diff --git a/src/plugins/memory-embedding-providers.ts b/src/plugins/memory-embedding-providers.ts index 83ff0db0d9ef..5ee1831a44f0 100644 --- a/src/plugins/memory-embedding-providers.ts +++ b/src/plugins/memory-embedding-providers.ts @@ -62,6 +62,7 @@ export type MemoryEmbeddingProviderCreateOptions = { local?: { modelPath?: string; modelCacheDir?: string; + contextSize?: number | "auto"; }; outputDimensionality?: number; taskType?: diff --git a/tsdown.config.ts b/tsdown.config.ts index 425ead1324ff..06941873b5dc 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -323,6 +323,8 @@ function buildUnifiedDistEntries(): Record { "plugin-sdk/qa-runtime": "src/plugin-sdk/qa-runtime.ts", } : {}), + "memory-core-local-embedding-worker": + "packages/memory-host-sdk/src/host/embeddings-worker-child.ts", ...listBundledPluginEntrySources(rootBundledPluginBuildEntries), ...bundledHookEntries, };