diff --git a/README.md b/README.md index a08c866..8175218 100644 --- a/README.md +++ b/README.md @@ -405,6 +405,10 @@ All settings are environment-driven (or `.env`): | `MNEMA_DREAM_ENABLED` | `false` | Auto Dream background consolidation | | `MNEMA_DREAM_INTERVAL_SECONDS` | `3600` | Seconds between dream cycles | | `MNEMA_DREAM_DECAY_THRESHOLD` | `0.05` | Decay cutoff for forgetting during dreams | +| `MNEMA_SMART_FORGET_ENABLED` | `false` | Opt-in LLM veto on decay deletions (see [Smart forgetting](#-smart-forgetting-opt-in)) | +| `MNEMA_JUDGE_MODEL` | — | Chat model for the smart-forgetting judge (required when enabled) | +| `MNEMA_JUDGE_API_KEY` | — | Judge endpoint key (only sent when set) | +| `MNEMA_JUDGE_BASE_URL` | `https://api.openai.com/v1` | OpenAI-compatible judge endpoint | | `MNEMA_LOG_LEVEL` | `WARNING` | `DEBUG` \| `INFO` \| `WARNING` \| `ERROR` — verbose logs for bug reports | --- @@ -565,6 +569,34 @@ score = 0.7·vector + 0.2·keyword + 0.1·decay where `decay = recency(half-life) × frequency × importance`. +### 🤖 Smart forgetting (opt-in) + +By default Mnema is **deliberately LLM-free**: decay deletions come from the +formula above alone. If you want a second opinion on borderline memories, you +can opt in to an LLM judge that vets only the records the formula already +selected for deletion: + +- The judge can only **rescue** — it never adds new deletion candidates. +- **Fail-safe = KEEP** — any judge error, timeout, or unparseable answer + keeps the memory. +- Disabled by default; when `MNEMA_SMART_FORGET_ENABLED` is unset/`false`, + no judge is constructed and **no LLM call is ever made**. + +| Variable | Default | Description | +|---|---|---| +| `MNEMA_SMART_FORGET_ENABLED` | `false` | Master opt-in for the LLM judge | +| `MNEMA_JUDGE_MODEL` | — | Chat model id (required when enabled) | +| `MNEMA_JUDGE_API_KEY` | — | Sent as `Authorization: Bearer` only when set | +| `MNEMA_JUDGE_BASE_URL` | `https://api.openai.com/v1` | Any OpenAI-compatible endpoint | + +Fully local example with Ollama (no paid key needed): + +```bash +export MNEMA_SMART_FORGET_ENABLED=true +export MNEMA_JUDGE_BASE_URL=http://localhost:11434/v1 +export MNEMA_JUDGE_MODEL=qwen3:8b +``` + --- ## 🗺️ Roadmap diff --git a/packages/mnema-python/src/mnema/config.py b/packages/mnema-python/src/mnema/config.py index 20a8e92..dbfd4cd 100644 --- a/packages/mnema-python/src/mnema/config.py +++ b/packages/mnema-python/src/mnema/config.py @@ -175,6 +175,34 @@ class MnemaConfig(BaseSettings): ), ) + # --- Smart forgetting (opt-in LLM judge) ----------------------------- + smart_forget_enabled: bool = Field( + default=False, + description=( + "Opt-in: let a configured LLM veto decay-sweep deletions. When " + "False (default) no judge is constructed and no LLM is ever " + "called." + ), + ) + judge_model: str | None = Field( + default=None, + description="Chat model id for the smart-forgetting judge (required when enabled)", + ) + judge_api_key: str | None = Field( + default=None, + description=( + "API key for the judge endpoint, sent as 'Authorization: Bearer' " + "only when set (local servers like Ollama need none)" + ), + ) + judge_base_url: str = Field( + default="https://api.openai.com/v1", + description=( + "OpenAI-compatible base URL for the judge endpoint " + "(use http://localhost:11434/v1 for Ollama)" + ), + ) + # --- Logging / diagnostics ------------------------------------------- log_level: str = Field( default="WARNING", @@ -222,6 +250,10 @@ def validate_runtime(self) -> None: raise ConfigError( "embedding='nomic' requires MNEMA_NOMIC_API_KEY to be set." ) + if self.smart_forget_enabled and not self.judge_model: + raise ConfigError( + "smart_forget_enabled=True requires MNEMA_JUDGE_MODEL to be set." + ) def load_config(**overrides: object) -> MnemaConfig: diff --git a/packages/mnema-python/src/mnema/judge.py b/packages/mnema-python/src/mnema/judge.py new file mode 100644 index 0000000..83baf43 --- /dev/null +++ b/packages/mnema-python/src/mnema/judge.py @@ -0,0 +1,167 @@ +"""Smart forgetting — an opt-in LLM veto on the decay sweep. + +Mnema is *deliberately* LLM-free by default (see :mod:`mnema.summarize`). +This module is the single, strictly opt-in exception: when +``MNEMA_SMART_FORGET_ENABLED=true``, each memory the pure decay formula has +already selected for deletion (``score <= threshold``) is asked KEEP/FORGET +by a configured chat model, and only explicit FORGET verdicts are deleted. + +Two conservative guarantees hold throughout: + +* The LLM can only **rescue**, never condemn — memories above the decay + threshold are never shown to the judge. +* **Fail-safe = KEEP** — any error, timeout, HTTP failure, or unparseable + response means the memory is kept. Deletion on ambiguity is forbidden. + +The pure helpers (:func:`build_forget_prompt`, :func:`parse_forget_verdict`) +do no I/O so they are trivially testable; :class:`OpenAICompatibleJudge` is a +thin async httpx client modelled on the Ollama embedding provider. +""" + +from __future__ import annotations + +import logging +from abc import ABC, abstractmethod + +import httpx + +from mnema.config import MnemaConfig +from mnema.models import MemoryRecord + +logger = logging.getLogger("mnema.judge") + + +def build_forget_prompt(record: MemoryRecord, decay_score: float) -> str: + """Build the KEEP/FORGET prompt for one decay-sweep candidate. Pure.""" + tags = ", ".join(record.tags) if record.tags else "(none)" + return ( + "You are the forgetting judge of a long-term memory store. A memory " + "has been selected for deletion by a heuristic decay formula. Decide " + "whether it is truly no longer relevant.\n" + "\n" + "Answer with a first token of exactly KEEP or FORGET, optionally " + "followed by a one-line reason.\n" + "\n" + f"Memory: {record.text}\n" + f"Tags: {tags}\n" + f"Importance (1-10): {int(record.importance)}\n" + f"Decay score (0-1, lower means more forgotten): {decay_score:.3f}\n" + ) + + +def parse_forget_verdict(text: str) -> bool | None: + """Parse the judge's answer. Pure. + + Returns ``True`` for FORGET, ``False`` for KEEP, and ``None`` for + anything unparseable (empty, ambiguous, or a near-miss like + "FORGETFUL") — callers must treat ``None`` as KEEP. + """ + stripped = (text or "").strip() + if not stripped: + return None + first_token = stripped.split(maxsplit=1)[0].lower() + if first_token == "forget": + return True + if first_token == "keep": + return False + return None + + +class MemoryJudge(ABC): + """Pluggable KEEP/FORGET judge for decay-sweep candidates. + + Implementations must **never raise** from :meth:`should_forget` — any + failure returns ``False`` (keep) and logs the reason. + """ + + @abstractmethod + async def should_forget(self, record: MemoryRecord, decay_score: float) -> bool: + """Return True only when the judge confidently votes FORGET.""" + + async def aclose(self) -> None: + """Release resources. Default no-op.""" + return None + + +class OpenAICompatibleJudge(MemoryJudge): + """Judge backed by any OpenAI-compatible chat completions endpoint. + + With ``judge_base_url=http://localhost:11434/v1`` this works against + Ollama and other local servers, so no paid API key is ever required. + """ + + def __init__(self, config: MnemaConfig) -> None: + self._model = config.judge_model + self._base_url = config.judge_base_url.rstrip("/") + headers = {} + if config.judge_api_key: + headers["Authorization"] = f"Bearer {config.judge_api_key}" + self._client = httpx.AsyncClient(base_url=self._base_url, headers=headers) + + async def should_forget(self, record: MemoryRecord, decay_score: float) -> bool: + """Ask the endpoint KEEP/FORGET; fail-safe to KEEP on any problem.""" + try: + response = await self._client.post( + "/chat/completions", + json={ + "model": self._model, + "messages": [ + { + "role": "user", + "content": build_forget_prompt(record, decay_score), + } + ], + "temperature": 0, + "max_tokens": 64, + }, + ) + response.raise_for_status() + data = response.json() + content = data["choices"][0]["message"]["content"] + except Exception as exc: + logger.info( + "smart-forget judge call failed for memory %s; keeping it: %s", + record.id, + exc, + ) + return False + + verdict = parse_forget_verdict(content if isinstance(content, str) else "") + if verdict is None: + logger.info( + "smart-forget judge returned an unparseable verdict for memory " + "%s; keeping it: %r", + record.id, + content, + ) + return False + logger.debug( + "smart-forget judge verdict for memory %s: %s (%r)", + record.id, + "FORGET" if verdict else "KEEP", + content, + ) + return verdict + + async def aclose(self) -> None: + await self._client.aclose() + + +def make_judge(config: MnemaConfig) -> MemoryJudge | None: + """Build a judge from config, or ``None`` when smart mode is disabled. + + When ``smart_forget_enabled`` is False (the default), no judge is + constructed and no LLM call is possible. + """ + if not config.smart_forget_enabled: + return None + return OpenAICompatibleJudge(config) + + +__all__ = [ + "MemoryJudge", + "OpenAICompatibleJudge", + "build_forget_prompt", + "make_judge", + "parse_forget_verdict", +] diff --git a/packages/mnema-python/src/mnema/service.py b/packages/mnema-python/src/mnema/service.py index 5c4299a..1cce5d0 100644 --- a/packages/mnema-python/src/mnema/service.py +++ b/packages/mnema-python/src/mnema/service.py @@ -8,6 +8,7 @@ from __future__ import annotations import contextlib +import logging import time from collections.abc import Callable, Sequence @@ -16,6 +17,7 @@ from mnema.decay import DecayParams, combine, decay_score from mnema.embeddings import EmbeddingProvider, make_embedding from mnema.errors import MemoryNotFoundError, ScopeError +from mnema.judge import MemoryJudge, make_judge from mnema.models import ( Importance, MemoryRecord, @@ -26,6 +28,8 @@ ) from mnema.summarize import SummarizationPlan, plan_summarization +logger = logging.getLogger("mnema.service") + def _coerce_importance(value: int | Importance) -> Importance: """Convert any int 1–10 to the nearest :class:`Importance` level. @@ -55,11 +59,13 @@ def __init__( *, backend: VectorBackend | None = None, embedding: EmbeddingProvider | None = None, + judge: MemoryJudge | None = None, ) -> None: self.config = config or load_config() self.config.validate_runtime() self._backend = backend or make_backend(self.config) self._embedding = embedding or make_embedding(self.config) + self._judge = judge or make_judge(self.config) self._decay = DecayParams( half_life_days=self.config.decay_half_life_days, floor=self.config.decay_floor, @@ -278,6 +284,10 @@ async def apply_decay( ) -> list[MemoryRecord]: """Compute decay scores and (optionally) forget below ``threshold``. + When smart forgetting is enabled (``MNEMA_SMART_FORGET_ENABLED``), + each candidate is additionally vetted by the configured judge, which + can only rescue memories — never condemn new ones. + Args: scope: Restrict the sweep to one scope, or None for all. threshold: Decay score below which a memory is a forget candidate. @@ -300,11 +310,44 @@ async def apply_decay( if score <= threshold: candidates.append(record.model_copy(update={"score": score})) + if self._judge is not None: + candidates = await self._veto_with_judge(candidates) + if not dry_run: for c in candidates: await self._backend.delete(c.id) return candidates + async def _veto_with_judge( + self, candidates: list[MemoryRecord] + ) -> list[MemoryRecord]: + """Let the smart-forgetting judge veto (rescue) candidates. + + The judge can only narrow the candidate list, never widen it, and a + judge failure always keeps the memory (fail-safe = KEEP). + """ + judge = self._judge + if judge is None: + return candidates + condemned: list[MemoryRecord] = [] + for c in candidates: + score = float(c.score) if c.score is not None else 0.0 + try: + forget = await judge.should_forget(c, score) + except Exception as exc: + # A conforming judge never raises; if one does, fail safe. + logger.warning( + "smart-forget judge raised for memory %s; keeping it: %s", + c.id, + exc, + ) + forget = False + if forget: + condemned.append(c) + else: + logger.info("smart-forget judge rescued memory %s", c.id) + return condemned + async def summarize( self, *, @@ -324,6 +367,8 @@ async def summarize( async def aclose(self) -> None: await self._embedding.aclose() await self._backend.aclose() + if self._judge is not None: + await self._judge.aclose() # --- private -------------------------------------------------------- async def _touch(self, record: MemoryRecord) -> None: diff --git a/packages/mnema-python/tests/fakes.py b/packages/mnema-python/tests/fakes.py index 8f73f94..5367ab8 100644 --- a/packages/mnema-python/tests/fakes.py +++ b/packages/mnema-python/tests/fakes.py @@ -19,6 +19,7 @@ from mnema.backends.base import BackendHit, BackendQuery, VectorBackend from mnema.config import MnemaConfig from mnema.embeddings.base import EmbeddingProvider +from mnema.judge import MemoryJudge from mnema.models import Importance, MemoryRecord @@ -157,6 +158,37 @@ def _keyword(record_tags: Sequence[str], query_tags: Sequence[str] | None) -> fl return inter / len(a | b) if inter else 0.0 +# --------------------------------------------------------------------------- +# Scripted judge for smart-forgetting tests. +# --------------------------------------------------------------------------- +class FakeJudge(MemoryJudge): + """A scripted :class:`MemoryJudge` for tests. + + ``verdicts`` maps memory id → FORGET (True) / KEEP (False); anything not + in the map gets ``default``. When ``raises`` is set, ``should_forget`` + raises it instead (to prove the service fails safe). Every invocation is + recorded in ``calls`` as ``(memory_id, decay_score)``. + """ + + def __init__( + self, + verdicts: dict[str, bool] | None = None, + *, + default: bool = True, + raises: Exception | None = None, + ) -> None: + self.verdicts = dict(verdicts or {}) + self.default = default + self.raises = raises + self.calls: list[tuple[str, float]] = [] + + async def should_forget(self, record: MemoryRecord, decay_score: float) -> bool: + self.calls.append((record.id, decay_score)) + if self.raises is not None: + raise self.raises + return self.verdicts.get(record.id, self.default) + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -183,11 +215,16 @@ def fake_config(tmp_path) -> MnemaConfig: return cfg -def make_service(config: MnemaConfig, backend: VectorBackend, embedding: EmbeddingProvider): +def make_service( + config: MnemaConfig, + backend: VectorBackend, + embedding: EmbeddingProvider, + judge: MemoryJudge | None = None, +): """Build a MemoryService with injected fakes (bypassing make_backend).""" from mnema.service import MemoryService - return MemoryService(config, backend=backend, embedding=embedding) + return MemoryService(config, backend=backend, embedding=embedding, judge=judge) def make_record( diff --git a/packages/mnema-python/tests/test_config.py b/packages/mnema-python/tests/test_config.py index 9800688..5de4208 100644 --- a/packages/mnema-python/tests/test_config.py +++ b/packages/mnema-python/tests/test_config.py @@ -34,6 +34,22 @@ def test_validate_runtime_accepts_openai_with_key(self): cfg = MnemaConfig(embedding="openai", openai_api_key="sk-test") cfg.validate_runtime() # should not raise + def test_validate_runtime_rejects_smart_forget_without_model(self): + cfg = MnemaConfig(smart_forget_enabled=True, judge_model=None) + with pytest.raises(ConfigError, match="MNEMA_JUDGE_MODEL"): + cfg.validate_runtime() + + def test_validate_runtime_accepts_smart_forget_with_model(self): + cfg = MnemaConfig(smart_forget_enabled=True, judge_model="qwen3:8b") + cfg.validate_runtime() # should not raise + + def test_smart_forget_disabled_by_default(self): + cfg = MnemaConfig() + assert cfg.smart_forget_enabled is False + assert cfg.judge_model is None + assert cfg.judge_api_key is None + assert cfg.judge_base_url == "https://api.openai.com/v1" + def test_env_override(self, monkeypatch): monkeypatch.setenv("MNEMA_BACKEND", "qdrant") monkeypatch.setenv("MNEMA_DECAY_HALF_LIFE_DAYS", "7") diff --git a/packages/mnema-python/tests/test_judge.py b/packages/mnema-python/tests/test_judge.py new file mode 100644 index 0000000..8a4aabc --- /dev/null +++ b/packages/mnema-python/tests/test_judge.py @@ -0,0 +1,191 @@ +"""Tests for the smart-forgetting judge (opt-in LLM veto on decay). + +The OpenAI-compatible judge talks to an HTTP server we don't have in CI, so +its tests mock httpx exactly like tests/test_embeddings.py does for the +Ollama provider — no network anywhere. +""" + +from __future__ import annotations + +import pytest + +from mnema.config import MnemaConfig +from mnema.judge import ( + MemoryJudge, + OpenAICompatibleJudge, + build_forget_prompt, + make_judge, + parse_forget_verdict, +) +from tests.fakes import make_record + + +class TestParseForgetVerdict: + def test_forget_is_true(self): + assert parse_forget_verdict("FORGET") is True + assert parse_forget_verdict("FORGET outdated project note") is True + + def test_keep_is_false(self): + assert parse_forget_verdict("KEEP") is False + assert parse_forget_verdict("KEEP still referenced weekly") is False + + def test_case_insensitive_and_leading_whitespace(self): + assert parse_forget_verdict("forget — stale") is True + assert parse_forget_verdict(" keep this one") is False + assert parse_forget_verdict("\n\tForget") is True + + def test_unparseable_is_none(self): + assert parse_forget_verdict("") is None + assert parse_forget_verdict(" ") is None + assert parse_forget_verdict("maybe") is None + assert parse_forget_verdict("FORGETFUL memories fade") is None + assert parse_forget_verdict("KEEPER of facts") is None + + +class TestBuildForgetPrompt: + def test_contains_text_tags_and_score(self): + record = make_record( + text="Alice prefers Earl Grey tea", + tags=["preferences", "tea"], + importance=5, + ) + prompt = build_forget_prompt(record, 0.042) + assert "Alice prefers Earl Grey tea" in prompt + assert "preferences" in prompt + assert "tea" in prompt + assert "0.042" in prompt + + def test_mentions_verdict_format(self): + prompt = build_forget_prompt(make_record(), 0.5) + assert "KEEP" in prompt + assert "FORGET" in prompt + + +@pytest.fixture +def judge_config() -> MnemaConfig: + return MnemaConfig( + smart_forget_enabled=True, + judge_model="test-model", + judge_base_url="http://localhost:11434/v1", + ) + + +def _chat_response(content: str): + class _FakeResp: + def raise_for_status(self): + pass + + def json(self): + return {"choices": [{"message": {"role": "assistant", "content": content}}]} + + return _FakeResp() + + +class TestOpenAICompatibleJudge: + async def test_forget_response_is_true(self, judge_config, monkeypatch): + judge = OpenAICompatibleJudge(judge_config) + + captured = {} + + async def _fake_post(url, json): + captured["url"] = url + captured["json"] = json + return _chat_response("FORGET no longer relevant") + + monkeypatch.setattr(judge._client, "post", _fake_post) + + record = make_record(text="old standup note") + assert await judge.should_forget(record, 0.03) is True + # Verify the request shape. + assert captured["url"] == "/chat/completions" + assert captured["json"]["model"] == "test-model" + messages = captured["json"]["messages"] + assert len(messages) == 1 + assert messages[0]["role"] == "user" + assert "old standup note" in messages[0]["content"] + + async def test_keep_response_is_false(self, judge_config, monkeypatch): + judge = OpenAICompatibleJudge(judge_config) + + async def _fake_post(url, json): # noqa: ARG001 + return _chat_response("KEEP still relevant") + + monkeypatch.setattr(judge._client, "post", _fake_post) + assert await judge.should_forget(make_record(), 0.03) is False + + async def test_unparseable_response_fails_safe(self, judge_config, monkeypatch): + judge = OpenAICompatibleJudge(judge_config) + + async def _fake_post(url, json): # noqa: ARG001 + return _chat_response("I think maybe it depends") + + monkeypatch.setattr(judge._client, "post", _fake_post) + assert await judge.should_forget(make_record(), 0.03) is False + + async def test_http_error_fails_safe(self, judge_config, monkeypatch): + import httpx + + judge = OpenAICompatibleJudge(judge_config) + + async def _fake_post(url, json): # noqa: ARG001 + raise httpx.ConnectError("connection refused") + + monkeypatch.setattr(judge._client, "post", _fake_post) + # Must not raise — fail-safe is KEEP (False). + assert await judge.should_forget(make_record(), 0.03) is False + + async def test_malformed_payload_fails_safe(self, judge_config, monkeypatch): + judge = OpenAICompatibleJudge(judge_config) + + class _FakeResp: + def raise_for_status(self): + pass + + def json(self): + return {"unexpected": "shape"} + + async def _fake_post(url, json): # noqa: ARG001 + return _FakeResp() + + monkeypatch.setattr(judge._client, "post", _fake_post) + assert await judge.should_forget(make_record(), 0.03) is False + + def test_base_url_trailing_slash_stripped(self): + cfg = MnemaConfig( + smart_forget_enabled=True, + judge_model="m", + judge_base_url="http://localhost:11434/v1/", + ) + judge = OpenAICompatibleJudge(cfg) + assert judge._base_url == "http://localhost:11434/v1" + + def test_is_a_memory_judge(self, judge_config): + assert isinstance(OpenAICompatibleJudge(judge_config), MemoryJudge) + + def test_authorization_header_only_when_key_set(self, judge_config): + no_key = OpenAICompatibleJudge(judge_config) + assert "authorization" not in no_key._client.headers + + with_key = OpenAICompatibleJudge( + judge_config.model_copy(update={"judge_api_key": "sk-test"}) + ) + assert with_key._client.headers["authorization"] == "Bearer sk-test" + + async def test_aclose_releases_client(self, judge_config): + judge = OpenAICompatibleJudge(judge_config) + await judge.aclose() + assert judge._client.is_closed + + async def test_default_aclose_is_noop(self): + from tests.fakes import FakeJudge + + await FakeJudge().aclose() # must not raise + + +class TestMakeJudge: + def test_disabled_by_default(self): + assert make_judge(MnemaConfig()) is None + + def test_enabled_builds_judge(self, judge_config): + judge = make_judge(judge_config) + assert isinstance(judge, OpenAICompatibleJudge) diff --git a/packages/mnema-python/tests/test_service.py b/packages/mnema-python/tests/test_service.py index f9b375e..beec767 100644 --- a/packages/mnema-python/tests/test_service.py +++ b/packages/mnema-python/tests/test_service.py @@ -174,6 +174,118 @@ async def test_summarize_returns_plan(self, service): assert len(plan.clusters) >= 1 +class TestSmartForgetting: + """Opt-in LLM veto on the decay sweep (judge can only rescue, never condemn).""" + + @pytest.fixture + def smart_config(self, basic_config): + return basic_config.model_copy( + update={"smart_forget_enabled": True, "judge_model": "test-model"} + ) + + async def _seed_decayed(self, backend, embedding, text="ancient note"): + """Add a record old enough to fall at/below the default threshold.""" + from tests.fakes import make_record + + record = make_record(text=text, importance=1, age_days=365.0) + vec = (await embedding.embed([record.text]))[0] + await backend.add(record, vec) + return record + + async def test_disabled_never_constructs_judge( + self, basic_config, memory_backend, hashing_embedding + ): + """Default config: no judge is built, so no LLM call is possible.""" + from tests.fakes import make_service + + service = make_service(basic_config, memory_backend, hashing_embedding) + assert service._judge is None + rec = await self._seed_decayed(memory_backend, hashing_embedding) + candidates = await service.apply_decay(threshold=0.05, dry_run=False) + # Formula behavior unchanged: the decayed record is deleted. + assert [c.id for c in candidates] == [rec.id] + with pytest.raises(MemoryNotFoundError): + await service.get(rec.id) + + async def test_judge_forget_deletes( + self, smart_config, memory_backend, hashing_embedding + ): + from tests.fakes import FakeJudge, make_service + + judge = FakeJudge(default=True) + service = make_service( + smart_config, memory_backend, hashing_embedding, judge=judge + ) + rec = await self._seed_decayed(memory_backend, hashing_embedding) + candidates = await service.apply_decay(threshold=0.05, dry_run=False) + assert [c.id for c in candidates] == [rec.id] + assert [mid for mid, _ in judge.calls] == [rec.id] + with pytest.raises(MemoryNotFoundError): + await service.get(rec.id) + + async def test_judge_keep_rescues( + self, smart_config, memory_backend, hashing_embedding + ): + """Formula selected the record, but a KEEP verdict saves it.""" + from tests.fakes import FakeJudge, make_service + + judge = FakeJudge(default=False) + service = make_service( + smart_config, memory_backend, hashing_embedding, judge=judge + ) + rec = await self._seed_decayed(memory_backend, hashing_embedding) + candidates = await service.apply_decay(threshold=0.05, dry_run=False) + assert candidates == [] + assert [mid for mid, _ in judge.calls] == [rec.id] + # Survived the real (non-dry-run) sweep. + assert (await service.get(rec.id)).text == "ancient note" + + async def test_judge_error_fails_safe( + self, smart_config, memory_backend, hashing_embedding + ): + """A raising judge must never cause a deletion.""" + from tests.fakes import FakeJudge, make_service + + judge = FakeJudge(raises=RuntimeError("judge exploded")) + service = make_service( + smart_config, memory_backend, hashing_embedding, judge=judge + ) + rec = await self._seed_decayed(memory_backend, hashing_embedding) + candidates = await service.apply_decay(threshold=0.05, dry_run=False) + assert candidates == [] + assert (await service.get(rec.id)).text == "ancient note" + + async def test_judge_applies_in_dry_run( + self, smart_config, memory_backend, hashing_embedding + ): + """Dry-run reflects what smart mode would actually do.""" + from tests.fakes import FakeJudge, make_service + + judge = FakeJudge(default=False) + service = make_service( + smart_config, memory_backend, hashing_embedding, judge=judge + ) + rec = await self._seed_decayed(memory_backend, hashing_embedding) + candidates = await service.apply_decay(threshold=0.05, dry_run=True) + assert candidates == [] + assert [mid for mid, _ in judge.calls] == [rec.id] + + async def test_judge_never_sees_above_threshold( + self, smart_config, memory_backend, hashing_embedding + ): + """The judge can only rescue: fresh memories are never shown to it.""" + from tests.fakes import FakeJudge, make_service + + judge = FakeJudge(default=True) + service = make_service( + smart_config, memory_backend, hashing_embedding, judge=judge + ) + await service.remember("fresh memory", importance=5) + candidates = await service.apply_decay(threshold=0.05, dry_run=True) + assert candidates == [] + assert judge.calls == [] + + class TestSDK: async def test_sdk_memory_client_lifecycle(self, basic_config, memory_backend, hashing_embedding): from mnema.sdk import MemoryClient