diff --git a/app/src/components/settings/panels/AIPanel.tsx b/app/src/components/settings/panels/AIPanel.tsx index c89438030..3745e4a41 100644 --- a/app/src/components/settings/panels/AIPanel.tsx +++ b/app/src/components/settings/panels/AIPanel.tsx @@ -45,8 +45,10 @@ import { openhumanHeartbeatSettingsSet, openhumanHeartbeatTickNow, } from '../../../utils/tauriCommands/heartbeat'; +import { ConfirmationModal } from '../../intelligence/ConfirmationModal'; import SettingsHeader from '../components/SettingsHeader'; import { useSettingsNavigation } from '../hooks/useSettingsNavigation'; +import { useReembedBackfillModal } from './useReembedBackfillModal'; // ───────────────────────────────────────────────────────────────────────────── // Types @@ -335,7 +337,9 @@ function useAISettings() { [saved] ); - const save = useCallback(async () => { + // Returns true only when persistence actually succeeded, so callers + // (e.g. the #1574 re-embed-status check) don't act on a failed save. + const save = useCallback(async (): Promise => { try { // Defensive verification at global-Save time. Each provider that is new // or whose endpoint changed since the last saved snapshot is re-probed @@ -358,14 +362,16 @@ function useAISettings() { } catch (probeErr) { const msg = probeErr instanceof Error ? probeErr.message : String(probeErr); setError(`Could not reach ${p.label}: ${msg}. Settings were not saved.`); - return; + return false; } } await persist(draft); + return true; } catch (err) { const message = err instanceof Error ? err.message : 'Failed to save AI settings'; setError(message); + return false; } }, [saved, draft, persist]); @@ -1974,6 +1980,9 @@ const AIPanel = ({ embedded = false }: AIPanelProps = {}) => { const { navigateBack, breadcrumbs } = useSettingsNavigation(); const { saved, draft, setDraft, isDirty, save, persist, discard, loading, error, reload } = useAISettings(); + // #1574 §4b: advisory re-embed modal, driven by the backend status RPC. + // Logic lives in a unit-testable hook (see useReembedBackfillModal). + const { reembed, handleSave, dismissReembed } = useReembedBackfillModal(save); const ollama = useOllamaStatus(); const installed = useInstalledModels(ollama.snapshot); const [editing, setEditing] = useState(null); @@ -2220,11 +2229,27 @@ const AIPanel = ({ embedded = false }: AIPanelProps = {}) => { void save()} + onSave={() => void handleSave()} onDiscard={discard} /> )} + + {editing && ( ({ memoryTreeBackfillStatus: vi.fn() })); + +const status = vi.mocked(memoryTreeBackfillStatus); + +beforeEach(() => { + vi.clearAllMocks(); +}); +afterEach(() => { + vi.useRealTimers(); +}); + +describe('useReembedBackfillModal (#1574 §4b)', () => { + it('does not check status or open the modal when save fails', async () => { + const save = vi.fn().mockResolvedValue(false); + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + + expect(save).toHaveBeenCalledTimes(1); + expect(status).not.toHaveBeenCalled(); + expect(result.current.reembed).toEqual({ open: false, pending: 0 }); + }); + + it('opens the modal when save succeeds and a backfill is in progress', async () => { + const save = vi.fn().mockResolvedValue(true); + status.mockResolvedValue({ in_progress: true, pending_jobs: 7 }); + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + + expect(result.current.reembed).toEqual({ open: true, pending: 7 }); + }); + + it('stays closed when save succeeds but no backfill is in progress', async () => { + const save = vi.fn().mockResolvedValue(true); + status.mockResolvedValue({ in_progress: false, pending_jobs: 0 }); + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + + expect(result.current.reembed.open).toBe(false); + }); + + it('swallows a status-check error and stays closed', async () => { + const save = vi.fn().mockResolvedValue(true); + status.mockRejectedValue(new Error('rpc down')); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + + expect(result.current.reembed.open).toBe(false); + expect(warn).toHaveBeenCalled(); + warn.mockRestore(); + }); + + it('polls while open: updates pending, then auto-closes when the backfill drains', async () => { + vi.useFakeTimers(); + const save = vi.fn().mockResolvedValue(true); + status + .mockResolvedValueOnce({ in_progress: true, pending_jobs: 9 }) // handleSave open + .mockResolvedValueOnce({ in_progress: true, pending_jobs: 4 }) // poll #1 + .mockResolvedValueOnce({ in_progress: false, pending_jobs: 0 }); // poll #2 → close + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + expect(result.current.reembed).toEqual({ open: true, pending: 9 }); + + await act(async () => { + await vi.advanceTimersByTimeAsync(2000); + }); + expect(result.current.reembed.pending).toBe(4); + + await act(async () => { + await vi.advanceTimersByTimeAsync(2000); + }); + expect(result.current.reembed.open).toBe(false); + }); + + it('dismissReembed closes the modal (background backfill unaffected)', async () => { + const save = vi.fn().mockResolvedValue(true); + status.mockResolvedValue({ in_progress: true, pending_jobs: 2 }); + const { result } = renderHook(() => useReembedBackfillModal(save)); + + await act(async () => { + await result.current.handleSave(); + }); + expect(result.current.reembed.open).toBe(true); + + act(() => { + result.current.dismissReembed(); + }); + expect(result.current.reembed).toEqual({ open: false, pending: 0 }); + }); +}); diff --git a/app/src/components/settings/panels/useReembedBackfillModal.ts b/app/src/components/settings/panels/useReembedBackfillModal.ts new file mode 100644 index 000000000..ec5561b9c --- /dev/null +++ b/app/src/components/settings/panels/useReembedBackfillModal.ts @@ -0,0 +1,84 @@ +/** + * #1574 §4b: advisory re-embed-backfill modal state, driven entirely by the + * backend status RPC. + * + * After a settings save, the core's coverage-gated `ensure_reembed_backfill` + * has already decided whether the embedder change needs a re-embed; this hook + * just surfaces its progress (no fragile frontend "did the embedder change" + * detection). Extracted from `AIPanel` so the logic is unit-testable in + * isolation rather than only via a full 2000-line component render. + */ +import { useCallback, useEffect, useState } from 'react'; + +import { memoryTreeBackfillStatus } from '../../../utils/tauriCommands/memoryTree'; + +export interface ReembedState { + open: boolean; + pending: number; +} + +export interface ReembedBackfillModal { + reembed: ReembedState; + /** Wrap a settings-save: persist, then (only on success) surface re-embed progress. */ + handleSave: () => Promise; + /** Close the advisory modal (re-embed continues in the background). */ + dismissReembed: () => void; +} + +const POLL_MS = 2000; + +/** + * @param save Persists settings; resolves `true` only on a successful write + * (a failed save did not change the embedder, so nothing to surface). + */ +export function useReembedBackfillModal(save: () => Promise): ReembedBackfillModal { + const [reembed, setReembed] = useState({ open: false, pending: 0 }); + + const handleSave = useCallback(async () => { + const ok = await save(); + if (!ok) return; + try { + const st = await memoryTreeBackfillStatus(); + if (st.in_progress) { + setReembed({ open: true, pending: st.pending_jobs }); + } + } catch (e) { + console.warn('[ai-panel] backfill status check failed', e); + } + }, [save]); + + const dismissReembed = useCallback(() => setReembed({ open: false, pending: 0 }), []); + + useEffect(() => { + if (!reembed.open) return; + let cancelled = false; + // Serialize polls — if a status call takes >POLL_MS, skip the next tick + // rather than overlapping requests. + let inFlight = false; + const id = window.setInterval(() => { + if (inFlight) return; + inFlight = true; + void (async () => { + try { + const st = await memoryTreeBackfillStatus(); + if (cancelled) return; + if (!st.in_progress) { + setReembed({ open: false, pending: 0 }); + } else { + setReembed(r => ({ ...r, pending: st.pending_jobs })); + } + } catch (e) { + console.warn('[ai-panel] backfill poll failed', e); + } finally { + inFlight = false; + } + })(); + }, POLL_MS); + return () => { + cancelled = true; + window.clearInterval(id); + }; + }, [reembed.open]); + + return { reembed, handleSave, dismissReembed }; +} diff --git a/app/src/utils/tauriCommands/memoryTree.test.ts b/app/src/utils/tauriCommands/memoryTree.test.ts index 8ac1df81f..68a5cfd2c 100644 --- a/app/src/utils/tauriCommands/memoryTree.test.ts +++ b/app/src/utils/tauriCommands/memoryTree.test.ts @@ -8,6 +8,7 @@ import { beforeEach, describe, expect, type Mock, test, vi } from 'vitest'; import { callCoreRpc } from '../../services/coreRpcClient'; import { + memoryTreeBackfillStatus, memoryTreeChunkScore, memoryTreeDeleteChunk, memoryTreeEntityIndexFor, @@ -355,3 +356,27 @@ describe('memoryTreeGraphExport', () => { expect(out.edges).toHaveLength(1); }); }); + +describe('memoryTreeBackfillStatus', () => { + test('dispatches openhuman.memory_tree_memory_backfill_status and unwraps', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ + result: { in_progress: true, pending_jobs: 3 }, + logs: ['memory_tree: backfill_status in_progress=true pending=3'], + }); + + const out = await memoryTreeBackfillStatus(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.memory_tree_memory_backfill_status', + }); + expect(out.in_progress).toBe(true); + expect(out.pending_jobs).toBe(3); + }); + + test('handles bare-value responses (no logs envelope)', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ in_progress: false, pending_jobs: 0 }); + const out = await memoryTreeBackfillStatus(); + expect(out.in_progress).toBe(false); + expect(out.pending_jobs).toBe(0); + }); +}); diff --git a/app/src/utils/tauriCommands/memoryTree.ts b/app/src/utils/tauriCommands/memoryTree.ts index f01e4ff26..adc83a400 100644 --- a/app/src/utils/tauriCommands/memoryTree.ts +++ b/app/src/utils/tauriCommands/memoryTree.ts @@ -603,3 +603,31 @@ export async function memoryTreeGraphExport( ); return out; } + +/** + * #1574 §4b: per-model embedding re-embed backfill status. The AI settings + * panel polls this after an embedder change to warn that semantic recall + * is reduced until the new embedding space is fully re-embedded, and to + * dismiss the warning once the chain drains. Backed by + * `openhuman.memory_tree_memory_backfill_status`. + */ +export interface BackfillStatus { + /** True while a re-embed backfill still has work pending. */ + in_progress: boolean; + /** Count of `reembed_backfill` jobs in ready/running state. */ + pending_jobs: number; +} + +export async function memoryTreeBackfillStatus(): Promise { + console.debug('[memory-tree-rpc] memoryTreeBackfillStatus: entry'); + const resp = await callCoreRpc>({ + method: 'openhuman.memory_tree_memory_backfill_status', + }); + const out = unwrapResult(resp); + console.debug( + '[memory-tree-rpc] memoryTreeBackfillStatus: exit in_progress=%s pending=%d', + out.in_progress, + out.pending_jobs + ); + return out; +} diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index cdab7fbb7..ffa22fb7e 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -386,45 +386,31 @@ mod tests { // construction time, so a mid-session `composio.mode` toggle is // honoured on the very next per-action execute. - #[tokio::test] - async fn factory_routes_through_backend_when_mode_is_backend() { - // Default `Config` has `composio.mode = "backend"`. Without a - // stored backend session token the factory returns - // `Err("no backend session ...")`. Assert that the error text - // points at the backend code path (not direct-mode or staging- - // api), confirming the routing branch. - // - // Production `.execute(..)` calls `load_config_with_timeout()` - // per call which reads from `~/.openhuman/config.toml` (or the - // workspace pointed at by `OPENHUMAN_WORKSPACE`). To isolate - // the test from the dev's real config we hold `TEST_ENV_LOCK`, - // point `OPENHUMAN_WORKSPACE` at a tempdir, and persist the - // test's `Config` to that tempdir's `config.toml` before - // invoking the tool. - use crate::openhuman::config::TEST_ENV_LOCK; - let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - + // These two tests assert the *factory routing decision* by mode. They + // call `create_composio_client(&Config)` directly — the pure routing + // function — instead of going through `tool.execute()`, which reloads + // config via `load_config_with_timeout()` (reads `OPENHUMAN_WORKSPACE`) + // and was therefore subject to a parallel-test env-var race: another + // non-`TEST_ENV_LOCK` test mutating `OPENHUMAN_WORKSPACE` in the await + // window flipped the reloaded config, intermittently failing + // `factory_routes_through_direct_when_mode_is_direct`. The factory reads + // mode + session purely from the passed `Config` (the auth-store path is + // derived from the config's own paths, not the env var), so pointing + // those at a fresh tempdir is fully isolated, deterministic, and needs + // no env mutation / `TEST_ENV_LOCK` / async. + #[test] + fn factory_routes_through_backend_when_mode_is_backend() { let tmp = tempfile::tempdir().expect("tempdir"); - unsafe { - std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path()); - } - - let mut config = Config::default(); + let mut config = Config::default(); // composio.mode defaults to "backend" config.config_path = tmp.path().join("config.toml"); config.workspace_dir = tmp.path().join("workspace"); - config.save().await.expect("save fake config to disk"); - let tool = ComposioActionTool::new( - Arc::new(config), - "GMAIL_FETCH_EMAILS".to_string(), - "read-shaped slug so sandbox/scope gates don't short-circuit \ - the dispatch site" - .to_string(), - None, - ); - let result = tool.execute(serde_json::json!({})).await.unwrap(); - assert!(result.is_error, "no backend session must error"); - let msg = error_text(&result); + // `ComposioClientKind` isn't `Debug`, so match rather than + // `expect_err` (which would need to format the unexpected `Ok`). + let msg = match crate::openhuman::composio::client::create_composio_client(&config) { + Ok(_) => panic!("backend mode with no session must error, but a client resolved"), + Err(e) => e.to_string(), + }; assert!( msg.contains("backend") || msg.contains("session"), "expected backend-mode session error, got: {msg}" @@ -433,59 +419,29 @@ mod tests { !msg.contains("direct mode"), "backend-mode failure must not surface direct-mode artifacts: {msg}" ); - - unsafe { - std::env::remove_var("OPENHUMAN_WORKSPACE"); - } } - #[tokio::test] - async fn factory_routes_through_direct_when_mode_is_direct() { - // Direct-mode config with an inline api_key — factory resolves - // to the `Direct` variant. The downstream call will fail when - // it attempts to hit `backend.composio.dev` from the unit test - // sandbox, but the error must come from the direct path, not - // a backend session lookup. - // - // Production `.execute(..)` calls `load_config_with_timeout()` - // per call which reads from disk — see the matching note on - // `factory_routes_through_backend_when_mode_is_backend`. - use crate::openhuman::config::TEST_ENV_LOCK; - let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - + #[test] + fn factory_routes_through_direct_when_mode_is_direct() { let tmp = tempfile::tempdir().expect("tempdir"); - unsafe { - std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path()); - } - let mut config = Config::default(); config.config_path = tmp.path().join("config.toml"); config.workspace_dir = tmp.path().join("workspace"); config.composio.mode = crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT.to_string(); config.composio.api_key = Some("test-direct-key".to_string()); - config.save().await.expect("save fake config to disk"); - let tool = ComposioActionTool::new( - Arc::new(config), - "GMAIL_FETCH_EMAILS".to_string(), - "read-shaped slug".to_string(), - None, - ); - let result = tool.execute(serde_json::json!({})).await.unwrap(); - // Direct-mode resolve succeeds → no `factory failed` error. - // The error (if any) will come from the downstream HTTP call, - // which is fine — we just need to confirm the dispatch routed - // through the direct branch rather than the backend branch. - let msg = error_text(&result); + // Direct mode + an api key must resolve to the Direct variant — + // never the backend branch. (Deterministic: pure factory call, no + // env / reload / await; see the note on the backend test.) + let kind = crate::openhuman::composio::client::create_composio_client(&config) + .expect("direct mode with an api key must resolve"); assert!( - !msg.contains("no backend session") && !msg.contains("staging-api"), - "direct-mode dispatch must not leak backend session / staging-api \ - artifacts: {msg}" + matches!( + kind, + crate::openhuman::composio::client::ComposioClientKind::Direct(_) + ), + "direct-mode config must route to the Direct client, not backend" ); - - unsafe { - std::env::remove_var("OPENHUMAN_WORKSPACE"); - } } #[tokio::test] diff --git a/src/openhuman/config/ops.rs b/src/openhuman/config/ops.rs index 126b0ad68..644d9242b 100644 --- a/src/openhuman/config/ops.rs +++ b/src/openhuman/config/ops.rs @@ -496,6 +496,13 @@ pub async fn apply_model_settings( } config.save().await.map_err(|e| e.to_string())?; + // #1574 §4: the AIPanel workload matrix changes the embedder via THIS + // (model-settings) path — `embeddings_provider` above — not the + // memory-settings path. Trigger the same idempotent re-embed backfill + // so a UI embedder switch recovers prior memory under the new + // signature. Coverage-gated + non-fatal: if the active signature did + // not actually change, this enqueues nothing. + crate::openhuman::memory::tree::jobs::ensure_reembed_backfill(config); let snapshot = snapshot_config_json(config)?; Ok(RpcOutcome::new( snapshot, @@ -539,6 +546,13 @@ pub async fn apply_memory_settings( } } config.save().await.map_err(|e| e.to_string())?; + // #1574 §4: the embedder may have just changed (provider/model/dims). + // Ensure a re-embed backfill chain exists for the new active signature + // so prior memory becomes retrievable again instead of silently going + // dark. Idempotent + non-fatal (covered space enqueues nothing; errors + // are logged, never fail the settings save). §7's migration is + // one-shot so it does not cover a later switch — this does. + crate::openhuman::memory::tree::jobs::ensure_reembed_backfill(config); let snapshot = snapshot_config_json(config)?; Ok(RpcOutcome::new( snapshot, diff --git a/src/openhuman/embeddings/mod.rs b/src/openhuman/embeddings/mod.rs index 783b24e36..e9c7130e8 100644 --- a/src/openhuman/embeddings/mod.rs +++ b/src/openhuman/embeddings/mod.rs @@ -28,7 +28,7 @@ pub use factory::{ pub use noop::NoopEmbedding; pub use ollama::{OllamaEmbedding, DEFAULT_OLLAMA_DIMENSIONS, DEFAULT_OLLAMA_MODEL}; pub use openai::OpenAiEmbedding; -pub use provider_trait::EmbeddingProvider; +pub use provider_trait::{format_embedding_signature, EmbeddingProvider}; pub use store::{bytes_to_vec, cosine_similarity, vec_to_bytes, SearchResult, VectorStore}; #[cfg(test)] diff --git a/src/openhuman/embeddings/provider_trait.rs b/src/openhuman/embeddings/provider_trait.rs index 6c194831b..262e1045f 100644 --- a/src/openhuman/embeddings/provider_trait.rs +++ b/src/openhuman/embeddings/provider_trait.rs @@ -2,6 +2,18 @@ use async_trait::async_trait; +/// Formats the canonical embedding-space signature string. +/// +/// This is the **single source of truth** for the signature format. Both the +/// live-provider [`EmbeddingProvider::signature`] and the config-derived +/// `active_embedding_signature` (memory store factories) route through here so +/// a signature computed from configuration is byte-identical to one computed +/// from an instantiated provider. Drift between the two would silently split +/// one embedding space into two (#1574). +pub fn format_embedding_signature(name: &str, model_id: &str, dims: usize) -> String { + format!("provider={name};model={model_id};dims={dims}") +} + /// Interface for embedding providers that convert text into numerical vectors. #[async_trait] pub trait EmbeddingProvider: Send + Sync { @@ -20,12 +32,7 @@ pub trait EmbeddingProvider: Send + Sync { /// comparable with newly-generated vectors and should be stored / queried /// separately by follow-up storage migrations. fn signature(&self) -> String { - format!( - "provider={};model={};dims={}", - self.name(), - self.model_id(), - self.dimensions() - ) + format_embedding_signature(self.name(), self.model_id(), self.dimensions()) } /// Generates embeddings for a batch of strings. diff --git a/src/openhuman/memory/store/factories.rs b/src/openhuman/memory/store/factories.rs index 8ce4a8818..08b50ba57 100644 --- a/src/openhuman/memory/store/factories.rs +++ b/src/openhuman/memory/store/factories.rs @@ -14,8 +14,8 @@ use std::sync::Arc; use crate::openhuman::config::{EmbeddingRouteConfig, MemoryConfig, StorageProviderConfig}; use crate::openhuman::embeddings::{ - self, EmbeddingProvider, DEFAULT_CLOUD_EMBEDDING_DIMENSIONS, DEFAULT_CLOUD_EMBEDDING_MODEL, - DEFAULT_OLLAMA_DIMENSIONS, DEFAULT_OLLAMA_MODEL, + self, format_embedding_signature, EmbeddingProvider, DEFAULT_CLOUD_EMBEDDING_DIMENSIONS, + DEFAULT_CLOUD_EMBEDDING_MODEL, DEFAULT_OLLAMA_DIMENSIONS, DEFAULT_OLLAMA_MODEL, }; use crate::openhuman::memory::store::agentmemory::AgentMemoryBackend; use crate::openhuman::memory::store::unified::UnifiedMemory; @@ -202,6 +202,25 @@ pub fn effective_embedding_settings( ) } +/// The **active embedding signature** — the canonical key every per-model +/// sidecar read/write is scoped by (#1574). +/// +/// Derived from [`effective_embedding_settings`] (the *intended*, non-probed +/// selection) — deliberately **not** [`effective_embedding_settings_probed`]. +/// A transient Ollama-down fallback to cloud must never silently redefine the +/// signature: that would re-key every read at a different space and trigger a +/// spurious full re-embed on the next cold-Ollama launch (spec §3 oscillation +/// guard). The string is produced by [`format_embedding_signature`], the same +/// formatter [`EmbeddingProvider::signature`] uses, so a config-derived +/// signature is byte-identical to a live provider's. +pub fn active_embedding_signature( + memory: &MemoryConfig, + local_embedding_model: Option<&str>, +) -> String { + let (provider, model, dims) = effective_embedding_settings(memory, local_embedding_model); + format_embedding_signature(&provider, &model, dims) +} + /// Async, health-checked variant of [`effective_embedding_settings`]. /// /// If the intended provider is `"ollama"` but the daemon doesn't respond at @@ -555,6 +574,43 @@ mod tests { ); } + /// #1574 invariant: a config-derived `active_embedding_signature` MUST be + /// byte-identical to the live provider's `.signature()` for the same + /// (provider, model, dims). Drift here silently splits one embedding space + /// into two — copied/queried vectors would never match. + #[test] + fn active_signature_matches_live_provider_signature() { + for local in [None, Some("nomic-embed-text:latest"), Some("bge-m3")] { + let mem = MemoryConfig::default(); + let (provider, model, dims) = effective_embedding_settings(&mem, local); + let live = embeddings::create_embedding_provider(&provider, &model, dims) + .expect("provider builds for test triple"); + assert_eq!( + active_embedding_signature(&mem, local), + live.signature(), + "config-derived signature must equal live provider signature (local={local:?})" + ); + } + } + + #[test] + fn active_signature_ignores_probe_fallback() { + // active_embedding_signature keys off the *intended* selection + // (effective_embedding_settings), NOT the health-checked variant — so + // a transient Ollama-down fallback can't flip it to cloud. The dim is + // base/config-dependent (not what this test pins); the provider+model + // staying the intended ollama/bge-m3 is the probe-stability property. + let mem = MemoryConfig::default(); + let sig = active_embedding_signature(&mem, Some("bge-m3")); + assert!( + sig.starts_with("provider=ollama;model=bge-m3;dims="), + "intended local selection must survive (no cloud fallback); got {sig}" + ); + // And it must equal the non-probed settings, formatted identically. + let (p, m, d) = effective_embedding_settings(&mem, Some("bge-m3")); + assert_eq!(sig, format_embedding_signature(&p, &m, d)); + } + #[test] fn effective_memory_backend_name_always_returns_namespace() { assert_eq!(effective_memory_backend_name("sqlite", None), "namespace"); diff --git a/src/openhuman/memory/store/mod.rs b/src/openhuman/memory/store/mod.rs index 740563a06..ba0366f88 100644 --- a/src/openhuman/memory/store/mod.rs +++ b/src/openhuman/memory/store/mod.rs @@ -28,8 +28,8 @@ pub use agentmemory::{agentmemory_default_url, AgentMemoryBackend, DEFAULT_AGENT pub use client::{MemoryClient, MemoryClientRef, MemoryState}; pub use factories::{ - create_memory, create_memory_for_migration, create_memory_with_local_ai, - create_memory_with_storage, create_memory_with_storage_and_routes, + active_embedding_signature, create_memory, create_memory_for_migration, + create_memory_with_local_ai, create_memory_with_storage, create_memory_with_storage_and_routes, effective_embedding_settings, effective_embedding_settings_probed, effective_memory_backend_name, }; diff --git a/src/openhuman/memory/tree/jobs/handlers/mod.rs b/src/openhuman/memory/tree/jobs/handlers/mod.rs index ed0edfa93..d58766930 100644 --- a/src/openhuman/memory/tree/jobs/handlers/mod.rs +++ b/src/openhuman/memory/tree/jobs/handlers/mod.rs @@ -18,7 +18,8 @@ use crate::openhuman::memory::tree::content_store::{ use crate::openhuman::memory::tree::jobs::store; use crate::openhuman::memory::tree::jobs::types::{ AppendBufferPayload, AppendTarget, DigestDailyPayload, ExtractChunkPayload, FlushStalePayload, - Job, JobKind, JobOutcome, NewJob, NodeRef, SealPayload, TopicRoutePayload, + Job, JobKind, JobOutcome, NewJob, NodeRef, ReembedBackfillPayload, SealPayload, + TopicRoutePayload, }; use crate::openhuman::memory::tree::score; use crate::openhuman::memory::tree::score::embed::{build_embedder_from_config, pack_checked}; @@ -45,6 +46,7 @@ pub async fn handle_job(config: &Config, job: &Job) -> Result { JobKind::TopicRoute => handle_topic_route(config, job).await, JobKind::DigestDaily => handle_digest_daily(config, job).await, JobKind::FlushStale => handle_flush_stale(config, job).await, + JobKind::ReembedBackfill => handle_reembed_backfill(config, job).await, } } @@ -74,7 +76,7 @@ async fn handle_extract(config: &Config, job: &Job) -> Result { let scoring_cfg = score::ScoringConfig::from_config(config); let result = score::score_chunk(&chunk_with_body, &scoring_cfg).await?; - let packed_embedding = if result.kept { + let chunk_embedding: Option> = if result.kept { let embedder = build_embedder_from_config(config).context("build embedder in extract handler")?; // Reuse the body already read — avoid a second disk read. @@ -82,10 +84,13 @@ async fn handle_extract(config: &Config, job: &Job) -> Result { .embed(&body) .await .with_context(|| format!("embed chunk_id={} in extract handler", chunk.id))?; - Some( - pack_checked(&vector) - .with_context(|| format!("pack embedding for chunk_id={}", chunk.id))?, - ) + // Preserve the pre-cutover dimension guard (the job fails fast on a + // misconfigured embedder) even though #1574 no longer persists the + // packed blob to the legacy `mem_tree_chunks.embedding` column — + // the vector now goes to the per-model sidecar instead. + pack_checked(&vector) + .with_context(|| format!("validate embedding dims for chunk_id={}", chunk.id))?; + Some(vector) } else { None }; @@ -117,6 +122,9 @@ async fn handle_extract(config: &Config, job: &Job) -> Result { None }; + // #1574: resolve the active embedding signature once (probe-stable, + // config-derived) so the sidecar write below is keyed correctly. + let active_sig = chunk_store::tree_active_signature(config); let (did_enqueue_source, did_enqueue_route) = chunk_store::with_connection(config, |conn| { let tx = conn.unchecked_transaction()?; score::persist_score_tx( @@ -129,15 +137,24 @@ async fn handle_extract(config: &Config, job: &Job) -> Result { if result.kept { tx.execute( "UPDATE mem_tree_chunks - SET embedding = ?1, - lifecycle_status = ?2 - WHERE id = ?3", - rusqlite::params![ - packed_embedding, - chunk_store::CHUNK_STATUS_ADMITTED, - chunk.id, - ], + SET lifecycle_status = ?1 + WHERE id = ?2", + rusqlite::params![chunk_store::CHUNK_STATUS_ADMITTED, chunk.id], )?; + // #1574 write-side cutover: persist the embedding to the + // per-model `mem_tree_chunk_embeddings` sidecar at the active + // signature, inside THIS tx so it commits atomically with the + // lifecycle / score / job-enqueue writes. The legacy + // `mem_tree_chunks.embedding` column is no longer written + // (left intact for the §7 one-shot migration to read). + if let Some(emb) = chunk_embedding.as_deref() { + chunk_store::set_chunk_embedding_for_signature_tx( + &tx, + &chunk.id, + &active_sig, + emb, + )?; + } } else { tx.execute( "UPDATE mem_tree_chunks @@ -525,6 +542,156 @@ async fn handle_flush_stale(config: &Config, job: &Job) -> Result { Ok(JobOutcome::Done) } +/// Texts per `ReembedBackfill` run. Bounded so one run holds the global +/// single-LLM-slot (the job is `is_llm_bound`) for a predictable spell — +/// the laptop-RAM safety the local-LLM-load rule requires. The chain +/// self-continues via `Defer` until no rows remain. +const REEMBED_BACKFILL_BATCH: usize = 16; +/// Delay before the deferred chain revisits this same job row. +const REEMBED_BACKFILL_REVISIT_MS: i64 = 750; + +/// #1574 §6: re-embed a bounded batch of chunks/summaries that lack a +/// vector at the **active** signature, then `Defer` to revisit until the +/// space is fully covered. Sources: the §7 dim-mismatch slice and any +/// embedder switch (post-switch every prior row is missing at the new +/// signature). One chain per signature (dedupe key); self-continues via +/// `Defer` (reschedules this row — no re-enqueue, no dedupe race). +/// +/// Per-row read/embed failures are logged and skipped, never fail the +/// chain — one unreadable row must not strand the rest of memory. +async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result { + let payload: ReembedBackfillPayload = + serde_json::from_str(&job.payload_json).context("parse ReembedBackfill payload")?; + let active_sig = chunk_store::tree_active_signature(config); + if active_sig != payload.signature { + // The embedder changed since this chain started; a fresh chain for + // the new signature supersedes it. Finish this stale one. + log::info!( + "[memory_tree::jobs] reembed_backfill: stale signature (job sig={}, active={active_sig}); finishing", + payload.signature + ); + return Ok(JobOutcome::Done); + } + + // Phase 1 (short read): up to BATCH ids lacking a sidecar vector at the + // active signature — chunks first, then summaries to fill the batch. + let (chunk_ids, summary_ids): (Vec, Vec) = + chunk_store::with_connection(config, |conn| { + let chunks: Vec = { + let mut stmt = conn.prepare( + "SELECT id FROM mem_tree_chunks c + WHERE NOT EXISTS ( + SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = c.id AND e.model_signature = ?1) + LIMIT ?2", + )?; + let ids = stmt + .query_map( + rusqlite::params![active_sig, REEMBED_BACKFILL_BATCH as i64], + |r| r.get::<_, String>(0), + )? + .collect::>>()?; + ids + }; + let remaining = REEMBED_BACKFILL_BATCH.saturating_sub(chunks.len()); + let summaries: Vec = if remaining == 0 { + Vec::new() + } else { + let mut stmt = conn.prepare( + "SELECT id FROM mem_tree_summaries s + WHERE s.deleted = 0 AND NOT EXISTS ( + SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1) + LIMIT ?2", + )?; + let ids = stmt + .query_map(rusqlite::params![active_sig, remaining as i64], |r| { + r.get::<_, String>(0) + })? + .collect::>>()?; + ids + }; + Ok((chunks, summaries)) + })?; + + if chunk_ids.is_empty() && summary_ids.is_empty() { + crate::openhuman::memory::tree::jobs::set_backfill_in_progress(false); + log::info!( + "[memory_tree::jobs] reembed_backfill: sig={active_sig} fully covered; chain complete" + ); + return Ok(JobOutcome::Done); + } + crate::openhuman::memory::tree::jobs::set_backfill_in_progress(true); + + // Phase 2 (no tx held): embed each row's stored source text. Per-row + // errors are skipped (logged) so a single bad row can't strand memory. + let embedder = + build_embedder_from_config(config).context("build embedder in reembed_backfill")?; + let mut chunk_vecs: Vec<(String, Vec)> = Vec::new(); + for id in &chunk_ids { + match content_read::read_chunk_body(config, id) { + Ok(body) => match embedder.embed(&body).await { + Ok(v) if pack_checked(&v).is_ok() => chunk_vecs.push((id.clone(), v)), + Ok(_) => log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} embed wrong dim, skipping" + ), + Err(e) => log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} embed failed: {e}; skipping" + ), + }, + Err(e) => log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} body read failed: {e}; skipping" + ), + } + } + let mut summary_vecs: Vec<(String, Vec)> = Vec::new(); + for id in &summary_ids { + match content_read::read_summary_body(config, id) { + Ok(body) => match embedder.embed(&body).await { + Ok(v) if pack_checked(&v).is_ok() => summary_vecs.push((id.clone(), v)), + Ok(_) => log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} embed wrong dim, skipping" + ), + Err(e) => log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} embed failed: {e}; skipping" + ), + }, + Err(e) => log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} body read failed: {e}; skipping" + ), + } + } + + // Phase 3 (one short tx): persist all collected vectors to the sidecar. + chunk_store::with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + for (id, v) in &chunk_vecs { + chunk_store::set_chunk_embedding_for_signature_tx(&tx, id, &active_sig, v)?; + } + for (id, v) in &summary_vecs { + crate::openhuman::memory::tree::tree_source::store::set_summary_embedding_for_signature_tx( + &tx, id, &active_sig, v, + )?; + } + tx.commit()?; + Ok(()) + })?; + + log::info!( + "[memory_tree::jobs] reembed_backfill: sig={active_sig} embedded chunks={} summaries={} (scanned c={} s={}); revisiting", + chunk_vecs.len(), + summary_vecs.len(), + chunk_ids.len(), + summary_ids.len() + ); + // More rows may remain (this batch was bounded). Reschedule THIS row — + // no re-enqueue, so the per-signature dedupe key stays valid. + Ok(JobOutcome::Defer { + until_ms: chrono::Utc::now().timestamp_millis() + REEMBED_BACKFILL_REVISIT_MS, + reason: "#1574 §6 re-embed backfill: batch done, more pending".to_string(), + }) +} + #[cfg(test)] mod tests { use super::*; @@ -880,4 +1047,171 @@ mod tests { // No new jobs should have been enqueued (buffer didn't cross gate). assert_eq!(count_total(&cfg).unwrap(), pre); } + + /// #1574 §6: a chunk with content but no sidecar vector at the active + /// signature (the post-switch / dim-mismatch state) is re-embedded by + /// `handle_reembed_backfill`; the chain `Defer`s while work remains and + /// returns `Done` once the space is covered; a stale-signature job + /// finishes immediately without touching anything. + /// + /// (The process-global `backfill_in_progress` flag is intentionally not + /// asserted here — it is shared across parallel tests and set widely by + /// the §7 trigger, so asserting it would be flaky. The handler's + /// deterministic effects are what this test pins.) + #[tokio::test] + async fn reembed_backfill_repopulates_then_completes() { + use crate::openhuman::memory::tree::store::{ + get_chunk_embedding_for_signature, tree_active_signature, upsert_chunks, + upsert_staged_chunks_tx, + }; + use crate::openhuman::memory::tree::types::{ + chunk_id, Chunk, Metadata, SourceKind, SourceRef, + }; + + let (_tmp, cfg) = test_config(); + let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let chunk = Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "reembed-seed"), + content: "memory content about the phoenix migration project".into(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 12, + seq_in_source: 0, + created_at: ts, + partial_message: false, + }; + upsert_chunks(&cfg, &[chunk.clone()]).unwrap(); + // Stage the body to disk so `read_chunk_body` succeeds in the handler. + let content_root = cfg.memory_tree_content_root(); + std::fs::create_dir_all(&content_root).unwrap(); + let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap(); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_staged_chunks_tx(&tx, &staged)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + + let sig = tree_active_signature(&cfg); + assert!( + get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig) + .unwrap() + .is_none(), + "precondition: no sidecar vector at the active signature" + ); + + // Work present → re-embed + write sidecar, Defer to revisit. + let job = mk_running_job( + JobKind::ReembedBackfill, + serde_json::to_string(&ReembedBackfillPayload { + signature: sig.clone(), + }) + .unwrap(), + ); + let out = handle_reembed_backfill(&cfg, &job).await.unwrap(); + assert!( + matches!(out, JobOutcome::Defer { .. }), + "work present must Defer (self-continue), got {out:?}" + ); + assert!( + get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig) + .unwrap() + .is_some(), + "chunk re-embedded into the sidecar at the active signature" + ); + + // Nothing left → Done. + let out2 = handle_reembed_backfill(&cfg, &job).await.unwrap(); + assert_eq!(out2, JobOutcome::Done, "covered space must complete"); + + // Stale signature (embedder changed since enqueue) → finishes + // immediately, no work, no panic. + let stale = mk_running_job( + JobKind::ReembedBackfill, + serde_json::to_string(&ReembedBackfillPayload { + signature: "provider=other;model=x;dims=1".into(), + }) + .unwrap(), + ); + assert_eq!( + handle_reembed_backfill(&cfg, &stale).await.unwrap(), + JobOutcome::Done + ); + } + + /// #1574 §4: `ensure_reembed_backfill` (the switch-path trigger) enqueues + /// exactly one chain when there is uncovered work, is idempotent on + /// re-call (per-signature dedupe), and enqueues nothing for an + /// empty/covered space. + #[tokio::test] + async fn ensure_reembed_backfill_enqueues_only_when_uncovered() { + use crate::openhuman::memory::tree::jobs::ensure_reembed_backfill; + use crate::openhuman::memory::tree::store::{upsert_chunks, upsert_staged_chunks_tx}; + use crate::openhuman::memory::tree::types::{ + chunk_id, Chunk, Metadata, SourceKind, SourceRef, + }; + + // Empty space → nothing to do → no job. + let (_t0, empty_cfg) = test_config(); + ensure_reembed_backfill(&empty_cfg); + assert_eq!( + count_jobs_of_kind(&empty_cfg, "reembed_backfill"), + 0, + "empty/covered space must not enqueue a backfill" + ); + + // Chunk with content but no sidecar vector → exactly one chain. + let (_t1, cfg) = test_config(); + let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let chunk = Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "ensure-seed"), + content: "memory content needing a re-embed".into(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 12, + seq_in_source: 0, + created_at: ts, + partial_message: false, + }; + upsert_chunks(&cfg, &[chunk.clone()]).unwrap(); + let content_root = cfg.memory_tree_content_root(); + std::fs::create_dir_all(&content_root).unwrap(); + let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap(); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_staged_chunks_tx(&tx, &staged)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + + ensure_reembed_backfill(&cfg); + assert_eq!( + count_jobs_of_kind(&cfg, "reembed_backfill"), + 1, + "uncovered work must enqueue exactly one backfill chain" + ); + // Idempotent — re-call must not create a second chain (dedupe by sig). + ensure_reembed_backfill(&cfg); + assert_eq!( + count_jobs_of_kind(&cfg, "reembed_backfill"), + 1, + "re-call must dedupe to a single chain per signature" + ); + } } diff --git a/src/openhuman/memory/tree/jobs/mod.rs b/src/openhuman/memory/tree/jobs/mod.rs index 77e8287f2..7550e98ce 100644 --- a/src/openhuman/memory/tree/jobs/mod.rs +++ b/src/openhuman/memory/tree/jobs/mod.rs @@ -33,6 +33,96 @@ pub mod testing; pub mod types; mod worker; +use std::sync::atomic::{AtomicBool, Ordering}; + +/// #1574 §6 / #1365: set while a re-embed backfill chain has work pending. +/// +/// Read by the first-person / subconscious retrieval layer so an empty +/// vector-search result during the backfill window is interpreted as +/// "not searched yet" rather than "no such memory" — preventing the agent +/// from confidently asserting false self-ignorance mid-re-embed. Set true +/// when a backfill is enqueued / still has rows; cleared when the chain +/// drains. Process-global (resets to false on restart; the worker re-sets +/// it on the next backfill tick — acceptable for v1). +static BACKFILL_IN_PROGRESS: AtomicBool = AtomicBool::new(false); + +/// Mark whether a re-embed backfill currently has pending work. +pub fn set_backfill_in_progress(v: bool) { + BACKFILL_IN_PROGRESS.store(v, Ordering::Relaxed); +} + +/// True while a re-embed backfill chain still has rows to process. The +/// #1365 absence-reasoning consumer checks this before treating an empty +/// semantic-recall result as "no memory exists". +pub fn backfill_in_progress() -> bool { + BACKFILL_IN_PROGRESS.load(Ordering::Relaxed) +} + +/// #1574 §4: ensure a re-embed backfill chain exists for the **current** +/// active signature, if (and only if) there is uncovered work. +/// +/// This is the switch-path trigger: call it after the embedder config +/// changes (a new signature → every prior row is missing at it). The §7 +/// migration is one-shot (`user_version`-gated) so it does NOT fire on a +/// later model switch — without this, switching silently blinds prior +/// memory. Standalone (own connection); the §7 migration keeps its own +/// in-tx enqueue (atomic with the copy). Idempotent + non-fatal: the +/// per-signature dedupe key means at most one chain per space, and a +/// covered space enqueues nothing. Errors are logged, never propagated — +/// a failed enqueue must not fail the user's settings save. +pub fn ensure_reembed_backfill(config: &crate::openhuman::config::Config) { + let sig = crate::openhuman::memory::tree::store::tree_active_signature(config); + let result = crate::openhuman::memory::tree::store::with_connection(config, |conn| { + let has_uncovered: bool = conn.query_row( + "SELECT EXISTS( + SELECT 1 FROM mem_tree_chunks c + WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = c.id AND e.model_signature = ?1)) + OR EXISTS( + SELECT 1 FROM mem_tree_summaries s + WHERE s.deleted = 0 AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1))", + rusqlite::params![sig], + |r| r.get(0), + )?; + Ok(has_uncovered) + }); + match result { + Ok(true) => { + let job = match types::NewJob::reembed_backfill(&types::ReembedBackfillPayload { + signature: sig.clone(), + }) { + Ok(j) => j, + Err(e) => { + log::warn!( + "[memory_tree::jobs] ensure_reembed_backfill: build job failed: {e}" + ); + return; + } + }; + match store::enqueue(config, &job) { + Ok(_) => { + set_backfill_in_progress(true); + log::info!( + "[memory_tree::jobs] ensure_reembed_backfill: enqueued chain for sig={sig}" + ); + } + Err(e) => log::warn!( + "[memory_tree::jobs] ensure_reembed_backfill: enqueue failed for sig={sig}: {e}" + ), + } + } + Ok(false) => { + log::debug!( + "[memory_tree::jobs] ensure_reembed_backfill: sig={sig} fully covered; nothing to do" + ); + } + Err(e) => log::warn!( + "[memory_tree::jobs] ensure_reembed_backfill: coverage probe failed for sig={sig}: {e}" + ), + } +} + pub use scheduler::{backfill_missing_digests, trigger_digest}; pub use store::{ claim_next, count_by_status, count_total, enqueue, enqueue_tx, get_job, mark_deferred, diff --git a/src/openhuman/memory/tree/jobs/types.rs b/src/openhuman/memory/tree/jobs/types.rs index 6aaab3582..8b1598d02 100644 --- a/src/openhuman/memory/tree/jobs/types.rs +++ b/src/openhuman/memory/tree/jobs/types.rs @@ -24,6 +24,10 @@ pub enum JobKind { DigestDaily, /// Walk stale buffers and enqueue `Seal` jobs for any over the age cap. FlushStale, + /// #1574 §6: re-embed a bounded batch of chunks/summaries that lack a + /// vector at the active embedding signature (post model-switch, or the + /// §7 dim-mismatch slice), then self-continue until none remain. + ReembedBackfill, } impl JobKind { @@ -36,6 +40,7 @@ impl JobKind { JobKind::TopicRoute => "topic_route", JobKind::DigestDaily => "digest_daily", JobKind::FlushStale => "flush_stale", + JobKind::ReembedBackfill => "reembed_backfill", } } @@ -48,6 +53,7 @@ impl JobKind { "topic_route" => JobKind::TopicRoute, "digest_daily" => JobKind::DigestDaily, "flush_stale" => JobKind::FlushStale, + "reembed_backfill" => JobKind::ReembedBackfill, other => return Err(anyhow!("unknown JobKind '{other}'")), }) } @@ -60,7 +66,11 @@ impl JobKind { pub fn is_llm_bound(&self) -> bool { matches!( self, - JobKind::ExtractChunk | JobKind::Seal | JobKind::DigestDaily | JobKind::TopicRoute + JobKind::ExtractChunk + | JobKind::Seal + | JobKind::DigestDaily + | JobKind::TopicRoute + | JobKind::ReembedBackfill ) } } @@ -267,6 +277,26 @@ impl FlushStalePayload { } } +/// #1574 §6 re-embed backfill. One chain per embedding signature: the +/// `dedupe_key` is the signature, so re-triggering while a chain is +/// in-flight is correctly suppressed (exactly one backfill per space). +/// The handler self-continues via `JobOutcome::Defer` (reschedules this +/// same row) rather than re-enqueuing, so the fixed dedupe key is safe. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ReembedBackfillPayload { + /// The embedding signature this chain re-embeds under. If the active + /// signature has since changed, the handler treats this job as stale + /// and finishes (a fresh chain for the new signature takes over). + pub signature: String, +} + +impl ReembedBackfillPayload { + /// Stable dedupe key — one in-flight backfill chain per signature. + pub fn dedupe_key(&self) -> String { + format!("reembed_backfill:{}", self.signature) + } +} + /// One row in `mem_tree_jobs`. `payload_json` is left as a raw string so /// callers parse it lazily based on `kind`. #[derive(Clone, Debug)] @@ -365,6 +395,17 @@ impl NewJob { max_attempts: None, }) } + + /// Build a [`JobKind::ReembedBackfill`] enqueue request (#1574 §6). + pub fn reembed_backfill(p: &ReembedBackfillPayload) -> Result { + Ok(Self { + kind: JobKind::ReembedBackfill, + payload_json: serde_json::to_string(p)?, + dedupe_key: Some(p.dedupe_key()), + available_at_ms: None, + max_attempts: None, + }) + } } #[cfg(test)] @@ -380,6 +421,7 @@ mod tests { JobKind::TopicRoute, JobKind::DigestDaily, JobKind::FlushStale, + JobKind::ReembedBackfill, ] { assert_eq!(JobKind::parse(k.as_str()).unwrap(), k); } @@ -454,6 +496,7 @@ mod tests { assert!(JobKind::Seal.is_llm_bound()); assert!(JobKind::DigestDaily.is_llm_bound()); assert!(JobKind::TopicRoute.is_llm_bound()); + assert!(JobKind::ReembedBackfill.is_llm_bound()); assert!(!JobKind::AppendBuffer.is_llm_bound()); assert!(!JobKind::FlushStale.is_llm_bound()); } diff --git a/src/openhuman/memory/tree/retrieval/drill_down.rs b/src/openhuman/memory/tree/retrieval/drill_down.rs index f5190f324..c131f4c87 100644 --- a/src/openhuman/memory/tree/retrieval/drill_down.rs +++ b/src/openhuman/memory/tree/retrieval/drill_down.rs @@ -544,9 +544,9 @@ mod tests { with_connection(cfg, |conn| { let tx = conn.unchecked_transaction()?; tree_store::insert_tree_conn(&tx, &tree)?; - tree_store::insert_summary_tx(&tx, &l1_a, None)?; - tree_store::insert_summary_tx(&tx, &l1_b, None)?; - tree_store::insert_summary_tx(&tx, &root, None)?; + tree_store::insert_summary_tx(&tx, &l1_a, None, "test")?; + tree_store::insert_summary_tx(&tx, &l1_b, None, "test")?; + tree_store::insert_summary_tx(&tx, &root, None, "test")?; tx.commit()?; Ok(()) }) diff --git a/src/openhuman/memory/tree/retrieval/integration_test.rs b/src/openhuman/memory/tree/retrieval/integration_test.rs index 3fd58822e..11291c1bd 100644 --- a/src/openhuman/memory/tree/retrieval/integration_test.rs +++ b/src/openhuman/memory/tree/retrieval/integration_test.rs @@ -271,11 +271,19 @@ async fn seal_populates_summary_embedding() { .unwrap(); assert_eq!(sealed.len(), 1, "expected one seal at the budget crossing"); + // #1574 cutover: the seal path no longer writes the legacy + // `mem_tree_summaries.embedding` column — the vector is persisted to the + // per-model sidecar at the active signature inside the seal tx. Assert + // it round-trips through the public accessor (which reads the sidecar at + // the active signature), validating the write-side cutover end to end. let summary = src_store::get_summary(&cfg, &sealed[0]).unwrap().unwrap(); - let emb = summary - .embedding - .as_ref() - .expect("sealed summary must have embedding"); + assert!( + summary.embedding.is_none(), + "legacy summary embedding column must be NULL post-cutover" + ); + let emb = src_store::get_summary_embedding(&cfg, &sealed[0]) + .unwrap() + .expect("sealed summary must have an embedding in the per-model sidecar"); assert_eq!(emb.len(), EMBEDDING_DIM); } diff --git a/src/openhuman/memory/tree/retrieval/topic.rs b/src/openhuman/memory/tree/retrieval/topic.rs index c87353093..7c049c0bb 100644 --- a/src/openhuman/memory/tree/retrieval/topic.rs +++ b/src/openhuman/memory/tree/retrieval/topic.rs @@ -590,7 +590,7 @@ mod tests { with_connection(&cfg, |conn| { let tx = conn.unchecked_transaction()?; tree_store::insert_tree_conn(&tx, &tree)?; - tree_store::insert_summary_tx(&tx, &summary, None)?; + tree_store::insert_summary_tx(&tx, &summary, None, "test")?; score_store::index_entities_tx( &tx, &[entity], diff --git a/src/openhuman/memory/tree/rpc.rs b/src/openhuman/memory/tree/rpc.rs index 5b2beff25..9d59a831e 100644 --- a/src/openhuman/memory/tree/rpc.rs +++ b/src/openhuman/memory/tree/rpc.rs @@ -256,6 +256,61 @@ pub async fn trigger_digest_rpc( )) } +/// Response from the `memory_backfill_status` RPC (#1574 §4b). The frontend +/// polls this while the re-embed modal is open to surface progress and to +/// dismiss the modal once the new embedding space is fully covered. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct BackfillStatusResponse { + /// True while a re-embed backfill chain still has work pending — the + /// #1365 flag OR a queued/running `reembed_backfill` job. + pub in_progress: bool, + /// Count of `reembed_backfill` jobs in `ready` or `running` state. `0` + /// with `in_progress=false` means the active embedding space is fully + /// covered (modal can close). + pub pending_jobs: u64, +} + +/// `memory_backfill_status` RPC handler (#1574 §4b). No inputs — reports +/// whether a per-model re-embed backfill is in flight so the UI can warn +/// the user that semantic recall is reduced until it drains. +pub async fn backfill_status_rpc( + config: &Config, +) -> Result, String> { + log::debug!("[memory_tree::rpc] backfill_status: entry"); + // SQLite I/O off the async runtime thread, matching the sibling + // DB-backed handlers in this module (`get_chunk_rpc`, etc.). + let pending_jobs: u64 = tokio::task::spawn_blocking({ + let config = config.clone(); + move || { + store::with_connection(&config, |conn| { + let n: i64 = conn.query_row( + "SELECT COUNT(*) FROM mem_tree_jobs + WHERE kind = 'reembed_backfill' AND status IN ('ready', 'running')", + [], + |r| r.get(0), + )?; + Ok(n.max(0) as u64) + }) + } + }) + .await + .map_err(|e| format!("memory_backfill_status join error: {e}"))? + .map_err(|e| { + let msg = format!("memory_backfill_status: {e}"); + log::debug!("[memory_tree::rpc] backfill_status: error: {msg}"); + msg + })?; + let in_progress = + crate::openhuman::memory::tree::jobs::backfill_in_progress() || pending_jobs > 0; + Ok(RpcOutcome::single_log( + BackfillStatusResponse { + in_progress, + pending_jobs, + }, + format!("memory_tree: backfill_status in_progress={in_progress} pending={pending_jobs}"), + )) +} + #[cfg(test)] mod tests { use super::*; @@ -327,4 +382,30 @@ mod tests { assert!(second.job_id.is_none()); assert_eq!(count_total(&cfg).unwrap(), 1); } + + /// #1574 §4b: `backfill_status_rpc` reports 0 pending on an idle space + /// and reflects a queued `reembed_backfill` job (forcing `in_progress`). + /// `in_progress` for the empty case is intentionally not asserted — the + /// underlying flag is a process-global shared across parallel tests. + #[tokio::test] + async fn backfill_status_reports_pending_jobs() { + use crate::openhuman::memory::tree::jobs; + let (_tmp, cfg) = test_config(); + + let s0 = backfill_status_rpc(&cfg).await.unwrap().value; + assert_eq!(s0.pending_jobs, 0, "idle space has no pending backfill"); + + let job = jobs::types::NewJob::reembed_backfill(&jobs::types::ReembedBackfillPayload { + signature: "provider=test;model=x;dims=1".into(), + }) + .unwrap(); + jobs::enqueue(&cfg, &job).unwrap(); + + let s1 = backfill_status_rpc(&cfg).await.unwrap().value; + assert_eq!( + s1.pending_jobs, 1, + "a ready reembed_backfill job must count" + ); + assert!(s1.in_progress, "pending>0 forces in_progress=true"); + } } diff --git a/src/openhuman/memory/tree/schemas.rs b/src/openhuman/memory/tree/schemas.rs index b4ff42519..a3caa22a3 100644 --- a/src/openhuman/memory/tree/schemas.rs +++ b/src/openhuman/memory/tree/schemas.rs @@ -30,6 +30,7 @@ pub fn all_controller_schemas() -> Vec { schemas("list_chunks"), schemas("get_chunk"), schemas("trigger_digest"), + schemas("memory_backfill_status"), schemas("list_sources"), schemas("search"), schemas("recall"), @@ -67,6 +68,10 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("trigger_digest"), handler: handle_trigger_digest, }, + RegisteredController { + schema: schemas("memory_backfill_status"), + handler: handle_memory_backfill_status, + }, RegisteredController { schema: schemas("list_sources"), handler: handle_list_sources, @@ -712,6 +717,32 @@ pub fn schemas(function: &str) -> ControllerSchema { }, ], }, + "memory_backfill_status" => ControllerSchema { + namespace: NAMESPACE, + function: "memory_backfill_status", + description: "Report whether a per-model embedding re-embed \ + backfill (#1574) is in flight. The UI polls this while the \ + re-embed modal is open: semantic recall over not-yet-\ + re-embedded memory is reduced until the chain drains.", + inputs: vec![], + outputs: vec![ + FieldSchema { + name: "in_progress", + ty: TypeSchema::Bool, + comment: "True while a re-embed backfill still has work \ + pending (flag set or a ready/running job).", + required: true, + }, + FieldSchema { + name: "pending_jobs", + ty: TypeSchema::U64, + comment: "Count of reembed_backfill jobs in ready or \ + running state; 0 with in_progress=false means the \ + active embedding space is fully covered.", + required: true, + }, + ], + }, _ => ControllerSchema { namespace: NAMESPACE, function: "unknown", @@ -756,6 +787,13 @@ fn handle_trigger_digest(params: Map) -> ControllerFuture { }) } +fn handle_memory_backfill_status(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(tree_rpc::backfill_status_rpc(&config).await?) + }) +} + // ── New read RPCs (Memory-tab UI) ──────────────────────────────────────── fn handle_list_chunks(params: Map) -> ControllerFuture { diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index f0c8a5067..4b03e3a1a 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -43,6 +43,11 @@ pub const CHUNK_STATUS_SEALED: &str = "sealed"; /// Chunk lifecycle: rejected by the admission gate (too low signal). pub const CHUNK_STATUS_DROPPED: &str = "dropped"; +/// `PRAGMA user_version` value once the one-shot legacy→sidecar embedding +/// migration (#1574 §7) has run. `0` (fresh/legacy DB) triggers the copy on +/// next open; `>= 1` skips it. Bump only for a new one-shot data migration. +const TREE_EMBEDDING_MIGRATION_VERSION: i64 = 1; + const SCHEMA: &str = " PRAGMA foreign_keys = ON; @@ -776,9 +781,125 @@ pub(crate) fn with_connection( "is_user", "INTEGER NOT NULL DEFAULT 0", )?; + // #1574 §7: one-shot, version-gated copy of legacy embedding columns + // into the per-model sidecar at the active signature. Runs once + // (PRAGMA user_version gate); cheap no-op on every subsequent open. + migrate_legacy_embeddings_to_sidecar(&conn, config)?; f(&conn) } +/// One-shot migration (#1574 §7, vN): copy legacy `mem_tree_chunks.embedding` +/// / `mem_tree_summaries.embedding` blobs into the per-model sidecar tables +/// under the **active** signature, when (and only when) the legacy vector's +/// dimensionality matches the active embedder's. +/// +/// Version-gated via `PRAGMA user_version`: returns immediately once +/// `>= TREE_EMBEDDING_MIGRATION_VERSION`, so the per-open cost is a single +/// pragma read. Dim-mismatched rows are left for the §6 re-embed backfill — +/// the blob's signature is unrecoverable (see spec §7b), so a same-length +/// copy under the active signature is the only provably-safe move and +/// anything else must be re-embedded. The legacy columns are **kept** (read +/// here, dropped only in a later release — spec §7c). Idempotent: re-running +/// before the version bumps re-copies the same rows harmlessly (sidecar +/// upsert is ON CONFLICT); after the bump it is skipped entirely. +fn migrate_legacy_embeddings_to_sidecar(conn: &Connection, config: &Config) -> Result<()> { + let version: i64 = conn + .query_row("PRAGMA user_version", [], |r| r.get(0)) + .context("read PRAGMA user_version for #1574 migration")?; + if version >= TREE_EMBEDDING_MIGRATION_VERSION { + return Ok(()); + } + + let (provider, model, dims) = crate::openhuman::memory::store::effective_embedding_settings( + &config.memory, + config.workload_local_model("embeddings").as_deref(), + ); + let sig = crate::openhuman::embeddings::format_embedding_signature(&provider, &model, dims); + log::info!( + "[memory_tree::migrate] #1574 §7: copying legacy embeddings → sidecar at sig={sig} (dims={dims})" + ); + + let tx = conn.unchecked_transaction()?; + let mut copied_chunks = 0usize; + let mut copied_summaries = 0usize; + let mut skipped_dim_mismatch = 0usize; + + for (table, is_chunk) in [("mem_tree_chunks", true), ("mem_tree_summaries", false)] { + let mut stmt = tx.prepare(&format!( + "SELECT id, embedding FROM {table} WHERE embedding IS NOT NULL" + ))?; + let rows = stmt.query_map([], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, Vec>(1)?)) + })?; + for row in rows { + let (id, blob) = row?; + if !blob.len().is_multiple_of(4) { + log::warn!( + "[memory_tree::migrate] {table} id={id}: legacy blob len {} not /4, skipping", + blob.len() + ); + continue; + } + if blob.len() / 4 != dims { + // Different embedding space — unrecoverable from the blob. + // Leave for the §6 re-embed backfill. + skipped_dim_mismatch += 1; + continue; + } + let vec: Vec = blob + .chunks_exact(4) + .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .collect(); + if is_chunk { + set_chunk_embedding_for_signature_tx(&tx, &id, &sig, &vec)?; + copied_chunks += 1; + } else { + crate::openhuman::memory::tree::tree_source::store::set_summary_embedding_for_signature_tx( + &tx, &id, &sig, &vec, + )?; + copied_summaries += 1; + } + } + } + + // #1574 §6: enqueue the re-embed backfill ONLY if there is genuinely + // uncovered work at the active signature (the dim-mismatch slice, or + // content-bearing rows with no vector). Gating this avoids queuing a + // no-op job on every DB open — which would otherwise pollute the jobs + // table for unrelated callers/tests. Enqueued atomically with the + // migration; dedupe key = signature, so exactly one chain per space. + let has_uncovered: bool = tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM mem_tree_chunks c + WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = c.id AND e.model_signature = ?1)) + OR EXISTS( + SELECT 1 FROM mem_tree_summaries s + WHERE s.deleted = 0 AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1))", + rusqlite::params![sig], + |r| r.get(0), + )?; + if has_uncovered { + let backfill_job = crate::openhuman::memory::tree::jobs::types::NewJob::reembed_backfill( + &crate::openhuman::memory::tree::jobs::types::ReembedBackfillPayload { + signature: sig.clone(), + }, + )?; + crate::openhuman::memory::tree::jobs::enqueue_tx(&tx, &backfill_job)?; + crate::openhuman::memory::tree::jobs::set_backfill_in_progress(true); + } + + tx.commit()?; + conn.pragma_update(None, "user_version", TREE_EMBEDDING_MIGRATION_VERSION) + .context("set PRAGMA user_version after #1574 migration")?; + log::info!( + "[memory_tree::migrate] #1574 §7 done: copied chunks={copied_chunks} summaries={copied_summaries} \ + skipped_dim_mismatch={skipped_dim_mismatch} (left for §6 re-embed); user_version={TREE_EMBEDDING_MIGRATION_VERSION}" + ); + Ok(()) +} + /// One pointer into the raw archive. A chunk's body is reconstructed by /// reading each [`RawRef`] in order and joining with `"\n\n"`. /// @@ -945,31 +1066,45 @@ fn add_column_if_missing(conn: &Connection, table: &str, name: &str, sql_type: & // ── Phase 2: embedding column accessors ───────────────────────────────── -/// Store a chunk's embedding as a packed little-endian `f32` blob. -/// -/// Length is `embedding.len() * 4` bytes. The caller is responsible for -/// ensuring all embeddings in a given deployment share the same dimension. -pub fn set_chunk_embedding(config: &Config, chunk_id: &str, embedding: &[f32]) -> Result<()> { - let bytes: Vec = embedding.iter().flat_map(|f| f.to_le_bytes()).collect(); - with_connection(config, |conn| { - let changed = conn.execute( - "UPDATE mem_tree_chunks SET embedding = ?1 WHERE id = ?2", - rusqlite::params![bytes, chunk_id], - )?; - if changed == 0 { - log::warn!("[memory_tree::store] set_chunk_embedding: no row for chunk_id={chunk_id}"); - } - Ok(()) - }) +/// Resolve the active embedding signature for the memory tree from the global +/// [`Config`] — the canonical key every per-model sidecar read/write is scoped +/// by (#1574). Reuses the established local-AI workload derivation +/// ([`Config::workload_local_model`]) and the probe-stable +/// `active_embedding_signature`; introduces no parallel resolution path. +/// `pub(crate)` so the sibling `tree_source` summary store shares the exact +/// same resolution. +pub(crate) fn tree_active_signature(config: &Config) -> String { + let local_model = config.workload_local_model("embeddings"); + crate::openhuman::memory::store::active_embedding_signature( + &config.memory, + local_model.as_deref(), + ) } -/// Store a chunk embedding for a specific provider/model/dimension signature. +/// Store a chunk's embedding under the active model signature. /// -/// This is the Stage-1 per-model table write path for #1574. The legacy -/// `mem_tree_chunks.embedding` column is intentionally left untouched by this -/// helper so callers can dual-write while query paths migrate. -pub fn set_chunk_embedding_for_signature( - config: &Config, +/// #1574 cutover: this now writes the per-model `mem_tree_chunk_embeddings` +/// sidecar (via [`set_chunk_embedding_for_signature`]) instead of the legacy +/// `mem_tree_chunks.embedding` column. Call sites are unchanged — the signature +/// is resolved internally from `config`. The legacy column is left intact for +/// the §7 one-shot migration to read; it is dropped only in a later release. +pub fn set_chunk_embedding(config: &Config, chunk_id: &str, embedding: &[f32]) -> Result<()> { + let signature = tree_active_signature(config); + log::debug!( + "[memory_tree::store] set_chunk_embedding: chunk_id={chunk_id} sig={signature} dims={}", + embedding.len() + ); + set_chunk_embedding_for_signature(config, chunk_id, &signature, embedding) +} + +/// Core upsert into `mem_tree_chunk_embeddings` over an arbitrary +/// `&Connection`. Shared by the standalone ([`set_chunk_embedding_for_signature`]) +/// and in-transaction ([`set_chunk_embedding_for_signature_tx`]) write paths so +/// the SQL exists exactly once. `rusqlite::Transaction` derefs to `Connection`, +/// so an in-tx caller passes `&tx` and the sidecar row commits atomically with +/// the surrounding work (#1574 write-side cutover). +fn upsert_chunk_embedding_conn( + conn: &rusqlite::Connection, chunk_id: &str, model_signature: &str, embedding: &[f32], @@ -977,21 +1112,50 @@ pub fn set_chunk_embedding_for_signature( let bytes = embedding_to_blob(embedding); let dim = i64::try_from(embedding.len()).context("embedding dimension does not fit i64")?; let created_at = Utc::now().timestamp_millis() as f64 / 1000.0; - with_connection(config, |conn| { - conn.execute( - "INSERT INTO mem_tree_chunk_embeddings + conn.execute( + "INSERT INTO mem_tree_chunk_embeddings (chunk_id, model_signature, vector, dim, created_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(chunk_id, model_signature) DO UPDATE SET vector = excluded.vector, dim = excluded.dim, created_at = excluded.created_at", - rusqlite::params![chunk_id, model_signature, bytes, dim, created_at], - )?; - Ok(()) + rusqlite::params![chunk_id, model_signature, bytes, dim, created_at], + )?; + Ok(()) +} + +/// Store a chunk embedding for a specific provider/model/dimension signature. +/// +/// Per-model table write path for #1574. The legacy +/// `mem_tree_chunks.embedding` column is intentionally left untouched by this +/// helper (read by the §7 migration; dropped only in a later release). +pub fn set_chunk_embedding_for_signature( + config: &Config, + chunk_id: &str, + model_signature: &str, + embedding: &[f32], +) -> Result<()> { + with_connection(config, |conn| { + upsert_chunk_embedding_conn(conn, chunk_id, model_signature, embedding) }) } +/// Transaction-scoped variant of [`set_chunk_embedding_for_signature`]. +/// +/// For callers that already hold a `Transaction` (e.g. the chunk-admission +/// handler, which commits the sidecar row in the SAME tx as the lifecycle +/// + score + job-enqueue writes — #1574 write-side cutover). Opening a fresh +/// connection there would break atomicity / deadlock on the busy DB. +pub(crate) fn set_chunk_embedding_for_signature_tx( + tx: &rusqlite::Transaction<'_>, + chunk_id: &str, + model_signature: &str, + embedding: &[f32], +) -> Result<()> { + upsert_chunk_embedding_conn(tx, chunk_id, model_signature, embedding) +} + /// Fetch a chunk embedding for exactly one provider/model/dimension signature. pub fn get_chunk_embedding_for_signature( config: &Config, @@ -1015,32 +1179,17 @@ pub fn get_chunk_embedding_for_signature( }) } -/// Fetch a chunk's embedding, decoding the stored little-endian `f32` blob. +/// Fetch a chunk's embedding for the active model signature. /// -/// Returns `Ok(None)` if the chunk doesn't exist or has no embedding stored. +/// #1574 cutover: reads the per-model `mem_tree_chunk_embeddings` sidecar at +/// the active signature (via [`get_chunk_embedding_for_signature`]) instead of +/// the legacy `mem_tree_chunks.embedding` column. Returns `Ok(None)` if the +/// chunk has no vector under the active signature — e.g. during the §7 +/// backfill window, where this degrades retrieval gracefully (the row is +/// simply absent from vector results, never cross-space compared). pub fn get_chunk_embedding(config: &Config, chunk_id: &str) -> Result>> { - with_connection(config, |conn| { - let blob: Option>> = conn - .query_row( - "SELECT embedding FROM mem_tree_chunks WHERE id = ?1", - rusqlite::params![chunk_id], - |r| r.get::<_, Option>>(0), - ) - .optional()?; - match blob.flatten() { - None => Ok(None), - Some(bytes) => { - if !bytes.len().is_multiple_of(4) { - anyhow::bail!("embedding blob length {} not a multiple of 4", bytes.len()); - } - let floats: Vec = bytes - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) - .collect(); - Ok(Some(floats)) - } - } - }) + let signature = tree_active_signature(config); + get_chunk_embedding_for_signature(config, chunk_id, &signature) } fn embedding_to_blob(embedding: &[f32]) -> Vec { diff --git a/src/openhuman/memory/tree/store_tests.rs b/src/openhuman/memory/tree/store_tests.rs index cb70b7dda..ff77d4364 100644 --- a/src/openhuman/memory/tree/store_tests.rs +++ b/src/openhuman/memory/tree/store_tests.rs @@ -101,7 +101,26 @@ fn chunk_embeddings_are_scoped_by_model_signature() { .unwrap() .is_none() ); + + // #1574 cutover: the public `get_chunk_embedding` now reads the sidecar at + // the *active* signature (not the legacy column). Nothing was written + // there yet, so it is absent — graceful, never a cross-space read of the + // openai/local rows above. assert!(get_chunk_embedding(&cfg, &c.id).unwrap().is_none()); + + // The public setter targets the active signature and round-trips through + // the public getter — proves the cutover wiring end to end. + set_chunk_embedding(&cfg, &c.id, &[0.7, 0.8]).unwrap(); + assert_eq!( + get_chunk_embedding(&cfg, &c.id).unwrap(), + Some(vec![0.7, 0.8]) + ); + + // ...and the earlier per-signature rows remain independently scoped. + assert_eq!( + get_chunk_embedding_for_signature(&cfg, &c.id, "local/bge-small@384").unwrap(), + Some(vec![0.3, 0.4, 0.5]) + ); } #[test] @@ -236,3 +255,88 @@ fn schema_has_content_path_and_content_sha256_columns() { }) .unwrap(); } + +/// #1574 §7: the one-shot, version-gated legacy→sidecar migration copies a +/// legacy `embedding` blob whose dimensionality matches the active embedder +/// into the per-model sidecar at the active signature, skips dim-mismatched +/// rows (left for the §6 re-embed), keeps the legacy column, and runs exactly +/// once (PRAGMA user_version gate). +#[test] +fn legacy_embeddings_migrate_to_sidecar_once() { + let (_tmp, cfg) = test_config(); + let c_match = sample_chunk("slack:#eng", 0, 1_700_000_000_000); + let c_mismatch = sample_chunk("slack:#eng", 1, 1_700_000_000_001); + // First open runs the (no-op) migration and sets user_version = 1. + upsert_chunks(&cfg, &[c_match.clone(), c_mismatch.clone()]).unwrap(); + + // Resolve the active signature/dims exactly as the migration does — + // base-independent, never hard-coded (see the brittle-literal lesson). + let (p, m, dims) = crate::openhuman::memory::store::effective_embedding_settings( + &cfg.memory, + cfg.workload_local_model("embeddings").as_deref(), + ); + let sig = crate::openhuman::embeddings::format_embedding_signature(&p, &m, dims); + let match_vec = vec![0.25f32; dims]; + let mismatch_vec = vec![0.5f32; dims + 1]; + + // Simulate a pre-#1574 DB: legacy columns populated, migration not yet + // run. On entry user_version is 1 (from upsert above) so the migration + // is skipped here; the closure then resets the gate to 0. + with_connection(&cfg, |conn| { + conn.execute( + "UPDATE mem_tree_chunks SET embedding = ?1 WHERE id = ?2", + params![embedding_to_blob(&match_vec), c_match.id], + )?; + conn.execute( + "UPDATE mem_tree_chunks SET embedding = ?1 WHERE id = ?2", + params![embedding_to_blob(&mismatch_vec), c_mismatch.id], + )?; + conn.pragma_update(None, "user_version", 0i64)?; + Ok(()) + }) + .unwrap(); + + // The next store open (this getter's `with_connection`) sees + // user_version = 0 and runs the migration before returning. + assert_eq!( + get_chunk_embedding_for_signature(&cfg, &c_match.id, &sig).unwrap(), + Some(match_vec.clone()), + "matching-dim legacy row must be copied to the sidecar at the active sig" + ); + assert!( + get_chunk_embedding_for_signature(&cfg, &c_mismatch.id, &sig) + .unwrap() + .is_none(), + "dim-mismatched legacy row must be skipped (left for §6 re-embed)" + ); + + with_connection(&cfg, |conn| { + let legacy: Option> = conn + .query_row( + "SELECT embedding FROM mem_tree_chunks WHERE id = ?1", + params![c_match.id], + |r| r.get(0), + ) + .unwrap(); + assert!( + legacy.is_some(), + "legacy column must be KEPT post-migration (vN+1 drops it later)" + ); + let v: i64 = conn + .query_row("PRAGMA user_version", [], |r| r.get(0)) + .unwrap(); + assert_eq!( + v, TREE_EMBEDDING_MIGRATION_VERSION, + "version gate must be set" + ); + Ok(()) + }) + .unwrap(); + + // Idempotent: subsequent opens are no-ops (gate set); sidecar unchanged. + with_connection(&cfg, |_| Ok(())).unwrap(); + assert_eq!( + get_chunk_embedding_for_signature(&cfg, &c_match.id, &sig).unwrap(), + Some(match_vec) + ); +} diff --git a/src/openhuman/memory/tree/tree_global/digest.rs b/src/openhuman/memory/tree/tree_global/digest.rs index 87e3cde77..36c8666d5 100644 --- a/src/openhuman/memory/tree/tree_global/digest.rs +++ b/src/openhuman/memory/tree/tree_global/digest.rs @@ -247,7 +247,12 @@ pub async fn end_of_day_digest( let tree_id_clone = global.id.clone(); with_connection(config, move |conn| { let tx = conn.unchecked_transaction()?; - store::insert_summary_tx(&tx, &daily_clone, Some(&staged_daily))?; + store::insert_summary_tx( + &tx, + &daily_clone, + Some(&staged_daily), + &crate::openhuman::memory::tree::store::tree_active_signature(config), + )?; // Index any entities the summariser emitted (no-op under inert). crate::openhuman::memory::tree::score::store::index_summary_entity_ids_tx( &tx, diff --git a/src/openhuman/memory/tree/tree_global/seal.rs b/src/openhuman/memory/tree/tree_global/seal.rs index 6cbe3738e..4364d10b1 100644 --- a/src/openhuman/memory/tree/tree_global/seal.rs +++ b/src/openhuman/memory/tree/tree_global/seal.rs @@ -307,7 +307,12 @@ async fn seal_one_level( .map(|n| n.max(0) as u32) .context("Failed to read current max_level for global tree")?; - store::insert_summary_tx(&tx, &node, Some(&staged_global))?; + store::insert_summary_tx( + &tx, + &node, + Some(&staged_global), + &crate::openhuman::memory::tree::store::tree_active_signature(config), + )?; // Index any entities the summariser emitted. No-op under // InertSummariser (entities stays empty by design — see // summariser/inert.rs). Becomes active when the Ollama summariser @@ -452,7 +457,12 @@ mod tests { fn insert_daily(cfg: &Config, node: &SummaryNode) { with_connection(cfg, |conn| { let tx = conn.unchecked_transaction()?; - store::insert_summary_tx(&tx, node, None)?; + store::insert_summary_tx( + &tx, + node, + None, + &crate::openhuman::memory::tree::store::tree_active_signature(cfg), + )?; tx.commit()?; Ok(()) }) diff --git a/src/openhuman/memory/tree/tree_source/bucket_seal.rs b/src/openhuman/memory/tree/tree_source/bucket_seal.rs index 9289fd0a5..e0a96526f 100644 --- a/src/openhuman/memory/tree/tree_source/bucket_seal.rs +++ b/src/openhuman/memory/tree/tree_source/bucket_seal.rs @@ -629,7 +629,12 @@ pub(crate) async fn seal_one_level( .map(|n| n.max(0) as u32) .context("Failed to read current max_level for tree")?; - store::insert_summary_tx(&tx, &node, Some(&staged))?; + store::insert_summary_tx( + &tx, + &node, + Some(&staged), + &crate::openhuman::memory::tree::store::tree_active_signature(config), + )?; // Forward-compat: index any entities the summariser emitted into // `mem_tree_entity_index` so Phase 4 retrieval can resolve // "summaries mentioning Alice" via the same inverted index as diff --git a/src/openhuman/memory/tree/tree_source/store.rs b/src/openhuman/memory/tree/tree_source/store.rs index 2ecd8721f..41a572519 100644 --- a/src/openhuman/memory/tree/tree_source/store.rs +++ b/src/openhuman/memory/tree/tree_source/store.rs @@ -190,14 +190,18 @@ pub(crate) fn insert_summary_tx( tx: &Transaction<'_>, node: &SummaryNode, staged: Option<&StagedSummary>, + model_signature: &str, ) -> Result<()> { - let embedding_blob: Option> = match node.embedding.as_deref() { - Some(v) => Some( - pack_checked(v) - .with_context(|| format!("Failed to pack embedding for summary id={}", node.id))?, - ), - None => None, - }; + // #1574 write-side cutover: keep the dimension guard (fail the seal fast + // on a misconfigured embedder) but DO NOT write the legacy + // `mem_tree_summaries.embedding` column — the vector is persisted to the + // per-model sidecar below, in THIS tx so it commits atomically with the + // summary row. The legacy column is left NULL for the §7 migration. + if let Some(v) = node.embedding.as_deref() { + pack_checked(v) + .with_context(|| format!("validate embedding dims for summary id={}", node.id))?; + } + let embedding_blob: Option> = None; // Phase MD-content: when a staged file exists, truncate `content` to a // ≤500-char plain-text preview (char boundary safe via chars().take(500)). @@ -244,59 +248,60 @@ pub(crate) fn insert_summary_tx( ], ) .with_context(|| format!("Failed to insert summary id={}", node.id))?; + + // #1574: persist the embedding to the per-model sidecar at the active + // signature, in the SAME tx as the summary row insert above. + if let Some(v) = node.embedding.as_deref() { + upsert_summary_embedding_conn(tx, &node.id, model_signature, v)?; + } Ok(()) } /// Set (or overwrite) the embedding for an existing summary row. -/// Exposed for a future backfill helper — not called by ingest/seal -/// today. Returns the number of rows updated (0 if the id is unknown). +/// +/// #1574 cutover: writes the per-model `mem_tree_summary_embeddings` sidecar +/// at the active signature (via [`set_summary_embedding_for_signature`]) +/// instead of the legacy `mem_tree_summaries.embedding` column. The signature +/// is resolved internally from `config` via the shared +/// [`crate::openhuman::memory::tree::store::tree_active_signature`] — same +/// resolution as the chunk path. Returns `1` on success (one sidecar row +/// written/updated); the legacy "0 if id unknown" count no longer applies +/// since the sidecar upsert does not join the parent summary row. pub fn set_summary_embedding( config: &Config, summary_id: &str, embedding: &[f32], ) -> Result { - let blob = pack_checked(embedding) - .with_context(|| format!("Failed to pack embedding for summary id={summary_id}"))?; - with_connection(config, |conn| { - let changed = conn.execute( - "UPDATE mem_tree_summaries SET embedding = ?1 WHERE id = ?2", - params![blob, summary_id], - )?; - if changed == 0 { - log::warn!( - "[tree_source::store] set_summary_embedding: no row for summary_id={summary_id}" - ); - } - Ok(changed) - }) + let signature = crate::openhuman::memory::tree::store::tree_active_signature(config); + log::debug!( + "[tree_source::store] set_summary_embedding: summary_id={summary_id} sig={signature} dims={}", + embedding.len() + ); + set_summary_embedding_for_signature(config, summary_id, &signature, embedding)?; + Ok(1) } -/// Fetch a summary's embedding, decoding the stored little-endian `f32` -/// blob. Returns `Ok(None)` if the summary doesn't exist OR if it exists -/// but has a NULL embedding (legacy / pre-Phase-4 rows). -pub fn get_summary_embedding(config: &Config, summary_id: &str) -> Result>> { - with_connection(config, |conn| { - let blob: Option>> = conn - .query_row( - "SELECT embedding FROM mem_tree_summaries WHERE id = ?1", - params![summary_id], - |r| r.get::<_, Option>>(0), - ) - .optional()?; - match blob { - None => Ok(None), - Some(inner) => decode_optional_blob(inner, &format!("summary_id={summary_id}")), - } - }) -} - -/// Store a summary embedding for a specific provider/model/dimension signature. +/// Fetch a summary's embedding for the active model signature. /// -/// This writes the #1574 per-model table only; the legacy -/// `mem_tree_summaries.embedding` column remains available for dual-read -/// fallback while query paths migrate. -pub fn set_summary_embedding_for_signature( - config: &Config, +/// #1574 cutover: reads the per-model `mem_tree_summary_embeddings` sidecar at +/// the active signature (via [`get_summary_embedding_for_signature`]) instead +/// of the legacy `mem_tree_summaries.embedding` column. `Ok(None)` when no +/// vector exists under the active signature — graceful absence during the §7 +/// backfill window, never a cross-space read. +pub fn get_summary_embedding(config: &Config, summary_id: &str) -> Result>> { + let signature = crate::openhuman::memory::tree::store::tree_active_signature(config); + get_summary_embedding_for_signature(config, summary_id, &signature) +} + +/// Core upsert into `mem_tree_summary_embeddings` over an arbitrary +/// `&Connection`. Shared by the standalone +/// ([`set_summary_embedding_for_signature`]) and in-transaction +/// ([`set_summary_embedding_for_signature_tx`]) write paths so the SQL exists +/// exactly once. `rusqlite::Transaction` derefs to `Connection`, so the seal +/// path passes `&tx` and the sidecar row commits atomically with the summary +/// row insert (#1574 write-side cutover). +fn upsert_summary_embedding_conn( + conn: &rusqlite::Connection, summary_id: &str, model_signature: &str, embedding: &[f32], @@ -304,21 +309,48 @@ pub fn set_summary_embedding_for_signature( let blob = pack_embedding_blob(embedding); let dim = i64::try_from(embedding.len()).context("embedding dimension does not fit i64")?; let created_at = Utc::now().timestamp_millis() as f64 / 1000.0; - with_connection(config, |conn| { - conn.execute( - "INSERT INTO mem_tree_summary_embeddings + conn.execute( + "INSERT INTO mem_tree_summary_embeddings (summary_id, model_signature, vector, dim, created_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(summary_id, model_signature) DO UPDATE SET vector = excluded.vector, dim = excluded.dim, created_at = excluded.created_at", - params![summary_id, model_signature, blob, dim, created_at], - )?; - Ok(()) + params![summary_id, model_signature, blob, dim, created_at], + )?; + Ok(()) +} + +/// Store a summary embedding for a specific provider/model/dimension signature. +/// +/// Per-model table write path for #1574. The legacy +/// `mem_tree_summaries.embedding` column is intentionally left untouched by +/// this helper (read by the §7 migration; dropped only in a later release). +pub fn set_summary_embedding_for_signature( + config: &Config, + summary_id: &str, + model_signature: &str, + embedding: &[f32], +) -> Result<()> { + with_connection(config, |conn| { + upsert_summary_embedding_conn(conn, summary_id, model_signature, embedding) }) } +/// Transaction-scoped variant of [`set_summary_embedding_for_signature`], for +/// the seal path which inserts the summary row and its embedding in one tx +/// (#1574 write-side cutover). Opening a fresh connection there would break +/// atomicity / deadlock on the busy DB. +pub(crate) fn set_summary_embedding_for_signature_tx( + tx: &rusqlite::Transaction<'_>, + summary_id: &str, + model_signature: &str, + embedding: &[f32], +) -> Result<()> { + upsert_summary_embedding_conn(tx, summary_id, model_signature, embedding) +} + /// Fetch a summary embedding for exactly one provider/model/dimension signature. pub fn get_summary_embedding_for_signature( config: &Config, diff --git a/src/openhuman/memory/tree/tree_source/store_tests.rs b/src/openhuman/memory/tree/tree_source/store_tests.rs index 9ec665a1b..9570e9364 100644 --- a/src/openhuman/memory/tree/tree_source/store_tests.rs +++ b/src/openhuman/memory/tree/tree_source/store_tests.rs @@ -74,7 +74,7 @@ fn summary_insert_and_fetch() { let node = sample_summary("sum-1", "tree-1", 1); with_connection(&cfg, |conn| { let tx = conn.unchecked_transaction()?; - insert_summary_tx(&tx, &node, None)?; + insert_summary_tx(&tx, &node, None, "test")?; tx.commit()?; Ok(()) }) @@ -93,8 +93,8 @@ fn summary_insert_is_idempotent_on_id() { let node = sample_summary("sum-1", "tree-1", 1); with_connection(&cfg, |conn| { let tx = conn.unchecked_transaction()?; - insert_summary_tx(&tx, &node, None)?; - insert_summary_tx(&tx, &node, None)?; + insert_summary_tx(&tx, &node, None, "test")?; + insert_summary_tx(&tx, &node, None, "test")?; tx.commit()?; Ok(()) }) @@ -109,7 +109,7 @@ fn summary_embeddings_are_scoped_by_model_signature() { let node = sample_summary("sum-embed", "tree-1", 1); with_connection(&cfg, |conn| { let tx = conn.unchecked_transaction()?; - insert_summary_tx(&tx, &node, None)?; + insert_summary_tx(&tx, &node, None, "test")?; tx.commit()?; Ok(()) }) @@ -143,7 +143,25 @@ fn summary_embeddings_are_scoped_by_model_signature() { .unwrap() .is_none() ); + + // #1574 cutover: the public `get_summary_embedding` now reads the sidecar + // at the *active* signature (not the legacy column). Nothing is written + // there yet → absent; never a cross-space read of the rows above. assert!(get_summary_embedding(&cfg, "sum-embed").unwrap().is_none()); + + // The public setter targets the active signature and round-trips through + // the public getter — proves the cutover wiring end to end. + set_summary_embedding(&cfg, "sum-embed", &[0.7, 0.8]).unwrap(); + assert_eq!( + get_summary_embedding(&cfg, "sum-embed").unwrap(), + Some(vec![0.7, 0.8]) + ); + + // ...and the earlier per-signature rows remain independently scoped. + assert_eq!( + get_summary_embedding_for_signature(&cfg, "sum-embed", "local/bge-small@384").unwrap(), + Some(vec![0.3, 0.4, 0.5]) + ); } #[test] diff --git a/tests/linux_cef_deb_runtime_e2e.rs b/tests/linux_cef_deb_runtime_e2e.rs index baf6c8aa7..2ef1ddd87 100644 --- a/tests/linux_cef_deb_runtime_e2e.rs +++ b/tests/linux_cef_deb_runtime_e2e.rs @@ -12,24 +12,47 @@ use std::fs; use std::io::Write; use std::path::PathBuf; +use std::sync::{Mutex, MutexGuard}; -/// Guard to temporarily set/unset environment variables. +/// Serializes every env-mutating test in this file. `std::env` is +/// process-global; cargo runs these tests in parallel within one binary, +/// and several mutate the SAME vars (e.g. `OPENHUMAN_CORE_BIN`), so without +/// this an interleaving made one test read back another's value and the +/// `assert_eq!` flaked. `EnvGuard` holds this lock for its whole lifetime, +/// so at most one env-mutating test runs at a time. Poison-safe (a +/// panicking test must not cascade-fail every later one). +static ENV_LOCK: Mutex<()> = Mutex::new(()); + +/// Guard to temporarily set/unset environment variables. Acquires +/// [`ENV_LOCK`] for its lifetime so concurrent tests can't observe each +/// other's mutations. struct EnvGuard { key: &'static str, old: Option, + _lock: MutexGuard<'static, ()>, } impl EnvGuard { fn set(key: &'static str, value: &str) -> Self { + let lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let old = std::env::var(key).ok(); std::env::set_var(key, value); - Self { key, old } + Self { + key, + old, + _lock: lock, + } } fn unset(key: &'static str) -> Self { + let lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let old = std::env::var(key).ok(); std::env::remove_var(key); - Self { key, old } + Self { + key, + old, + _lock: lock, + } } }