docs(mnemosyne): mark memory_decay as shipped

Part of #1258.

nexus/mnemosyne/FEATURES.yaml

@@ -151,13 +151,16 @@ frontend:
 planned:
 
   memory_decay:
-    status: planned
+    status: shipped
+    files: [entry.py, archive.py]
     description: >
       Memories have living energy that fades with neglect and
       brightens with access. Vitality score based on access
-      frequency and recency. Was attempted in PR #1221 but
-      went stale — needs fresh implementation against current main.
+      frequency and recency. Exponential decay with 30-day half-life.
+      Touch boost with diminishing returns.
     priority: medium
+    merged_prs:
+      - "#TBD"  # Will be filled when PR is created
 
   memory_pulse:
     status: planned

nexus/mnemosyne/archive.py

@@ -389,6 +389,17 @@ class MnemosyneArchive:
         oldest_entry = timestamps[0] if timestamps else None
         newest_entry = timestamps[-1] if timestamps else None
 
+        # Vitality summary
+        if n > 0:
+            vitalities = [self._compute_vitality(e) for e in entries]
+            avg_vitality = round(sum(vitalities) / n, 4)
+            fading_count = sum(1 for v in vitalities if v < 0.3)
+            vibrant_count = sum(1 for v in vitalities if v > 0.7)
+        else:
+            avg_vitality = 0.0
+            fading_count = 0
+            vibrant_count = 0
+
         return {
             "entries": n,
             "total_links": total_links,
@@ -398,6 +409,9 @@ class MnemosyneArchive:
             "link_density": link_density,
             "oldest_entry": oldest_entry,
             "newest_entry": newest_entry,
+            "avg_vitality": avg_vitality,
+            "fading_count": fading_count,
+            "vibrant_count": vibrant_count,
         }
 
     def _build_adjacency(self) -> dict[str, set[str]]:
@@ -742,6 +756,188 @@ class MnemosyneArchive:
         results.sort(key=lambda e: e.created_at)
         return results
 
