From 874b18c52f264856ae0ea3e4ee13230111c34485 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Tue, 23 Jun 2026 23:02:49 +0800 Subject: [PATCH 1/3] fix(intelligence): integrate retention fields into runtime paths (#1082) reinforcement_factor, initial_retention, and current_retention were written during process_memory_metadata() but never consumed in the core runtime paths (should_forget, on_get, search ranking). This commit: - should_forget() now computes effective_retention = initial_retention * decay_factor so higher-importance memories survive longer - New reinforce() method boosts current_retention with diminishing returns when a memory is accessed at or after its next_review time - on_get() triggers reinforcement on review-due access and preserves current_retention during periodic reprocessing (access_count % 5) - New calculate_current_retention() provides unified real-time retention for search ranking and display - process_search_results() uses effective_retention instead of raw decay Fixes #1082 --- .../intelligence/ebbinghaus_algorithm.py | 79 +++- .../intelligent_memory_manager.py | 12 +- src/powermem/intelligence/plugin.py | 44 ++- .../test_ebbinghaus_decay_rate.py | 10 +- .../intelligence/test_retention_runtime.py | 340 ++++++++++++++++++ 5 files changed, 459 insertions(+), 26 deletions(-) create mode 100644 tests/unit/intelligence/test_retention_runtime.py diff --git a/src/powermem/intelligence/ebbinghaus_algorithm.py b/src/powermem/intelligence/ebbinghaus_algorithm.py index 441f1d7eb..a175a2398 100644 --- a/src/powermem/intelligence/ebbinghaus_algorithm.py +++ b/src/powermem/intelligence/ebbinghaus_algorithm.py @@ -255,14 +255,15 @@ 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: + initial_retention = self._resolve_initial_retention(memory) + effective_retention = initial_retention * decay_factor + if effective_retention < self.working_threshold: return True return False @@ -303,6 +304,60 @@ 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 = intelligence.get("current_retention") + if current_retention is None: + current_retention = intelligence.get("initial_retention", 1.0) + try: + current_retention = float(current_retention) + except (TypeError, ValueError): + current_retention = 1.0 + + 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. + + ``effective_retention = initial_retention * decay_factor`` + """ + initial = self._resolve_initial_retention(memory) + created_at = memory.get("created_at") + decay = self.calculate_decay( + created_at, decay_rate=self._resolve_decay_rate(memory) + ) + return initial * decay + def get_review_schedule( self, memory: Dict[str, Any], *, prefer_stored: bool = True ) -> list: @@ -455,7 +510,25 @@ 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 _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..1aa87290a 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,15 @@ 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) + # 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 +172,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 +183,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 +200,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..29f067cc0 --- /dev/null +++ b/tests/unit/intelligence/test_retention_runtime.py @@ -0,0 +1,340 @@ +"""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): + """calculate_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_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"] From dcaf714a2956473d3ff5148a2542235401111f94 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Thu, 25 Jun 2026 23:41:53 +0800 Subject: [PATCH 2/3] fix(intelligence): make current_retention a runtime-effective field Address review feedback from #1084: current_retention was written by reinforce() but never consumed by should_forget() or search ranking. - should_forget() now uses max(initial_retention * decay_factor, current_retention) so reinforced memories are protected from premature forgetting - on_get() applies reinforcement result to normalized memory before the forget check, preventing the same-call reinforce-then-forget race - calculate_current_retention() returns max(base, stored current_retention) so search ranking reflects reinforcement boosts - 4 new tests covering reinforcement-protects-from-forgetting and search-ranking-reflects-reinforcement scenarios --- .../intelligence/ebbinghaus_algorithm.py | 32 +++- src/powermem/intelligence/plugin.py | 9 ++ .../intelligence/test_retention_runtime.py | 144 +++++++++++++++++- 3 files changed, 181 insertions(+), 4 deletions(-) diff --git a/src/powermem/intelligence/ebbinghaus_algorithm.py b/src/powermem/intelligence/ebbinghaus_algorithm.py index a175a2398..2f39b15b0 100644 --- a/src/powermem/intelligence/ebbinghaus_algorithm.py +++ b/src/powermem/intelligence/ebbinghaus_algorithm.py @@ -262,7 +262,21 @@ def should_forget(self, memory: Dict[str, Any]) -> bool: decay_rate=self._resolve_decay_rate(memory), ) initial_retention = self._resolve_initial_retention(memory) - effective_retention = initial_retention * decay_factor + base_retention = initial_retention * decay_factor + + _, intelligence = self._resolve_metadata_sections(memory) + stored_retention = intelligence.get("current_retention") + if stored_retention is not None: + try: + stored_retention = float(stored_retention) + except (TypeError, ValueError): + stored_retention = None + + effective_retention = ( + max(base_retention, stored_retention) + if stored_retention is not None + else base_retention + ) if effective_retention < self.working_threshold: return True @@ -349,14 +363,26 @@ def reinforce(self, memory: Dict[str, Any]) -> Dict[str, Any]: def calculate_current_retention(self, memory: Dict[str, Any]) -> float: """Return the real-time effective retention for display/ranking. - ``effective_retention = initial_retention * decay_factor`` + Uses ``max(initial_retention * decay_factor, current_retention)`` so + that review reinforcement is reflected in search ranking and display. """ initial = self._resolve_initial_retention(memory) created_at = memory.get("created_at") decay = self.calculate_decay( created_at, decay_rate=self._resolve_decay_rate(memory) ) - return initial * decay + base = initial * decay + + _, intelligence = self._resolve_metadata_sections(memory) + stored = intelligence.get("current_retention") + if stored is not None: + try: + stored = float(stored) + except (TypeError, ValueError): + stored = None + if stored is not None: + return max(base, stored) + return base def get_review_schedule( self, memory: Dict[str, Any], *, prefer_stored: bool = True diff --git a/src/powermem/intelligence/plugin.py b/src/powermem/intelligence/plugin.py index 1aa87290a..ec9fbcc27 100644 --- a/src/powermem/intelligence/plugin.py +++ b/src/powermem/intelligence/plugin.py @@ -159,6 +159,15 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool reinforcement_result = self._algo.reinforce(normalized) intel_updates.update(reinforcement_result) + # Apply reinforcement to normalized so downstream should_forget() + # and should_promote() see the boosted current_retention. + if intel_updates.get("current_retention") is not None: + norm_intel = dict(normalized.get("metadata", {}).get("intelligence") or {}) + norm_intel["current_retention"] = intel_updates["current_retention"] + 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 diff --git a/tests/unit/intelligence/test_retention_runtime.py b/tests/unit/intelligence/test_retention_runtime.py index 29f067cc0..0be2f1947 100644 --- a/tests/unit/intelligence/test_retention_runtime.py +++ b/tests/unit/intelligence/test_retention_runtime.py @@ -261,7 +261,7 @@ def test_reprocess_preserves_current_retention(): # ---- Test 6: calculate_current_retention combines initial and decay ---- def test_calculate_current_retention_combines_initial_and_decay(algo): - """calculate_current_retention should return initial_retention * decay_factor.""" + """Without stored current_retention, should return initial_retention * decay_factor.""" created_at = get_current_datetime() - timedelta(hours=24) memory = { @@ -338,3 +338,145 @@ def test_search_ranking_uses_effective_retention(): 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): + """A memory whose base retention (initial * decay) falls below threshold + should NOT be forgotten if current_retention (from reinforcement) is still + above the threshold.""" + created_at = get_current_datetime() - timedelta(hours=50) + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.8, + } + }, + } + + 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 return max(base, stored + current_retention) so search ranking reflects reinforcement.""" + created_at = get_current_datetime() - timedelta(hours=50) + + memory = { + "created_at": created_at, + "memory_type": "working", + "access_count": 0, + "metadata": { + "intelligence": { + "initial_retention": 0.3, + "current_retention": 0.85, + } + }, + } + + 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 == pytest.approx(max(base_retention, 0.85)) + 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) + + 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, + } + }, + }, + ] + + 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" From b74097311d99b5aff35c12a5c9a80d57cd9b6fce Mon Sep 17 00:00:00 2001 From: knqiufan Date: Sat, 27 Jun 2026 23:52:17 +0800 Subject: [PATCH 3/3] fix(intelligence): decay current retention snapshots Treat stored current_retention as a timestamped retention snapshot instead of a permanent floor, so normal memories can still decay while recent review boosts remain effective. --- .../intelligence/ebbinghaus_algorithm.py | 107 +++++++------- src/powermem/intelligence/plugin.py | 6 +- .../intelligence/test_retention_runtime.py | 130 +++++++++++++++++- 3 files changed, 184 insertions(+), 59 deletions(-) diff --git a/src/powermem/intelligence/ebbinghaus_algorithm.py b/src/powermem/intelligence/ebbinghaus_algorithm.py index 2f39b15b0..c12aa1f28 100644 --- a/src/powermem/intelligence/ebbinghaus_algorithm.py +++ b/src/powermem/intelligence/ebbinghaus_algorithm.py @@ -255,31 +255,8 @@ def should_forget(self, memory: Dict[str, Any]) -> bool: True if memory should be forgotten """ try: - created_at = memory.get("created_at") - if created_at: - decay_factor = self.calculate_decay( - created_at, - decay_rate=self._resolve_decay_rate(memory), - ) - initial_retention = self._resolve_initial_retention(memory) - base_retention = initial_retention * decay_factor - - _, intelligence = self._resolve_metadata_sections(memory) - stored_retention = intelligence.get("current_retention") - if stored_retention is not None: - try: - stored_retention = float(stored_retention) - except (TypeError, ValueError): - stored_retention = None - - effective_retention = ( - max(base_retention, stored_retention) - if stored_retention is not None - else base_retention - ) - if effective_retention < self.working_threshold: - return True - + if self.calculate_current_retention(memory) < self.working_threshold: + return True return False except Exception as e: @@ -329,14 +306,7 @@ def reinforce(self, memory: Dict[str, Any]) -> Dict[str, Any]: Dict with updated intelligence fields to merge back. """ _, intelligence = self._resolve_metadata_sections(memory) - current_retention = intelligence.get("current_retention") - if current_retention is None: - current_retention = intelligence.get("initial_retention", 1.0) - try: - current_retention = float(current_retention) - except (TypeError, ValueError): - current_retention = 1.0 - + current_retention = self.calculate_current_retention(memory) reinforcement_factor = self._resolve_reinforcement_factor(memory) review_count = int(intelligence.get("review_count") or 0) @@ -363,26 +333,27 @@ def reinforce(self, memory: Dict[str, Any]) -> Dict[str, Any]: def calculate_current_retention(self, memory: Dict[str, Any]) -> float: """Return the real-time effective retention for display/ranking. - Uses ``max(initial_retention * decay_factor, current_retention)`` so - that review reinforcement is reflected in search ranking and display. + ``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 = memory.get("created_at") + created_at = self._resolve_created_at(memory) decay = self.calculate_decay( - created_at, decay_rate=self._resolve_decay_rate(memory) + created_at, + decay_rate=self._resolve_decay_rate(memory), ) - base = initial * decay - - _, intelligence = self._resolve_metadata_sections(memory) - stored = intelligence.get("current_retention") - if stored is not None: - try: - stored = float(stored) - except (TypeError, ValueError): - stored = None - if stored is not None: - return max(base, stored) - return base + return max(0.0, min(1.0, initial * decay)) def get_review_schedule( self, memory: Dict[str, Any], *, prefer_stored: bool = True @@ -555,6 +526,44 @@ def _resolve_initial_retention(self, memory: Dict[str, Any]) -> float: 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/plugin.py b/src/powermem/intelligence/plugin.py index ec9fbcc27..d8dd497e5 100644 --- a/src/powermem/intelligence/plugin.py +++ b/src/powermem/intelligence/plugin.py @@ -160,10 +160,10 @@ def on_get(self, memory: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], bool intel_updates.update(reinforcement_result) # Apply reinforcement to normalized so downstream should_forget() - # and should_promote() see the boosted current_retention. - if intel_updates.get("current_retention") is not None: + # uses the boosted retention and its new timestamp anchor. + if intel_updates: norm_intel = dict(normalized.get("metadata", {}).get("intelligence") or {}) - norm_intel["current_retention"] = intel_updates["current_retention"] + norm_intel.update(intel_updates) norm_meta = dict(normalized.get("metadata") or {}) norm_meta["intelligence"] = norm_intel normalized = {**normalized, "metadata": norm_meta} diff --git a/tests/unit/intelligence/test_retention_runtime.py b/tests/unit/intelligence/test_retention_runtime.py index 0be2f1947..66bc0c036 100644 --- a/tests/unit/intelligence/test_retention_runtime.py +++ b/tests/unit/intelligence/test_retention_runtime.py @@ -301,6 +301,119 @@ def test_calculate_current_retention_defaults_without_stored_initial(algo): 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( @@ -343,10 +456,9 @@ def test_search_ranking_uses_effective_retention(): # ---- Tests for review-reinforcement protecting against forgetting ---- def test_should_forget_respects_stored_current_retention(algo): - """A memory whose base retention (initial * decay) falls below threshold - should NOT be forgotten if current_retention (from reinforcement) is still - above the threshold.""" + """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, @@ -356,6 +468,7 @@ def test_should_forget_respects_stored_current_retention(algo): "intelligence": { "initial_retention": 0.3, "current_retention": 0.8, + "last_reviewed": last_reviewed.isoformat(), } }, } @@ -412,9 +525,9 @@ def test_on_get_reinforced_memory_not_forgotten_same_call(): def test_calculate_current_retention_reflects_reinforcement(algo): - """calculate_current_retention should return max(base, stored - current_retention) so search ranking reflects reinforcement.""" + """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, @@ -424,6 +537,7 @@ def test_calculate_current_retention_reflects_reinforcement(algo): "intelligence": { "initial_retention": 0.3, "current_retention": 0.85, + "last_reviewed": last_reviewed.isoformat(), } }, } @@ -435,8 +549,8 @@ def test_calculate_current_retention_reflects_reinforcement(algo): result = algo.calculate_current_retention(memory) - assert result == pytest.approx(max(base_retention, 0.85)) - assert result >= 0.85 + assert result > base_retention + assert result <= 0.85 def test_search_ranking_reflects_reinforced_current_retention(): @@ -446,6 +560,7 @@ def test_search_ranking_reflects_reinforced_current_retention(): {"intelligent_memory": {"decay_rate": 1.5, "initial_retention": 1.0}} ) created_at = get_current_datetime() - timedelta(hours=50) + last_reviewed = get_current_datetime() results = [ { @@ -470,6 +585,7 @@ def test_search_ranking_reflects_reinforced_current_retention(): "intelligence": { "initial_retention": 0.3, "current_retention": 0.85, + "last_reviewed": last_reviewed.isoformat(), } }, },