ForgettingMechanism

Purpose

Applies an Ebbinghaus-forgetting-curve-inspired decay to a collection of MemoryEntry objects, combining time decay, access-frequency-based decay slowdown ("spaced repetition"), and an absolute importance floor to decide which entries should be kept vs. marked for deletion. Used by AIMemoryManager.run_maintenance() on the contents of WorkingMemory each maintenance pass.

ForgettingConfig (dataclass)

@dataclass
class ForgettingConfig:
    base_decay_rate: float = 0.15
    recency_boost_per_access: float = 0.05
    max_recency_boost: float = 2.0
    min_importance: float = 0.05
    deletion_threshold: float = 0.05
    high_frequency_threshold: int = 5
    apply_decay_on_read: bool = False

No validation is performed on any field (unlike SelectionConfig, which validates its weight sum) — negative rates, a deletion_threshold greater than 1.0, etc. are all accepted silently.

high_frequency_threshold and apply_decay_on_read are both declared but not read by any method in this class — apply_decay_on_read in particular suggests a documented-but-unimplemented feature (its name implies decay should be applied whenever an entry is read/accessed elsewhere in the pipeline, but no other class in this source tree checks this flag).

Constructor

def __init__(self, config: ForgettingConfig | None = None) -> None

Defaults to ForgettingConfig().

apply() — main entry point

def apply(self, entries: List[MemoryEntry]) -> Tuple[List[MemoryEntry], List[MemoryEntry]]

For each entry: applies _decay_entry() (mutates importance/decay_factor in place), then classifies as forgotten if effective_importance <= deletion_threshold, else alive. Returns (alive, forgotten).

apply_to_single(entry) -> bool

Decays one entry in place (same _decay_entry logic) and returns whether effective_importance > deletion_threshold (i.e. True = keep). Does not suffer from the empty-list issue since it operates on a single entry, not a list.

score_retention(entry) -> float

Non-mutating preview of retention likelihood — computes a projected decayed importance without changing entry.importance/decay_factor. Uses entry.last_access (not entry.timestamp) as the basis, unlike MemorySelector._recency_score, which uses entry.timestamp.

report(entries) -> Dict[str, Any]

Non-mutating diagnostic: identifies entries with score_retention(e) < 0.2 as "at risk," returns total, at_risk_count, at_risk_ids (first 10 only), and the active config's key parameters. Safe to call on an empty list — returns {"total": 0, "at_risk_count": 0, "at_risk_ids": [], "config": {...}} without hitting the tuple-unpacking bug (this method doesn't call apply()).

Internal helpers

_effective_decay_rate(entry) -> float

access_boost = min(max_recency_boost, 1.0 + recency_boost_per_access * entry.access_count)
return base_decay_rate / access_boost

Frequently-accessed entries get a larger access_boost divisor, hence a lower effective decay rate — this is the "spaced repetition" mechanic: entries you keep re-accessing decay more slowly.

_decay_entry(entry, now) -> None

hours_since_access = (now - entry.last_access) / 3600.0
rate = _effective_decay_rate(entry)
half_life_hours = 24.0 / max(rate, 1e-9)
lam = math.log(2) / half_life_hours
decay_multiplier = math.exp(-lam * hours_since_access)

entry.decay_factor = max(0.05, entry.decay_factor * decay_multiplier)
entry.importance = max(min_importance, entry.importance * decay_multiplier)

[!DANGER] Compounding decay on repeated apply() calls without an intervening access Both entry.decay_factor and entry.importance are updated by multiplying the current (already-decayed) value by a freshly-computed decay_multiplier, rather than recomputing from original_importance. If apply() (or apply_to_single()) is called more than once for the same entry without entry.last_access changing in between (e.g. two run_maintenance() passes back-to-back with no user activity), hours_since_access is nearly identical both times, so decay_multiplier is nearly identical both times — but it gets multiplied onto an importance that's already been reduced by the first call. The net effect is the entry decays roughly twice as much as a single correctly-computed decay for that elapsed time window would produce. This mirrors the same class of compounding issue present in LongTermMemory.apply_global_decay. See Known Issues.

This uses an exponential decay model (unlike MemoryEntry.apply_decay's linear model and LongTermMemory.apply_global_decay's linear-then-multiplicative model) — three different decay formulas coexist across the package for conceptually the same operation.

Raises

No exceptions are raised by any method here under normal (non-empty-list) operation — but see the apply([]) string-return defect above, which causes a ValueError at the caller's unpacking site, not inside ForgettingMechanism itself.

Thread safety

_decay_entry mutates the passed-in MemoryEntry objects directly — safe for concurrent calls with different entry lists, but calling apply()/apply_to_single() concurrently on overlapping sets of the same MemoryEntry objects from different threads is an unsynchronized race on those entries' importance/decay_factor fields (the entries themselves have no internal locking, per MemoryEntry).

Complexity

O(n) for apply()/report(), where n = len(entries). O(1) for apply_to_single()/score_retention().

Examples

from zerda.memory import ForgettingMechanism, ForgettingConfig, MemoryEntry, MemoryType
import time

fm = ForgettingMechanism(ForgettingConfig(base_decay_rate=0.2, deletion_threshold=0.1))

entries = [MemoryEntry(content={"input": f"note {i}", "output": "ok"},
                        timestamp=time.time() - i * 86400,
                        memory_type=MemoryType.WORKING, importance=0.5)
           for i in range(10)]
for e in entries:
    e.last_access = e.timestamp  # simulate no access since creation

# 1. Basic — always check for empty input first, given the known defect
if entries:
    alive, forgotten = fm.apply(entries)
    print(len(alive), len(forgotten))
else:
    alive, forgotten = [], []

# 2. Intermediate — single-entry decay
keep = fm.apply_to_single(entries[0])
print(keep, entries[0].effective_importance)

# 3. Advanced — non-mutating preview before committing to a full apply()
score = fm.score_retention(entries[5])
print(f"Retention score: {score:.2f}")

# 4. Error handling — reproduce and guard against the empty-list defect
try:
    alive, forgotten = fm.apply([])
except ValueError as e:
    print("Known defect triggered:", e)
    alive, forgotten = [], []  # safe fallback

# 5. Production — diagnostic report before committing to deletion
report = fm.report(entries)
print(f"{report['at_risk_count']} of {report['total']} entries at risk of forgetting")

Best practices

• Always guard apply() calls with if entries: (or equivalent) until the empty-list return-type defect is fixed upstream — this is the single most important operational precaution for this class.
• Use score_retention()/report() for read-only "what would happen" previews before committing to a destructive apply() pass.
• Don't call apply()/apply_to_single()/run_maintenance() more frequently than meaningful time has elapsed since the last call, given the compounding-decay caveat.

Common mistakes

• Calling alive, forgotten = forgetter.apply(entries) without checking entries is non-empty first — the single most impactful mistake possible with this class, given it crashes AIMemoryManager.run_maintenance() under an ordinary, easily-reached condition.
• Running maintenance passes back-to-back expecting linear, additive decay — it compounds instead.

Related APIs

• MemoryEntry.apply_decay — a different (linear) decay formula, applied by ShortTermMemory.
• LongTermMemory.apply_global_decay — a third decay formula, with the same compounding caveat.
• AIMemoryManager.run_maintenance() — the primary caller, and the concrete site where the empty-list defect manifests.