feat(memory): MemoryProvider abstraction + Snowflake backend — 2.7.0

Introduces a pluggable MemoryBackend system mirroring the LLMBackend/
LLMRouter pattern, so users can swap memory stores via config or env var.

New backends: SQLite (wraps existing MemoryStore), Redis (wraps
RedisMemoryStore), Snowflake (native VECTOR(FLOAT,1536) columns +
VECTOR_COSINE_SIMILARITY), Vektori/pgvector (default prod backend),
Platform (compat wrapper).

Factory: get_memory_provider() resolves from [memory] provider field,
DEVSPER_MEMORY_PROVIDER env var, or legacy backend field. Defaults to
Vektori; falls back to SQLite with a warning when DATABASE_URL is unset.

Snowflake credentials: resolved exclusively via the devsper credential
store (devsper credentials set snowflake password) or SNOWFLAKE_* env
vars — never stored in config files. Full CRUD + native vector search
tested live against Snowflake Enterprise (AWS, RZDAUHB-MV83636).

Also fixes two pre-existing test failures: test_pipelines used a
removed internal _tools attribute (replaced with new deregister()),
and test_e2e_redis_loop crashed on missing redis import before the
pytest.skip() could fire.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: rithulkamesh

CHANGELOG.md

@@ -7,6 +7,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.7.0] — 2026-04-07
+
+### Added
+
+- **MemoryProvider abstraction** — Introduced `MemoryBackend` ABC (`devsper/memory/providers/base.py`) mirroring the `LLMBackend` pattern. All memory backends implement a unified async interface: `store`, `retrieve`, `delete`, `list_memory`, `list_all_ids`, `query_similar`, `health`.
+- **MemoryProvider factory** — `get_memory_provider()` singleton factory resolves the active backend from config (`[memory] provider`), `DEVSPER_MEMORY_PROVIDER` env var, or legacy `backend` field. Defaults to Vektori (pgvector) with automatic SQLite fallback when `DATABASE_URL` is not set.
+- **Snowflake memory backend** — New `SnowflakeBackend` using `VECTOR(FLOAT, 1536)` columns and `VECTOR_COSINE_SIMILARITY` for native semantic search. Credentials resolved exclusively via the devsper credential store or `SNOWFLAKE_*` env vars — never from config files.
+- **SQLite, Redis, Vektori, Platform backends** — Existing stores wrapped as `MemoryBackend` implementations with `get_sync_store()` escape hatch for legacy sync callers.
+- **Credential store: Snowflake + Redis** — Added `snowflake` and `redis_memory` providers to keyring mappings. `devsper credentials set snowflake password` stores the Snowflake password securely; `inject_into_env()` propagates all `SNOWFLAKE_*` vars automatically.
+- **Config schema** — Added `RedisMemoryConfig` and `SnowflakeMemoryConfig` sub-models under `[memory]`. `SnowflakeMemoryConfig` has no `password` field — credentials are credential-store-only.
+- **`devsper doctor` memory provider health check** — `run_doctor()` now calls `provider.health()` and reports the active backend name and status.
+- **`devsper[snowflake]` and `devsper[redis-memory]` extras** — `snowflake-connector-python>=3.6.0` and `redis>=5.0.0` optional dependency groups.
+- **`deregister(name)` in tool registry** — Public function to remove a single tool by name without clearing the whole registry.
+- **`MemoryIndex` native search delegation** — When the active backend has `supports_native_vector_search=True`, `query_memory()` and `query_across_runs()` delegate directly to `backend.query_similar()`, bypassing in-process cosine ranking.
+
+### Changed
+
+- **`MemoryRouter` and `MemoryIndex`** accept an optional `backend: MemoryBackend` parameter. The `_build_memory_store()` fallback now calls the factory instead of constructing `MemoryStore()` directly.
+- **`get_effective_memory_store()`** preserved as a backwards-compatible alias; new `get_effective_memory_backend()` returns the full `MemoryBackend`. Async-only backends (Vektori, Snowflake) are bridged via `_AsyncBridgeStore` for sync callers.
+- **Legacy `backend` values** (`local`, `supermemory`, `hybrid`) now map to `"vektori"` as the default production store.
+
+### Fixed
+
+- **`test_pipelines.py`** — Replaced `registry._tools.pop()` (broken module-level access) with new `deregister()` function.
+- **`test_e2e_redis_loop.py`** — Import guard prevents `ModuleNotFoundError` when `redis` is not installed; test now correctly skips instead of failing.
+
 ## [2.6.0] — 2026-04-06
 
 ### Added
