diff --git a/src/powermem/intelligence/ebbinghaus_algorithm.py b/src/powermem/intelligence/ebbinghaus_algorithm.py index 441f1d7eb..c12aa1f28 100644 --- a/src/powermem/intelligence/ebbinghaus_algorithm.py +++ b/src/powermem/intelligence/ebbinghaus_algorithm.py @@ -255,16 +255,8 @@ def should_forget(self, memory: Dict[str, Any]) -> bool: True if memory should be forgotten """ try: - # Check decay factor - created_at = memory.get("created_at") - if created_at: - decay_factor = self.calculate_decay( - created_at, - decay_rate=self._resolve_decay_rate(memory), - ) - if decay_factor < self.working_threshold: - return True - + if self.calculate_current_retention(memory) < self.working_threshold: + return True return False except Exception as e: @@ -303,6 +295,66 @@ def should_archive(self, memory: Dict[str, Any]) -> bool: logger.error(f"Failed to check archiving: {e}") return False + def reinforce(self, memory: Dict[str, Any]) -> Dict[str, Any]: + """Boost current_retention on review and advance the review schedule. + + Called when a memory is accessed at or after its ``next_review`` time. + Uses diminishing-returns formula so retention approaches but never + exceeds 1.0. + + Returns: + Dict with updated intelligence fields to merge back. + """ + _, intelligence = self._resolve_metadata_sections(memory) + current_retention = self.calculate_current_retention(memory) + reinforcement_factor = self._resolve_reinforcement_factor(memory) + review_count = int(intelligence.get("review_count") or 0) + + new_retention = min( + 1.0, current_retention + reinforcement_factor * (1.0 - current_retention) + ) + new_review_count = review_count + 1 + + review_schedule = intelligence.get("review_schedule") or [] + next_review = ( + review_schedule[new_review_count] + if new_review_count < len(review_schedule) + else None + ) + + now = get_current_datetime() + return { + "current_retention": new_retention, + "review_count": new_review_count, + "last_reviewed": now.isoformat(), + "next_review": next_review, + } + + def calculate_current_retention(self, memory: Dict[str, Any]) -> float: + """Return the real-time effective retention for display/ranking. + + ``current_retention`` is a snapshot captured at ``last_reviewed`` (or + creation time for initial metadata), so it must decay before runtime + consumers use it. This avoids treating initialized retention as a + permanent floor while still reflecting recent review reinforcement. + """ + stored = self._resolve_current_retention(memory) + if stored is not None: + anchor = self._resolve_retention_anchor(memory) + decay = self.calculate_decay( + anchor, + decay_rate=self._resolve_decay_rate(memory), + ) + return max(0.0, min(1.0, stored * decay)) + + initial = self._resolve_initial_retention(memory) + created_at = self._resolve_created_at(memory) + decay = self.calculate_decay( + created_at, + decay_rate=self._resolve_decay_rate(memory), + ) + return max(0.0, min(1.0, initial * decay)) + def get_review_schedule( self, memory: Dict[str, Any], *, prefer_stored: bool = True ) -> list: @@ -455,7 +507,63 @@ def _resolve_reinforcement_factor(self, memory: Dict[str, Any]) -> float: logger.warning("Invalid reinforcement_factor: %s", raw) return 0.0 return max(factor, 0.0) - + + def _resolve_initial_retention(self, memory: Dict[str, Any]) -> float: + """Resolve initial_retention from memory metadata with config fallback.""" + meta, intelligence = self._resolve_metadata_sections(memory) + raw = self._first_present( + memory.get("initial_retention"), + meta.get("initial_retention"), + intelligence.get("initial_retention"), + ) + if raw is not None: + try: + val = float(raw) + except (TypeError, ValueError): + logger.warning("Invalid initial_retention: %s", raw) + else: + if 0.0 < val <= 1.0: + return val + return self.initial_retention + + def _resolve_current_retention( + self, memory: Dict[str, Any] + ) -> Optional[float]: + """Resolve stored current_retention as a bounded snapshot value.""" + meta, intelligence = self._resolve_metadata_sections(memory) + raw = self._first_present( + memory.get("current_retention"), + meta.get("current_retention"), + intelligence.get("current_retention"), + ) + if raw is None: + return None + try: + retention = float(raw) + except (TypeError, ValueError): + logger.warning("Invalid current_retention: %s", raw) + return None + return max(0.0, min(1.0, retention)) + + def _resolve_created_at(self, memory: Dict[str, Any]) -> Any: + """Resolve creation timestamp from supported memory layouts.""" + meta, intelligence = self._resolve_metadata_sections(memory) + return self._first_present( + memory.get("created_at"), + meta.get("created_at"), + intelligence.get("created_at"), + ) + + def _resolve_retention_anchor(self, memory: Dict[str, Any]) -> Any: + """Resolve the timestamp for decaying current_retention snapshots.""" + meta, intelligence = self._resolve_metadata_sections(memory) + return self._first_present( + intelligence.get("last_reviewed"), + memory.get("last_reviewed"), + meta.get("last_reviewed"), + self._resolve_created_at(memory), + ) + def _parse_datetime(self, value: Any) -> datetime: """Parse datetime from object or ISO string.""" if isinstance(value, datetime): diff --git a/src/powermem/intelligence/intelligent_memory_manager.py b/src/powermem/intelligence/intelligent_memory_manager.py index 68f98d992..efb2b5ccd 100644 --- a/src/powermem/intelligence/intelligent_memory_manager.py +++ b/src/powermem/intelligence/intelligent_memory_manager.py @@ -173,14 +173,10 @@ def process_search_results( Processed and ranked results """ try: - # Apply Ebbinghaus decay to results processed_results = [] for result in results: - # Apply decay based on age and memory type - decay_rate = self.ebbinghaus_algorithm._resolve_decay_rate(result) - decay_factor = self.ebbinghaus_algorithm.calculate_decay( - result.get("created_at", get_current_datetime()), - decay_rate=decay_rate, + effective_retention = ( + self.ebbinghaus_algorithm.calculate_current_retention(result) ) processed_result = result.copy() @@ -193,12 +189,12 @@ def process_search_results( ) if original_score is not None: processed_result["original_score"] = original_score - processed_result["decay_factor"] = decay_factor + processed_result["effective_retention"] = effective_retention processed_result["forgotten_score_multiplier"] = ( forgotten_score_multiplier ) processed_result["final_score"] = ( - base_score * decay_factor * forgotten_score_multiplier + base_score * effective_retention * forgotten_score_multiplier ) processed_result["score"] = processed_result["final_score"] diff --git a/src/powermem/intelligence/plugin.py b/src/powermem/intelligence/plugin.py index 5e512d1ff..d8dd497e5 100644 --- a/src/powermem/intelligence/plugin.py +++ b/src/powermem/intelligence/plugin.py @@ -122,9 +122,8 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool if not self.enabled or not self._algo: return None, False try: - # Normalize: intelligence fields may be stored inside the metadata - # JSON column and not exposed at the top level of the memory dict. meta = memory.get("metadata") or {} + intelligence = meta.get("intelligence") or memory.get("intelligence") or {} memory_type = memory.get("memory_type") or meta.get("memory_type") access_count_old = memory.get("access_count") if access_count_old is None: @@ -136,14 +135,14 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool importance_score = 0.5 new_access_count = access_count_old + 1 + now = get_current_datetime() updates: Dict[str, Any] = { "access_count": new_access_count, - "updated_at": get_current_datetime(), + "updated_at": now, } - # Track which fields need updating inside the metadata JSON column meta_updates: Dict[str, Any] = {"access_count": new_access_count} + intel_updates: Dict[str, Any] = {} - # Provide normalized values to algorithm checks normalized = { **memory, "memory_type": memory_type, @@ -151,6 +150,24 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool "importance_score": importance_score, } + # Review reinforcement: if the access happens at or after + # next_review, boost current_retention and advance the schedule. + next_review_raw = intelligence.get("next_review") + if next_review_raw: + next_review_dt = self._algo._parse_datetime(next_review_raw) + if now >= next_review_dt: + reinforcement_result = self._algo.reinforce(normalized) + intel_updates.update(reinforcement_result) + + # Apply reinforcement to normalized so downstream should_forget() + # uses the boosted retention and its new timestamp anchor. + if intel_updates: + norm_intel = dict(normalized.get("metadata", {}).get("intelligence") or {}) + norm_intel.update(intel_updates) + norm_meta = dict(normalized.get("metadata") or {}) + norm_meta["intelligence"] = norm_intel + normalized = {**normalized, "metadata": norm_meta} + # Check promotion first — an accessed memory that qualifies for # promotion should not be forgotten in the same on_get call. new_memory_type = memory_type @@ -164,8 +181,7 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool updates["memory_type"] = "long_term" meta_updates["memory_type"] = "long_term" - # Clear forget marker on promotion — a promoted memory should - # no longer carry the 0.1x search penalty from a prior soft-forget. + # Clear forget marker on promotion if new_memory_type != memory_type: meta_updates["should_forget"] = False meta_updates["marked_for_forgetting_at"] = None @@ -176,7 +192,6 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool if new_memory_type == memory_type and self._algo.should_forget(normalized): return None, True - # Check if memory should be archived if self._algo.should_archive(normalized): meta_updates["archived"] = True @@ -194,8 +209,26 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool original_content, importance_score, new_memory_type or "working" ) if "intelligence" in intelligence_metadata: - updates["metadata"]["intelligence"] = intelligence_metadata["intelligence"] - updates["last_reprocessed_at"] = get_current_datetime() + reprocessed_intel = intelligence_metadata["intelligence"] + # Preserve current_retention and review progress from + # reinforcement; reprocessing should only refresh + # decay_rate and review_schedule, not reset retention. + for keep_key in ( + "current_retention", + "review_count", + "last_reviewed", + "next_review", + ): + if keep_key in intel_updates: + reprocessed_intel[keep_key] = intel_updates[keep_key] + elif keep_key in intelligence: + reprocessed_intel[keep_key] = intelligence[keep_key] + updates["metadata"]["intelligence"] = reprocessed_intel + updates["last_reprocessed_at"] = now + elif intel_updates: + existing_intel = dict(updates["metadata"].get("intelligence") or intelligence) + existing_intel.update(intel_updates) + updates["metadata"]["intelligence"] = existing_intel return updates, False except Exception as e: diff --git a/tests/unit/intelligence/test_ebbinghaus_decay_rate.py b/tests/unit/intelligence/test_ebbinghaus_decay_rate.py index 54ba06714..395546fb7 100644 --- a/tests/unit/intelligence/test_ebbinghaus_decay_rate.py +++ b/tests/unit/intelligence/test_ebbinghaus_decay_rate.py @@ -147,8 +147,8 @@ def test_reinforced_memory_decays_slower_in_search_results(): by_id = {item["id"]: item for item in processed} assert ( - by_id["reinforced"]["decay_factor"] - > by_id["unreinforced"]["decay_factor"] + by_id["reinforced"]["effective_retention"] + > by_id["unreinforced"]["effective_retention"] ) assert processed[0]["id"] == "reinforced" @@ -216,7 +216,7 @@ def test_search_results_use_type_specific_decay_rate(): processed = manager.process_search_results(results, "keyword") by_id = {item["id"]: item for item in processed} - assert by_id["working"]["decay_factor"] < by_id["long"]["decay_factor"] + assert by_id["working"]["effective_retention"] < by_id["long"]["effective_retention"] assert processed[0]["id"] == "long" @@ -338,7 +338,7 @@ def test_search_results_do_not_demote_unmarked_memories(): assert processed[0]["forgotten_score_multiplier"] == pytest.approx(1.0) assert processed[0]["final_score"] == pytest.approx( - 0.85 * processed[0]["decay_factor"] + 0.85 * processed[0]["effective_retention"] ) @@ -355,7 +355,7 @@ def test_search_results_use_storage_score_for_ranking(): assert processed[0]["original_score"] == pytest.approx(0.92) assert processed[0]["final_score"] == pytest.approx( - 0.92 * processed[0]["decay_factor"] + 0.92 * processed[0]["effective_retention"] ) diff --git a/tests/unit/intelligence/test_retention_runtime.py b/tests/unit/intelligence/test_retention_runtime.py new file mode 100644 index 000000000..66bc0c036 --- /dev/null +++ b/tests/unit/intelligence/test_retention_runtime.py @@ -0,0 +1,598 @@ +"""Tests for retention fields runtime integration. + +Covers: +- should_forget() using initial_retention (effective_retention) +- reinforce() boosting current_retention +- on_get() triggering reinforcement when review is due +- on_get() NOT triggering reinforcement before review time +- Reprocessing preserving current_retention +- calculate_current_retention() combining initial_retention and decay +""" + +import math +from datetime import timedelta +from unittest.mock import patch + +import pytest + +from powermem.intelligence.ebbinghaus_algorithm import EbbinghausAlgorithm +from powermem.intelligence.plugin import EbbinghausIntelligencePlugin +from powermem.intelligence.intelligent_memory_manager import IntelligentMemoryManager +from powermem.utils.utils import get_current_datetime + + +@pytest.fixture +def algo(): + return EbbinghausAlgorithm({"decay_rate": 1.5, "initial_retention": 1.0}) + + +@pytest.fixture +def algo_low_retention(): + return EbbinghausAlgorithm({"decay_rate": 1.5, "initial_retention": 0.5}) + + +# ---- Test 1: should_forget considers initial_retention ---- + +def test_should_forget_considers_initial_retention(algo): + """High-importance memory should survive longer than low-importance at same age.""" + created_at = get_current_datetime() - timedelta(hours=30) + + low_importance = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + } + }, + } + high_importance = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.95, + } + }, + } + + # With effective_retention = initial_retention * decay_factor, + # the low-importance memory should be forgotten sooner. + assert algo.should_forget(low_importance) is True + assert algo.should_forget(high_importance) is False + + +# ---- Test 2: reinforce boosts current_retention ---- + +def test_reinforce_boosts_current_retention(algo): + """reinforce() should increase current_retention with diminishing returns.""" + now = get_current_datetime() + schedule = [ + (now - timedelta(hours=1)).isoformat(), + (now + timedelta(hours=5)).isoformat(), + (now + timedelta(hours=23)).isoformat(), + ] + memory = { + "metadata": { + "intelligence": { + "current_retention": 0.6, + "initial_retention": 0.6, + "reinforcement_factor": 0.3, + "review_count": 0, + "review_schedule": schedule, + } + } + } + + result = algo.reinforce(memory) + + expected = min(1.0, 0.6 + 0.3 * (1.0 - 0.6)) + assert result["current_retention"] == pytest.approx(expected) + assert result["review_count"] == 1 + assert result["last_reviewed"] is not None + assert result["next_review"] == schedule[1] + + +def test_reinforce_never_exceeds_one(algo): + """current_retention should never exceed 1.0 after reinforcement.""" + memory = { + "metadata": { + "intelligence": { + "current_retention": 0.95, + "reinforcement_factor": 0.5, + "review_count": 0, + "review_schedule": [], + } + } + } + + result = algo.reinforce(memory) + assert result["current_retention"] <= 1.0 + + +# ---- Test 3: on_get triggers reinforcement when review is due ---- + +def test_on_get_triggers_reinforcement_when_review_due(): + """When now >= next_review, on_get should boost current_retention.""" + config = { + "enabled": True, + "importance": {}, + "llm": {}, + "decay_rate": 1.5, + "initial_retention": 1.0, + "reinforcement_factor": 0.3, + } + plugin = EbbinghausIntelligencePlugin(config) + + now = get_current_datetime() + past_review = (now - timedelta(hours=1)).isoformat() + future_review = (now + timedelta(hours=23)).isoformat() + + memory = { + "id": "test-mem", + "content": "test content", + "memory_type": "working", + "access_count": 0, + "importance_score": 0.5, + "created_at": (now - timedelta(hours=2)).isoformat(), + "metadata": { + "memory_type": "working", + "intelligence": { + "current_retention": 0.7, + "initial_retention": 0.7, + "reinforcement_factor": 0.3, + "review_count": 0, + "next_review": past_review, + "review_schedule": [past_review, future_review], + } + }, + } + + updates, delete = plugin.on_get(memory) + + assert delete is False + assert updates is not None + intel = updates["metadata"]["intelligence"] + assert intel["current_retention"] > 0.7 + assert intel["review_count"] == 1 + assert intel["next_review"] == future_review + + +# ---- Test 4: on_get does NOT reinforce before review time ---- + +def test_on_get_does_not_reinforce_before_review_due(): + """When now < next_review, current_retention should not change via reinforce.""" + config = { + "enabled": True, + "importance": {}, + "llm": {}, + "decay_rate": 1.5, + "initial_retention": 1.0, + "reinforcement_factor": 0.3, + } + plugin = EbbinghausIntelligencePlugin(config) + + now = get_current_datetime() + future_review = (now + timedelta(hours=5)).isoformat() + + memory = { + "id": "test-mem", + "content": "test content", + "memory_type": "working", + "access_count": 0, + "importance_score": 0.5, + "created_at": now.isoformat(), + "metadata": { + "memory_type": "working", + "intelligence": { + "current_retention": 0.7, + "initial_retention": 0.7, + "reinforcement_factor": 0.3, + "review_count": 0, + "next_review": future_review, + "review_schedule": [future_review], + } + }, + } + + updates, delete = plugin.on_get(memory) + + assert delete is False + assert updates is not None + intel = updates["metadata"].get("intelligence", {}) + if "current_retention" in intel: + assert intel["current_retention"] == pytest.approx(0.7) + + +# ---- Test 5: reprocessing preserves current_retention ---- + +def test_reprocess_preserves_current_retention(): + """When access_count%5 triggers reprocessing, current_retention from + reinforcement should not be reset to initial_retention.""" + config = { + "enabled": True, + "importance": {}, + "llm": {}, + "decay_rate": 1.5, + "initial_retention": 1.0, + "reinforcement_factor": 0.3, + } + plugin = EbbinghausIntelligencePlugin(config) + + now = get_current_datetime() + past_review = (now - timedelta(hours=1)).isoformat() + future_review = (now + timedelta(hours=23)).isoformat() + + memory = { + "id": "test-mem", + "content": "test content", + "memory_type": "working", + "access_count": 4, + "importance_score": 0.5, + "created_at": (now - timedelta(hours=2)).isoformat(), + "metadata": { + "memory_type": "working", + "importance_score": 0.5, + "intelligence": { + "current_retention": 0.85, + "initial_retention": 0.5, + "reinforcement_factor": 0.3, + "review_count": 2, + "last_reviewed": (now - timedelta(hours=1)).isoformat(), + "next_review": past_review, + "review_schedule": [past_review, future_review], + } + }, + } + + updates, delete = plugin.on_get(memory) + + assert delete is False + assert updates is not None + intel = updates["metadata"]["intelligence"] + # current_retention should not have been reset to initial_retention (0.5); + # it should be >= the pre-existing 0.85 (reinforcement may boost it further). + assert intel["current_retention"] >= 0.85 + assert intel["review_count"] >= 2 + + +# ---- Test 6: calculate_current_retention combines initial and decay ---- + +def test_calculate_current_retention_combines_initial_and_decay(algo): + """Without stored current_retention, should return initial_retention * decay_factor.""" + created_at = get_current_datetime() - timedelta(hours=24) + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.8, + } + }, + } + + result = algo.calculate_current_retention(memory) + raw_decay = algo.calculate_decay( + created_at, decay_rate=algo._resolve_decay_rate(memory) + ) + + assert result == pytest.approx(0.8 * raw_decay) + + +def test_calculate_current_retention_defaults_without_stored_initial(algo): + """When no initial_retention is stored, use the config default.""" + created_at = get_current_datetime() - timedelta(hours=12) + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + } + + result = algo.calculate_current_retention(memory) + raw_decay = algo.calculate_decay( + created_at, decay_rate=algo._resolve_decay_rate(memory) + ) + + assert result == pytest.approx(algo.initial_retention * raw_decay) + + +def test_initialized_current_retention_decays_and_can_forget(algo): + """Real process_memory_metadata output should not create a permanent floor. + + process_memory_metadata initializes current_retention to initial_retention. + That stored snapshot must still decay over time, otherwise normal working + memories can never fall below the forget threshold. + """ + created_at = get_current_datetime() - timedelta(hours=50) + metadata = algo.process_memory_metadata( + "ordinary working memory", + importance_score=0.5, + memory_type="working", + ) + intelligence = metadata["intelligence"] + intelligence["last_reviewed"] = created_at.isoformat() + metadata["created_at"] = created_at.isoformat() + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "memory_type": "working", + "intelligence": intelligence, + }, + } + + assert intelligence["current_retention"] == pytest.approx( + intelligence["initial_retention"] + ) + assert algo.calculate_current_retention(memory) < algo.working_threshold + assert algo.should_forget(memory) is True + + +def test_search_ranking_decays_initialized_current_retention(): + """Search ranking should not treat initialized current_retention as fixed.""" + manager = IntelligentMemoryManager( + {"intelligent_memory": {"decay_rate": 1.5, "initial_retention": 1.0}} + ) + algo = manager.ebbinghaus_algorithm + old_time = get_current_datetime() - timedelta(hours=50) + fresh_time = get_current_datetime() + + old_meta = algo.process_memory_metadata("keyword", 0.5, "working") + old_meta["intelligence"]["last_reviewed"] = old_time.isoformat() + fresh_meta = algo.process_memory_metadata("keyword", 0.5, "working") + fresh_meta["intelligence"]["last_reviewed"] = fresh_time.isoformat() + + results = [ + { + "id": "old", + "content": "keyword", + "score": 0.8, + "created_at": old_time, + "memory_type": "working", + "access_count": 0, + "metadata": { + "memory_type": "working", + "intelligence": old_meta["intelligence"], + }, + }, + { + "id": "fresh", + "content": "keyword", + "score": 0.8, + "created_at": fresh_time, + "memory_type": "working", + "access_count": 0, + "metadata": { + "memory_type": "working", + "intelligence": fresh_meta["intelligence"], + }, + }, + ] + + processed = manager.process_search_results(results, "keyword") + by_id = {item["id"]: item for item in processed} + + assert by_id["old"]["effective_retention"] < by_id["fresh"]["effective_retention"] + assert processed[0]["id"] == "fresh" + + +def test_reinforce_uses_decayed_current_retention_before_boost(algo): + """reinforce() should boost the real-time retention, not a stale snapshot.""" + last_reviewed = get_current_datetime() - timedelta(hours=50) + memory = { + "created_at": last_reviewed, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.8, + "last_reviewed": last_reviewed.isoformat(), + "reinforcement_factor": 0.3, + "review_count": 0, + "review_schedule": [], + } + }, + } + decayed = 0.8 * algo.calculate_decay( + last_reviewed, + decay_rate=algo._resolve_decay_rate(memory), + ) + + result = algo.reinforce(memory) + + assert result["current_retention"] == pytest.approx( + decayed + 0.3 * (1.0 - decayed) + ) + assert result["current_retention"] < 0.8 + + +def test_search_ranking_uses_effective_retention(): + """process_search_results should rank by effective_retention, not raw decay.""" + manager = IntelligentMemoryManager( + {"intelligent_memory": {"decay_rate": 1.5, "initial_retention": 1.0}} + ) + created_at = get_current_datetime() - timedelta(hours=30) + + results = [ + { + "id": "low-init", + "content": "keyword", + "score": 0.8, + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": {"initial_retention": 0.3} + }, + }, + { + "id": "high-init", + "content": "keyword", + "score": 0.8, + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": {"initial_retention": 0.95} + }, + }, + ] + + processed = manager.process_search_results(results, "keyword") + by_id = {item["id"]: item for item in processed} + + assert by_id["high-init"]["final_score"] > by_id["low-init"]["final_score"] + assert "effective_retention" in by_id["high-init"] + + +# ---- Tests for review-reinforcement protecting against forgetting ---- + +def test_should_forget_respects_stored_current_retention(algo): + """Recent reinforced retention can protect a memory from forgetting.""" + created_at = get_current_datetime() - timedelta(hours=50) + last_reviewed = get_current_datetime() + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.8, + "last_reviewed": last_reviewed.isoformat(), + } + }, + } + + assert algo.should_forget(memory) is False + + +def test_on_get_reinforced_memory_not_forgotten_same_call(): + """A memory that is reinforced during on_get should not be deleted in the + same call, even if its pre-reinforcement effective retention was below the + forget threshold.""" + config = { + "enabled": True, + "importance": {}, + "llm": {}, + "decay_rate": 1.5, + "initial_retention": 1.0, + "reinforcement_factor": 0.3, + "working_threshold": 0.3, + } + plugin = EbbinghausIntelligencePlugin(config) + + now = get_current_datetime() + past_review = (now - timedelta(hours=1)).isoformat() + future_review = (now + timedelta(hours=23)).isoformat() + created_at = (now - timedelta(hours=50)).isoformat() + + memory = { + "id": "reinforce-protect", + "content": "important content", + "memory_type": "working", + "access_count": 0, + "importance_score": 0.3, + "created_at": created_at, + "metadata": { + "memory_type": "working", + "intelligence": { + "current_retention": 0.25, + "initial_retention": 0.3, + "reinforcement_factor": 0.3, + "review_count": 0, + "next_review": past_review, + "review_schedule": [past_review, future_review], + } + }, + } + + updates, delete = plugin.on_get(memory) + + assert delete is False + assert updates is not None + intel = updates["metadata"]["intelligence"] + assert intel["current_retention"] > 0.25 + + +def test_calculate_current_retention_reflects_reinforcement(algo): + """calculate_current_retention should decay a reinforced retention snapshot.""" + created_at = get_current_datetime() - timedelta(hours=50) + last_reviewed = get_current_datetime() + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.85, + "last_reviewed": last_reviewed.isoformat(), + } + }, + } + + base_decay = algo.calculate_decay( + created_at, decay_rate=algo._resolve_decay_rate(memory) + ) + base_retention = 0.3 * base_decay + + result = algo.calculate_current_retention(memory) + + assert result > base_retention + assert result <= 0.85 + + +def test_search_ranking_reflects_reinforced_current_retention(): + """process_search_results should rank a reinforced memory higher than + an unreinforced one with the same initial conditions.""" + manager = IntelligentMemoryManager( + {"intelligent_memory": {"decay_rate": 1.5, "initial_retention": 1.0}} + ) + created_at = get_current_datetime() - timedelta(hours=50) + last_reviewed = get_current_datetime() + + results = [ + { + "id": "unreinforced", + "content": "keyword", + "score": 0.8, + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": {"initial_retention": 0.3} + }, + }, + { + "id": "reinforced", + "content": "keyword", + "score": 0.8, + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.85, + "last_reviewed": last_reviewed.isoformat(), + } + }, + }, + ] + + processed = manager.process_search_results(results, "keyword") + by_id = {item["id"]: item for item in processed} + + assert by_id["reinforced"]["effective_retention"] > by_id["unreinforced"]["effective_retention"] + assert processed[0]["id"] == "reinforced"