diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index 8db762152..23befa899 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -12,7 +12,7 @@ Config in $HERMES_HOME/config.yaml (profile-scoped): auto_extract: false default_trust: 0.5 min_trust_threshold: 0.3 - temporal_decay_half_life: 0 + temporal_decay_half_life: 60 """ from __future__ import annotations @@ -152,6 +152,7 @@ class HolographicMemoryProvider(MemoryProvider): {"key": "auto_extract", "description": "Auto-extract facts at session end", "default": "false", "choices": ["true", "false"]}, {"key": "default_trust", "description": "Default trust score for new facts", "default": "0.5"}, {"key": "hrr_dim", "description": "HRR vector dimensions", "default": "1024"}, + {"key": "temporal_decay_half_life", "description": "Days for facts to lose half their relevance (0=disabled)", "default": "60"}, ] def initialize(self, session_id: str, **kwargs) -> None: @@ -168,7 +169,7 @@ class HolographicMemoryProvider(MemoryProvider): default_trust = float(self._config.get("default_trust", 0.5)) hrr_dim = int(self._config.get("hrr_dim", 1024)) hrr_weight = float(self._config.get("hrr_weight", 0.3)) - temporal_decay = int(self._config.get("temporal_decay_half_life", 0)) + temporal_decay = int(self._config.get("temporal_decay_half_life", 60)) self._store = MemoryStore(db_path=db_path, default_trust=default_trust, hrr_dim=hrr_dim) self._retriever = FactRetriever( diff --git a/plugins/memory/holographic/retrieval.py b/plugins/memory/holographic/retrieval.py index a673dcef8..a36f2ca8d 100644 --- a/plugins/memory/holographic/retrieval.py +++ b/plugins/memory/holographic/retrieval.py @@ -98,7 +98,15 @@ class FactRetriever: # Optional temporal decay if self.half_life > 0: - score *= self._temporal_decay(fact.get("updated_at") or fact.get("created_at")) + decay = self._temporal_decay(fact.get("updated_at") or fact.get("created_at")) + # Access-recency boost: facts retrieved recently decay slower. + # A fact accessed within 1 half-life gets up to 1.5x the decay + # factor, tapering to 1.0x (no boost) after 2 half-lives. + last_accessed = fact.get("last_accessed_at") + if last_accessed: + access_boost = self._access_recency_boost(last_accessed) + decay = min(1.0, decay * access_boost) + score *= decay fact["score"] = score scored.append(fact) @@ -591,3 +599,41 @@ class FactRetriever: return math.pow(0.5, age_days / self.half_life) except (ValueError, TypeError): return 1.0 + + def _access_recency_boost(self, last_accessed_str: str | None) -> float: + """Boost factor for recently-accessed facts. Range [1.0, 1.5]. + + Facts accessed within 1 half-life get up to 1.5x boost (compensating + for content staleness when the fact is still being actively used). + Boost decays linearly to 1.0 (no boost) at 2 half-lives. + + Returns 1.0 if half-life is disabled or timestamp is missing. + """ + if not self.half_life or not last_accessed_str: + return 1.0 + + try: + if isinstance(last_accessed_str, str): + ts = datetime.fromisoformat(last_accessed_str.replace("Z", "+00:00")) + else: + ts = last_accessed_str + + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + + age_days = (datetime.now(timezone.utc) - ts).total_seconds() / 86400 + if age_days < 0: + return 1.5 # Future timestamp = just accessed + + half_lives_since_access = age_days / self.half_life + + if half_lives_since_access <= 1.0: + # Within 1 half-life: linearly from 1.5 (just now) to 1.0 (at 1 HL) + return 1.0 + 0.5 * (1.0 - half_lives_since_access) + elif half_lives_since_access <= 2.0: + # Between 1 and 2 half-lives: linearly from 1.0 to 1.0 (no boost) + return 1.0 + else: + return 1.0 + except (ValueError, TypeError): + return 1.0 diff --git a/tests/plugins/memory/test_temporal_decay.py b/tests/plugins/memory/test_temporal_decay.py new file mode 100644 index 000000000..dbe12c969 --- /dev/null +++ b/tests/plugins/memory/test_temporal_decay.py @@ -0,0 +1,209 @@ +"""Tests for temporal decay and access-recency boost in holographic memory (#241).""" + +import math +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import pytest + + +class TestTemporalDecay: + """Test _temporal_decay exponential decay formula.""" + + def _make_retriever(self, half_life=60): + from plugins.memory.holographic.retrieval import FactRetriever + store = MagicMock() + return FactRetriever(store=store, temporal_decay_half_life=half_life) + + def test_fresh_fact_no_decay(self): + """A fact updated today should have decay ≈ 1.0.""" + r = self._make_retriever(half_life=60) + now = datetime.now(timezone.utc).isoformat() + decay = r._temporal_decay(now) + assert decay > 0.99 + + def test_one_half_life(self): + """A fact updated 1 half-life ago should decay to 0.5.""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat() + decay = r._temporal_decay(old) + assert abs(decay - 0.5) < 0.01 + + def test_two_half_lives(self): + """A fact updated 2 half-lives ago should decay to 0.25.""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=120)).isoformat() + decay = r._temporal_decay(old) + assert abs(decay - 0.25) < 0.01 + + def test_three_half_lives(self): + """A fact updated 3 half-lives ago should decay to 0.125.""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=180)).isoformat() + decay = r._temporal_decay(old) + assert abs(decay - 0.125) < 0.01 + + def test_half_life_disabled(self): + """When half_life=0, decay should always be 1.0.""" + r = self._make_retriever(half_life=0) + old = (datetime.now(timezone.utc) - timedelta(days=365)).isoformat() + assert r._temporal_decay(old) == 1.0 + + def test_none_timestamp(self): + """Missing timestamp should return 1.0 (no decay).""" + r = self._make_retriever(half_life=60) + assert r._temporal_decay(None) == 1.0 + + def test_empty_timestamp(self): + r = self._make_retriever(half_life=60) + assert r._temporal_decay("") == 1.0 + + def test_invalid_timestamp(self): + """Malformed timestamp should return 1.0 (fail open).""" + r = self._make_retriever(half_life=60) + assert r._temporal_decay("not-a-date") == 1.0 + + def test_future_timestamp(self): + """Future timestamp should return 1.0 (no decay for future dates).""" + r = self._make_retriever(half_life=60) + future = (datetime.now(timezone.utc) + timedelta(days=10)).isoformat() + assert r._temporal_decay(future) == 1.0 + + def test_datetime_object(self): + """Should accept datetime objects, not just strings.""" + r = self._make_retriever(half_life=60) + old = datetime.now(timezone.utc) - timedelta(days=60) + decay = r._temporal_decay(old) + assert abs(decay - 0.5) < 0.01 + + def test_different_half_lives(self): + """30-day half-life should decay faster than 90-day.""" + r30 = self._make_retriever(half_life=30) + r90 = self._make_retriever(half_life=90) + old = (datetime.now(timezone.utc) - timedelta(days=45)).isoformat() + assert r30._temporal_decay(old) < r90._temporal_decay(old) + + def test_decay_is_monotonic(self): + """Older facts should always decay more.""" + r = self._make_retriever(half_life=60) + now = datetime.now(timezone.utc) + d1 = r._temporal_decay((now - timedelta(days=10)).isoformat()) + d2 = r._temporal_decay((now - timedelta(days=30)).isoformat()) + d3 = r._temporal_decay((now - timedelta(days=60)).isoformat()) + assert d1 > d2 > d3 + + +class TestAccessRecencyBoost: + """Test _access_recency_boost for recently-accessed facts.""" + + def _make_retriever(self, half_life=60): + from plugins.memory.holographic.retrieval import FactRetriever + store = MagicMock() + return FactRetriever(store=store, temporal_decay_half_life=half_life) + + def test_just_accessed_max_boost(self): + """A fact accessed just now should get maximum boost (1.5).""" + r = self._make_retriever(half_life=60) + now = datetime.now(timezone.utc).isoformat() + boost = r._access_recency_boost(now) + assert boost > 1.45 # Near 1.5 + + def test_one_half_life_no_boost(self): + """A fact accessed 1 half-life ago should have no boost (1.0).""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat() + boost = r._access_recency_boost(old) + assert abs(boost - 1.0) < 0.01 + + def test_half_way_boost(self): + """A fact accessed 0.5 half-lives ago should get ~1.25 boost.""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat() + boost = r._access_recency_boost(old) + assert abs(boost - 1.25) < 0.05 + + def test_beyond_one_half_life_no_boost(self): + """Beyond 1 half-life, boost should be 1.0.""" + r = self._make_retriever(half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat() + boost = r._access_recency_boost(old) + assert boost == 1.0 + + def test_disabled_no_boost(self): + """When half_life=0, boost should be 1.0.""" + r = self._make_retriever(half_life=0) + now = datetime.now(timezone.utc).isoformat() + assert r._access_recency_boost(now) == 1.0 + + def test_none_timestamp(self): + r = self._make_retriever(half_life=60) + assert r._access_recency_boost(None) == 1.0 + + def test_invalid_timestamp(self): + r = self._make_retriever(half_life=60) + assert r._access_recency_boost("bad") == 1.0 + + def test_boost_range(self): + """Boost should always be in [1.0, 1.5].""" + r = self._make_retriever(half_life=60) + now = datetime.now(timezone.utc) + for days in [0, 1, 15, 30, 45, 59, 60, 90, 365]: + ts = (now - timedelta(days=days)).isoformat() + boost = r._access_recency_boost(ts) + assert 1.0 <= boost <= 1.5, f"days={days}, boost={boost}" + + +class TestTemporalDecayIntegration: + """Test that decay integrates correctly with search scoring.""" + + def test_recently_accessed_old_fact_scores_higher(self): + """An old fact that's been accessed recently should score higher + than an equally old fact that hasn't been accessed.""" + from plugins.memory.holographic.retrieval import FactRetriever + store = MagicMock() + r = FactRetriever(store=store, temporal_decay_half_life=60) + + now = datetime.now(timezone.utc) + old_date = (now - timedelta(days=120)).isoformat() # 2 half-lives old + recent_access = (now - timedelta(days=10)).isoformat() # accessed 10 days ago + old_access = (now - timedelta(days=200)).isoformat() # accessed 200 days ago + + # Old fact, recently accessed + decay1 = r._temporal_decay(old_date) + boost1 = r._access_recency_boost(recent_access) + effective1 = min(1.0, decay1 * boost1) + + # Old fact, not recently accessed + decay2 = r._temporal_decay(old_date) + boost2 = r._access_recency_boost(old_access) + effective2 = min(1.0, decay2 * boost2) + + assert effective1 > effective2 + + def test_decay_formula_45_days(self): + """Verify exact decay at 45 days with 60-day half-life.""" + from plugins.memory.holographic.retrieval import FactRetriever + r = FactRetriever(store=MagicMock(), temporal_decay_half_life=60) + old = (datetime.now(timezone.utc) - timedelta(days=45)).isoformat() + decay = r._temporal_decay(old) + expected = math.pow(0.5, 45/60) + assert abs(decay - expected) < 0.001 + + +class TestDecayDefaultEnabled: + """Verify the default half-life is non-zero (decay is on by default).""" + + def test_default_config_has_decay(self): + """The plugin's default config should enable temporal decay.""" + from plugins.memory.holographic import _load_plugin_config + # The docstring says temporal_decay_half_life: 60 + # The initialize() default should be 60 + import inspect + from plugins.memory.holographic import HolographicMemoryProvider + src = inspect.getsource(HolographicMemoryProvider.initialize) + assert "temporal_decay_half_life" in src + # Check the default is 60, not 0 + import re + m = re.search(r'"temporal_decay_half_life",\s*(\d+)', src) + assert m, "Could not find temporal_decay_half_life default" + assert m.group(1) == "60", f"Default is {m.group(1)}, expected 60"