+    # ─── Memory Decay ─────────────────────────────────────────
+
+    # Decay parameters
+    _DECAY_HALF_LIFE_DAYS: float = 30.0  # Half-life for exponential decay
+    _TOUCH_BOOST_FACTOR: float = 0.1     # Base boost on access (diminishes as vitality → 1.0)
+
+    def touch(self, entry_id: str) -> ArchiveEntry:
+        """Record an access to an entry, boosting its vitality.
+
+        The boost is ``_TOUCH_BOOST_FACTOR * (1 - current_vitality)`` —
+        diminishing returns as vitality approaches 1.0 ensures entries
+        can never exceed 1.0 through touch alone.
+
+        Args:
+            entry_id: ID of the entry to touch.
+
+        Returns:
+            The updated ArchiveEntry.
+
+        Raises:
+            KeyError: If entry_id does not exist.
+        """
+        entry = self._entries.get(entry_id)
+        if entry is None:
+            raise KeyError(entry_id)
+
+        now = datetime.now(timezone.utc).isoformat()
+
+        # Compute current decayed vitality before boosting
+        current = self._compute_vitality(entry)
+        boost = self._TOUCH_BOOST_FACTOR * (1.0 - current)
+        entry.vitality = min(1.0, current + boost)
+        entry.last_accessed = now
+        self._save()
+        return entry
+
+    def _compute_vitality(self, entry: ArchiveEntry) -> float:
+        """Compute the current vitality of an entry based on time decay.
+
+        Uses exponential decay: ``v = base * 0.5 ^ (hours_since_access / half_life_hours)``
+
+        If the entry has never been accessed, uses ``created_at`` as the
+        reference point. New entries with no access start at full vitality.
+
+        Args:
+            entry: The archive entry.
+
+        Returns:
+            Current vitality as a float in [0.0, 1.0].
+        """
+        if entry.last_accessed is None:
+            # Never accessed — check age from creation
+            created = self._parse_dt(entry.created_at)
+            hours_elapsed = (datetime.now(timezone.utc) - created).total_seconds() / 3600
+        else:
+            last = self._parse_dt(entry.last_accessed)
+            hours_elapsed = (datetime.now(timezone.utc) - last).total_seconds() / 3600
+
+        half_life_hours = self._DECAY_HALF_LIFE_DAYS * 24
+        if hours_elapsed <= 0 or half_life_hours <= 0:
+            return entry.vitality
+
+        decayed = entry.vitality * (0.5 ** (hours_elapsed / half_life_hours))
+        return max(0.0, min(1.0, decayed))
+
+    def get_vitality(self, entry_id: str) -> dict:
+        """Get the current vitality status of an entry.
+
+        Args:
+            entry_id: ID of the entry.
+
+        Returns:
+            Dict with keys: entry_id, title, vitality, last_accessed, age_days
+
+        Raises:
+            KeyError: If entry_id does not exist.
+        """
+        entry = self._entries.get(entry_id)
+        if entry is None:
+            raise KeyError(entry_id)
+
+        current_vitality = self._compute_vitality(entry)
+        created = self._parse_dt(entry.created_at)
+        age_days = (datetime.now(timezone.utc) - created).days
+
+        return {
+            "entry_id": entry.id,
+            "title": entry.title,
+            "vitality": round(current_vitality, 4),
+            "last_accessed": entry.last_accessed,
+            "age_days": age_days,
+        }
+
+    def fading(self, limit: int = 10) -> list[dict]:
+        """Return entries with the lowest vitality (most neglected).
+
+        Args:
+            limit: Maximum number of entries to return.
+
+        Returns:
+            List of dicts sorted by vitality ascending (most faded first).
+            Each dict has keys: entry_id, title, vitality, last_accessed, age_days
+        """
+        scored = []
+        for entry in self._entries.values():
+            v = self._compute_vitality(entry)
+            created = self._parse_dt(entry.created_at)
+            age_days = (datetime.now(timezone.utc) - created).days
+            scored.append({
+                "entry_id": entry.id,
+                "title": entry.title,
+                "vitality": round(v, 4),
+                "last_accessed": entry.last_accessed,
+                "age_days": age_days,
+            })
+        scored.sort(key=lambda x: x["vitality"])
+        return scored[:limit]
+
+    def vibrant(self, limit: int = 10) -> list[dict]:
+        """Return entries with the highest vitality (most alive).
+
+        Args:
+            limit: Maximum number of entries to return.
+
+        Returns:
+            List of dicts sorted by vitality descending (most vibrant first).
+            Each dict has keys: entry_id, title, vitality, last_accessed, age_days
+        """
+        scored = []
+        for entry in self._entries.values():
+            v = self._compute_vitality(entry)
+            created = self._parse_dt(entry.created_at)
+            age_days = (datetime.now(timezone.utc) - created).days
+            scored.append({
+                "entry_id": entry.id,
+                "title": entry.title,
+                "vitality": round(v, 4),
+                "last_accessed": entry.last_accessed,
+                "age_days": age_days,
+            })
+        scored.sort(key=lambda x: x["vitality"], reverse=True)
+        return scored[:limit]
+
+    def apply_decay(self) -> dict:
+        """Apply time-based decay to all entries and persist.
+
+        Recomputes each entry's vitality based on elapsed time since
+        its last access (or creation if never accessed). Saves the
+        archive after updating.
+
+        Returns:
+            Dict with keys: total_entries, decayed_count, avg_vitality,
+            fading_count (entries below 0.3), vibrant_count (entries above 0.7)
+        """
+        decayed = 0
+        total_vitality = 0.0
+        fading_count = 0
+        vibrant_count = 0
+
+        for entry in self._entries.values():
+            old_v = entry.vitality
+            new_v = self._compute_vitality(entry)
+            if abs(new_v - old_v) > 1e-6:
+                entry.vitality = new_v
+                decayed += 1
+            total_vitality += entry.vitality
+            if entry.vitality < 0.3:
+                fading_count += 1
+            if entry.vitality > 0.7:
+                vibrant_count += 1
+
+        n = len(self._entries)
+        self._save()
+
+        return {
+            "total_entries": n,
+            "decayed_count": decayed,
+            "avg_vitality": round(total_vitality / n, 4) if n else 0.0,
+            "fading_count": fading_count,
+            "vibrant_count": vibrant_count,
+        }
+
     def rebuild_links(self, threshold: Optional[float] = None) -> int:
         """Recompute all links from scratch.
 

nexus/mnemosyne/entry.py

@@ -34,6 +34,8 @@ class ArchiveEntry:
     updated_at: Optional[str] = None  # Set on mutation; None means same as created_at
     links: list[str] = field(default_factory=list)  # IDs of related entries
     content_hash: Optional[str] = None  # SHA-256 of title+content for dedup
+    vitality: float = 1.0  # 0.0 (dead) to 1.0 (fully alive)
+    last_accessed: Optional[str] = None  # ISO datetime of last access; None = never accessed
 
     def __post_init__(self):
         if self.content_hash is None:
@@ -52,6 +54,8 @@ class ArchiveEntry:
             "updated_at": self.updated_at,
             "links": self.links,
             "content_hash": self.content_hash,
+            "vitality": self.vitality,
+            "last_accessed": self.last_accessed,
         }
 
     @classmethod

nexus/mnemosyne/tests/test_memory_decay.py

@@ -0,0 +1,278 @@
+"""Tests for Mnemosyne memory decay system."""
+
+import json
+import os
+import tempfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from nexus.mnemosyne.archive import MnemosyneArchive
+from nexus.mnemosyne.entry import ArchiveEntry
+
+
+@pytest.fixture
+def archive(tmp_path):
+    """Create a fresh archive for testing."""
+    path = tmp_path / "test_archive.json"
+    return MnemosyneArchive(archive_path=path)
+
+
+@pytest.fixture
+def populated_archive(tmp_path):
+    """Create an archive with some entries."""
+    path = tmp_path / "test_archive.json"
+    arch = MnemosyneArchive(archive_path=path)
+    arch.add(ArchiveEntry(title="Fresh Entry", content="Just added", topics=["test"]))
+    arch.add(ArchiveEntry(title="Old Entry", content="Been here a while", topics=["test"]))
+    arch.add(ArchiveEntry(title="Another Entry", content="Some content", topics=["other"]))
+    return arch
+
+
+class TestVitalityFields:
+    """Test that vitality fields exist on entries."""
+
+    def test_entry_has_vitality_default(self):
+        entry = ArchiveEntry(title="Test", content="Content")
+        assert entry.vitality == 1.0
+
+    def test_entry_has_last_accessed_default(self):
+        entry = ArchiveEntry(title="Test", content="Content")
+        assert entry.last_accessed is None
+
+    def test_entry_roundtrip_with_vitality(self):
+        entry = ArchiveEntry(
+            title="Test", content="Content",
+            vitality=0.75,
+            last_accessed="2024-01-01T00:00:00+00:00"
+        )
+        d = entry.to_dict()
+        assert d["vitality"] == 0.75
+        assert d["last_accessed"] == "2024-01-01T00:00:00+00:00"
+        restored = ArchiveEntry.from_dict(d)
+        assert restored.vitality == 0.75
+        assert restored.last_accessed == "2024-01-01T00:00:00+00:00"
+
+
+class TestTouch:
+    """Test touch() access recording and vitality boost."""
+
+    def test_touch_sets_last_accessed(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        assert entry.last_accessed is None
+        touched = archive.touch(entry.id)
+        assert touched.last_accessed is not None
+
+    def test_touch_boosts_vitality(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content", vitality=0.5))
+        touched = archive.touch(entry.id)
+        # Boost = 0.1 * (1 - 0.5) = 0.05, so vitality should be ~0.55
+        # (assuming no time decay in test — instantaneous)
+        assert touched.vitality > 0.5
+        assert touched.vitality <= 1.0
+
+    def test_touch_diminishing_returns(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content", vitality=0.9))
+        touched = archive.touch(entry.id)
+        # Boost = 0.1 * (1 - 0.9) = 0.01, so vitality should be ~0.91
+        assert touched.vitality < 0.92
+        assert touched.vitality > 0.9
+
+    def test_touch_never_exceeds_one(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content", vitality=0.99))
+        for _ in range(10):
+            entry = archive.touch(entry.id)
+        assert entry.vitality <= 1.0
+
+    def test_touch_missing_entry_raises(self, archive):
+        with pytest.raises(KeyError):
+            archive.touch("nonexistent-id")
+
+    def test_touch_persists(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        archive.touch(entry.id)
+        # Reload archive
+        arch2 = MnemosyneArchive(archive_path=archive._path)
+        loaded = arch2.get(entry.id)
+        assert loaded.last_accessed is not None
+
+
+class TestGetVitality:
+    """Test get_vitality() status reporting."""
+
+    def test_get_vitality_basic(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        status = archive.get_vitality(entry.id)
+        assert status["entry_id"] == entry.id
+        assert status["title"] == "Test"
+        assert 0.0 <= status["vitality"] <= 1.0
+        assert status["age_days"] == 0
+
+    def test_get_vitality_missing_raises(self, archive):
+        with pytest.raises(KeyError):
+            archive.get_vitality("nonexistent-id")
+
+
+class TestComputeVitality:
+    """Test the decay computation."""
+
+    def test_new_entry_full_vitality(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        v = archive._compute_vitality(entry)
+        assert v == 1.0
+
+    def test_recently_touched_high_vitality(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        archive.touch(entry.id)
+        v = archive._compute_vitality(entry)
+        assert v > 0.99  # Should be essentially 1.0 since just touched
+
+    def test_old_entry_decays(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        # Simulate old access — set last_accessed to 60 days ago
+        old_date = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
+        entry.last_accessed = old_date
+        entry.vitality = 1.0
+        archive._save()
+        v = archive._compute_vitality(entry)
+        # 60 days with 30-day half-life: v = 1.0 * 0.5^(60/30) = 0.25
+        assert v < 0.3
+        assert v > 0.2
+
+    def test_very_old_entry_nearly_zero(self, archive):
+        entry = archive.add(ArchiveEntry(title="Test", content="Content"))
+        old_date = (datetime.now(timezone.utc) - timedelta(days=365)).isoformat()
+        entry.last_accessed = old_date
+        entry.vitality = 1.0
+        archive._save()
+        v = archive._compute_vitality(entry)
+        # 365 days / 30 half-life = ~12 half-lives -> ~0.0002
+        assert v < 0.01
+
+
+class TestFading:
+    """Test fading() — most neglected entries."""
+
+    def test_fading_returns_lowest_first(self, populated_archive):
+        entries = list(populated_archive._entries.values())
+        # Make one entry very old
+        old_entry = entries[1]
+        old_date = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat()
+        old_entry.last_accessed = old_date
+        old_entry.vitality = 1.0
+        populated_archive._save()
+
+        fading = populated_archive.fading(limit=3)
+        assert len(fading) <= 3
+        # First result should be the oldest
+        assert fading[0]["entry_id"] == old_entry.id
+        # Should be in ascending order
+        for i in range(len(fading) - 1):
+            assert fading[i]["vitality"] <= fading[i + 1]["vitality"]
+
+    def test_fading_empty_archive(self, archive):
+        fading = archive.fading()
+        assert fading == []
+
+    def test_fading_limit(self, populated_archive):
+        fading = populated_archive.fading(limit=2)
+        assert len(fading) == 2
+
+
+class TestVibrant:
+    """Test vibrant() — most alive entries."""
+
+    def test_vibrant_returns_highest_first(self, populated_archive):
+        entries = list(populated_archive._entries.values())
+        # Make one entry very old
+        old_entry = entries[1]
+        old_date = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat()
+        old_entry.last_accessed = old_date
+        old_entry.vitality = 1.0
+        populated_archive._save()
+
+        vibrant = populated_archive.vibrant(limit=3)
+        # Should be in descending order
+        for i in range(len(vibrant) - 1):
+            assert vibrant[i]["vitality"] >= vibrant[i + 1]["vitality"]
+        # First result should NOT be the old entry
+        assert vibrant[0]["entry_id"] != old_entry.id
+
+    def test_vibrant_empty_archive(self, archive):
+        vibrant = archive.vibrant()
+        assert vibrant == []
+
+
+class TestApplyDecay:
+    """Test apply_decay() bulk decay operation."""
+
+    def test_apply_decay_returns_stats(self, populated_archive):
+        result = populated_archive.apply_decay()
+        assert result["total_entries"] == 3
+        assert "decayed_count" in result
+        assert "avg_vitality" in result
+        assert "fading_count" in result
+        assert "vibrant_count" in result
+
+    def test_apply_decay_persists(self, populated_archive):
+        populated_archive.apply_decay()
+        # Reload
+        arch2 = MnemosyneArchive(archive_path=populated_archive._path)
+        result2 = arch2.apply_decay()
+        # Should show same entries
+        assert result2["total_entries"] == 3
+
+    def test_apply_decay_on_empty(self, archive):
+        result = archive.apply_decay()
+        assert result["total_entries"] == 0
+        assert result["avg_vitality"] == 0.0
+
+
+class TestStatsVitality:
+    """Test that stats() includes vitality summary."""
+
+    def test_stats_includes_vitality(self, populated_archive):
+        stats = populated_archive.stats()
+        assert "avg_vitality" in stats
+        assert "fading_count" in stats
+        assert "vibrant_count" in stats
+        assert 0.0 <= stats["avg_vitality"] <= 1.0
+
+    def test_stats_empty_archive(self, archive):
+        stats = archive.stats()
+        assert stats["avg_vitality"] == 0.0
+        assert stats["fading_count"] == 0
+        assert stats["vibrant_count"] == 0
+
+
+class TestDecayLifecycle:
+    """Integration test: full lifecycle from creation to fading."""
+
+    def test_entry_lifecycle(self, archive):
+        # Create
+        entry = archive.add(ArchiveEntry(title="Memory", content="A thing happened"))
+        assert entry.vitality == 1.0
+
+        # Touch a few times
+        for _ in range(5):
+            archive.touch(entry.id)
+
+        # Check it's vibrant
+        vibrant = archive.vibrant(limit=1)
+        assert len(vibrant) == 1
+        assert vibrant[0]["entry_id"] == entry.id
+
+        # Simulate time passing
+        entry.last_accessed = (datetime.now(timezone.utc) - timedelta(days=45)).isoformat()
+        entry.vitality = 0.8
+        archive._save()
+
+        # Apply decay
+        result = archive.apply_decay()
+        assert result["total_entries"] == 1
+
+        # Check it's now fading
+        fading = archive.fading(limit=1)
+        assert fading[0]["entry_id"] == entry.id
+        assert fading[0]["vitality"] < 0.5