feat(mnemosyne): add memory decay system to MnemosyneArchive

Part of #1258.
- touch(entry_id): record access, boost vitality
- get_vitality(entry_id): current vitality status  
- fading(limit): most neglected entries
- vibrant(limit): most alive entries
- apply_decay(): decay all entries, persist
- stats() updated with avg_vitality, fading_count, vibrant_count

Decay: exponential with 30-day half-life.
Touch: 0.1 * (1 - current_vitality) — diminishing returns.
This commit is contained in:
2026-04-12 05:42:37 +00:00
committed by Alexander Whitestone
parent be45c7f88b
commit 6dce93d948

View File

@@ -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.