@@ -82,7 +108,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - **Supermemory ranking backend** — `[memory] backend = "supermemory"` (default) uses the Rust `supermemory-core` ranker for hybrid lexical (+ optional embedding) memory retrieval, deduplication, and recency-aware context formatting.
 - **Platform memory backends** — `[memory] backend` can be `platform` or `hybrid` with `platform_api_url` and `platform_org_slug`; config resolver honors `DEVSPER_PLATFORM_API_URL` and `DEVSPER_PLATFORM_ORG` when set.
-- **Remote platform workflows** — `load_workflow("org/pkg@N")` fetches workflow specs from the Devsper Platform API when `DEVSPER_PLATFORM_API_URL`, `DEVSPER_PLATFORM_ORG`, and (when required) `DEVSPER_PLATFORM_TOKEN` are configured.
+- **Remote platform workflows** — `load_workflow("org/pkg@N")` fetches workflow specs from the Devsper Cloud API when `DEVSPER_PLATFORM_API_URL`, `DEVSPER_PLATFORM_ORG`, and (when required) `DEVSPER_PLATFORM_TOKEN` are configured.
 - **Local worker pool** — `devsper pool` subcommands to run the local worker pool manager (foreground spawner) alongside improved swarm execution when using pool-backed paths.
 - **Devsper Cloud CLI** — `devsper cloud login`, `logout`, `run`, `status`, and `logs` for JWT authentication (OS keychain) and runs against the platform API.
 

devsper/cli/init.py

@@ -534,6 +534,31 @@ def run_doctor() -> int:
     except Exception as e:
         warnings.append(f"Memory store: {e}")
 
+    # Memory provider health check
+    try:
+        import asyncio
+        from devsper.memory.providers.factory import get_memory_provider
+        provider = get_memory_provider()
+        try:
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                import concurrent.futures
+                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+                    healthy = pool.submit(asyncio.run, provider.health()).result()
+            else:
+                healthy = loop.run_until_complete(provider.health())
+        except Exception:
+            healthy = False
+        if healthy:
+            ok.append(f"Memory provider: {provider.name} — healthy")
+        else:
+            warnings.append(
+                f"Memory provider: {provider.name} — health check failed. "
+                "Check credentials and connectivity."
+            )
+    except Exception as e:
+        warnings.append(f"Memory provider health check failed: {e}")
+
     try:
         from devsper.knowledge.knowledge_graph import KnowledgeGraph
         kg = KnowledgeGraph(store=get_default_store())

devsper/config/schema.py

@@ -74,13 +74,46 @@ class ModelsConfig(BaseModel):
     quality: str | None = None  # v1.6: complex tier (defaults to planner)
 
 
+class RedisMemoryConfig(BaseModel):
+    """Connection config for Redis memory backend."""
+
+    redis_url: str = "redis://localhost:6379"
+    run_id: str = ""  # empty = auto-generated from os.getpid()
+
+
+class SnowflakeMemoryConfig(BaseModel):
+    """Connection config for Snowflake memory backend.
+
+    Credentials (account, user, password, etc.) are resolved from:
+      1. devsper credential store (devsper credentials set snowflake <key>)
+      2. SNOWFLAKE_* environment variables
+      3. These TOML fields as a last fallback (non-secret fields only)
+
+    NEVER put passwords in config files — use the credential store.
+    """
+
+    account: str = ""       # or SNOWFLAKE_ACCOUNT env var
+    user: str = ""          # or SNOWFLAKE_USER env var
+    database: str = ""      # or SNOWFLAKE_DATABASE env var
+    schema_name: str = ""   # "schema" is reserved by Pydantic; or SNOWFLAKE_SCHEMA
+    warehouse: str = ""     # or SNOWFLAKE_WAREHOUSE env var
+    role: str = ""          # or SNOWFLAKE_ROLE env var
+    table: str = "devsper_memory"
+
+
 class MemoryConfig(BaseModel):
     enabled: bool = True
     store_results: bool = True
     top_k: int = 5
+    # New explicit provider field. Takes precedence over legacy `backend`.
+    # Values: "sqlite" | "redis" | "snowflake" | "vektori"
+    # Empty string = auto-detect (vektori if DATABASE_URL set, else sqlite).
+    provider: str = ""
     backend: Literal["local", "platform", "hybrid", "supermemory"] = "supermemory"
     platform_api_url: str = ""
     platform_org_slug: str = ""
+    redis: RedisMemoryConfig = Field(default_factory=RedisMemoryConfig)
+    snowflake: SnowflakeMemoryConfig = Field(default_factory=SnowflakeMemoryConfig)
 
 
 class KnowledgeConfig(BaseModel):

devsper/credentials/cli.py

@@ -22,10 +22,12 @@
     "gemini": ["api_key"],
     "azure": ["endpoint", "api_key", "deployment", "api_version"],
     "azure_anthropic": ["endpoint", "api_key", "deployment"],
+    "snowflake": ["account", "user", "password", "database", "schema", "warehouse", "role"],
+    "redis_memory": ["url"],
 }
 
 # Keys that must never be shown in list (only "(stored)"); all others can show value
-SENSITIVE_KEYS = {"api_key", "token"}
+SENSITIVE_KEYS = {"api_key", "token", "password"}
 
 # (provider, key) -> env var name for export
 PROVIDER_KEY_TO_ENV: dict[tuple[str, str], str] = {
@@ -40,6 +42,14 @@
     ("azure_anthropic", "endpoint"): "AZURE_ANTHROPIC_ENDPOINT",
     ("azure_anthropic", "api_key"): "AZURE_ANTHROPIC_API_KEY",
     ("azure_anthropic", "deployment"): "AZURE_ANTHROPIC_DEPLOYMENT_NAME",
+    ("snowflake", "account"): "SNOWFLAKE_ACCOUNT",
+    ("snowflake", "user"): "SNOWFLAKE_USER",
+    ("snowflake", "password"): "SNOWFLAKE_PASSWORD",
+    ("snowflake", "database"): "SNOWFLAKE_DATABASE",
+    ("snowflake", "schema"): "SNOWFLAKE_SCHEMA",
+    ("snowflake", "warehouse"): "SNOWFLAKE_WAREHOUSE",
+    ("snowflake", "role"): "SNOWFLAKE_ROLE",
+    ("redis_memory", "url"): "REDIS_URL",
 }
 
 

devsper/credentials/store.py

@@ -19,6 +19,15 @@
     ("azure_anthropic", "api_key", "AZURE_ANTHROPIC_API_KEY"),
     ("azure_anthropic", "endpoint", "AZURE_ANTHROPIC_ENDPOINT"),
     ("azure_anthropic", "deployment", "AZURE_ANTHROPIC_DEPLOYMENT_NAME"),
+    # Memory backends
+    ("snowflake", "account", "SNOWFLAKE_ACCOUNT"),
+    ("snowflake", "user", "SNOWFLAKE_USER"),
+    ("snowflake", "password", "SNOWFLAKE_PASSWORD"),
+    ("snowflake", "database", "SNOWFLAKE_DATABASE"),
+    ("snowflake", "schema", "SNOWFLAKE_SCHEMA"),
+    ("snowflake", "warehouse", "SNOWFLAKE_WAREHOUSE"),
+    ("snowflake", "role", "SNOWFLAKE_ROLE"),
+    ("redis_memory", "url", "REDIS_URL"),
 ]
 
 SERVICE_NAME = "devsper"
@@ -31,6 +40,8 @@
     "gemini": ["api_key"],
     "azure": ["endpoint", "api_key", "deployment", "api_version"],
     "azure_anthropic": ["endpoint", "api_key", "deployment"],
+    "snowflake": ["account", "user", "password", "database", "schema", "warehouse", "role"],
+    "redis_memory": ["url"],
 }
 
 

devsper/memory/context.py

@@ -33,14 +33,95 @@ def detach_memory_context(
     _current_run_id.reset(tokens[2])
 
 
+def get_effective_memory_backend():
+    """
+    Return the active MemoryBackend for the current context.
+    Prefers the context-var store (set by attach_memory_context) if it is a MemoryBackend,
+    otherwise returns the process-level singleton from the factory.
+    """
+    s = _current_store.get()
+    if s is not None and hasattr(s, "health"):
+        return s  # it is already a MemoryBackend
+    from devsper.memory.providers.factory import get_memory_provider
+
+    return get_memory_provider()
+
+
 def get_effective_memory_store():
-    """Prefer agent/router store when set; else process default SQLite store."""
+    """
+    Legacy: return a synchronous MemoryStore compatible object.
+    Backends that wrap a sync store (sqlite, redis, platform) expose get_sync_store().
+    Async-only backends (vektori, snowflake) return an _AsyncBridgeStore shim.
+    """
     s = _current_store.get()
-    if s is not None:
+    if s is not None and not hasattr(s, "health"):
+        # Legacy MemoryStore object set directly via attach_memory_context
         return s
-    from devsper.memory.memory_store import get_default_store
+    backend = get_effective_memory_backend()
+    if hasattr(backend, "get_sync_store"):
+        return backend.get_sync_store()
+    # Async-only backend: return a sync bridge so legacy callers still work
+    return _AsyncBridgeStore(backend)
+
+
+class _AsyncBridgeStore:
+    """
+    Sync shim for async-only backends (VektoriBackend, SnowflakeBackend).
+    Runs coroutines via asyncio in a thread so sync callers don't need to be async.
+    This mirrors the _run_async() pattern already used in server/memory_utils.py.
+    """
+
+    def __init__(self, backend) -> None:
+        self._backend = backend
+
+    def _run(self, coro):
+        import asyncio
+
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        if loop and loop.is_running():
+            import concurrent.futures
+
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+                return pool.submit(asyncio.run, coro).result()
+        return asyncio.run(coro)
+
+    def store(self, record, namespace=None):
+        return self._run(self._backend.store(record, namespace))
+
+    def retrieve(self, memory_id, namespace=None):
+        try:
+            return self._run(self._backend.retrieve(memory_id, namespace))
+        except NotImplementedError:
+            return None
+
+    def delete(self, memory_id, namespace=None):
+        try:
+            return self._run(self._backend.delete(memory_id, namespace))
+        except NotImplementedError:
+            return False
+
+    def list_memory(
+        self,
+        memory_type=None,
+        limit=100,
+        offset=0,
+        tag_contains=None,
+        include_archived=False,
+        run_id_filter=None,
+        namespace=None,
+    ):
+        return self._run(
+            self._backend.list_memory(
+                memory_type, limit, offset, tag_contains,
+                include_archived, run_id_filter, namespace,
+            )
+        )
 
-    return get_default_store()
+    def list_all_ids(self, memory_type=None, namespace=None):
+        return self._run(self._backend.list_all_ids(memory_type, namespace))
 
 
 def get_effective_memory_namespace() -> str | None:

devsper/memory/memory_index.py

@@ -2,11 +2,18 @@
 Semantic search across stored memory via embeddings and top_k retrieval.
 """
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from devsper.memory.embeddings import embed_text
 from devsper.memory.memory_store import MemoryStore
 from devsper.memory.memory_types import MemoryRecord
 from devsper.memory.supermemory_rust_ranker import rank_memories
 
+if TYPE_CHECKING:
+    from devsper.memory.providers.base import MemoryBackend
+
 
 def _cosine_sim(a: list[float], b: list[float]) -> float:
     if not a or not b or len(a) != len(b):
@@ -19,14 +26,45 @@ def _cosine_sim(a: list[float], b: list[float]) -> float:
     return dot / (na * nb)
 
 
+def _run_async_in_thread(coro):
+    """Run a coroutine synchronously, safely bridging from sync context."""
+    import asyncio
+    import concurrent.futures
+
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = None
+    if loop and loop.is_running():
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            return pool.submit(asyncio.run, coro).result()
+    return asyncio.run(coro)
+
+
 class MemoryIndex:
     """
     Vector search over memory. Uses store for persistence and optional
     embeddings on records for query_memory(text, top_k).
+
+    When a MemoryBackend with supports_native_vector_search=True is provided,
+    query_memory() and query_across_runs() delegate to the backend's native
+    vector search (e.g. Snowflake VECTOR_COSINE_SIMILARITY, pgvector <=>),
+    bypassing in-process cosine ranking.
     """
 
-    def __init__(self, store: MemoryStore | None = None, ranking_backend: str = "local") -> None:
-        self.store = store or MemoryStore()
+    def __init__(
+        self,
+        store: MemoryStore | None = None,
+        ranking_backend: str = "local",
+        backend: "MemoryBackend | None" = None,
+    ) -> None:
+        self._backend = backend
+        if store is not None:
+            self.store = store
+        elif backend is not None and hasattr(backend, "get_sync_store"):
+            self.store = backend.get_sync_store()
+        else:
+            self.store = store or MemoryStore()
         self.ranking_backend = ranking_backend
 
     def query_memory(
@@ -39,6 +77,7 @@ def query_memory(
     ) -> list[MemoryRecord]:
         """
         Semantic search via ranking strategy:
+        - native (vektori/snowflake): delegates to backend.query_similar() directly.
         - local (default): embed query, cosine-rank records that have embeddings.
         - supermemory: hybrid local ranking (lexical token overlap + optional embedding
           cosine similarity) using `rank_memories()`; records without embeddings can
@@ -48,6 +87,22 @@ def query_memory(
         Use min_similarity > 0 (e.g. 0.45) to avoid injecting barely-related memory.
         By default excludes archived records (consolidation).
         """
+        # Fast path: delegate to native vector search when supported
+        if self._backend is not None and self._backend.supports_native_vector_search:
+            try:
+                from devsper.memory.providers.base import MemoryQuery
+
+                query = MemoryQuery(
+                    text=text,
+                    top_k=top_k,
+                    min_similarity=min_similarity,
+                    namespace=namespace,
+                    include_archived=include_archived,
+                )
+                return _run_async_in_thread(self._backend.query_similar(query))
+            except Exception:
+                pass  # fall through to in-process ranking
+
         records = self.store.list_memory(
             limit=500, include_archived=include_archived, namespace=namespace
         )
@@ -112,6 +167,22 @@ def query_across_runs(
         v1.8: Same as query_memory but over more records (all runs), optional run_id filter.
         Used by CrossRunSynthesizer. Excludes archived by default.
         """
+        # Fast path: delegate to native vector search when supported
+        if self._backend is not None and self._backend.supports_native_vector_search:
+            try:
+                from devsper.memory.providers.base import MemoryQuery
+
+                query = MemoryQuery(
+                    text=text,
+                    top_k=top_k,
+                    min_similarity=min_similarity,
+                    namespace=namespace,
+                    include_archived=include_archived,
+                )
+                return _run_async_in_thread(self._backend.query_similar(query))
+            except Exception:
+                pass  # fall through to in-process ranking
+
         records = self.store.list_memory(
             limit=2000,
             include_archived=include_archived,