"""
4-tier memory architecture for Ember V3.
Tiers (from ephemeral to permanent):
working → Current session context. Deleted at session close unless promoted.
session → Key decisions/outcomes from recent sessions. TTL: 30 days.
relational → User patterns, recurring priorities. TTL: 180 days.
glacier → Permanent foundational facts. No decay, manual deletion only.
Inspired by HOPE/Nested Learning (NeurIPS 2025) — different memory types
decay at different rates, with automatic promotion between tiers.
"""
from __future__ import annotations
import time
import logging
from typing import Optional
logger = logging.getLogger("ember.tiers")
VALID_TIERS = ("working", "session", "relational", "glacier")
TIER_CONFIG = {
"working": {
"ttl_days": None, # Handled at session close
"decay_per_day": 0.0,
"auto_delete_on_session_close": True,
"promote_to": "session",
"promote_threshold": 0.7, # importance score
},
"session": {
"ttl_days": 30,
"decay_per_day": 0.033, # Full decay in ~30 days
"auto_delete_on_session_close": False,
"promote_to": "relational",
"promote_threshold": 3, # appearances across 3+ sessions
},
"relational": {
"ttl_days": 180,
"decay_per_day": 0.006, # Full decay in ~180 days
"auto_delete_on_session_close": False,
"promote_to": "glacier",
"promote_threshold": 90, # days stable without contradiction
},
"glacier": {
"ttl_days": None,
"decay_per_day": 0.0,
"auto_delete_on_session_close": False,
"promote_to": None,
"promote_threshold": None,
},
}
def validate_tier(tier: str) -> str:
"""Validate and return tier, raising ValueError if invalid."""
if tier not in VALID_TIERS:
raise ValueError(f"Invalid tier '{tier}'. Must be one of: {VALID_TIERS}")
return tier
def get_decay_factor(tier: str, age_days: float) -> float:
"""
Calculate how much a memory's effective importance has decayed.
Returns a multiplier in [0, 1] where 1 = no decay.
"""
config = TIER_CONFIG.get(tier)
if not config or config["decay_per_day"] == 0:
return 1.0
decay = config["decay_per_day"] * age_days
return max(0.0, 1.0 - decay)
def should_promote(tier: str, memory: dict) -> Optional[str]:
"""
Check if a memory should be promoted to the next tier.
Returns the target tier name if promotion is warranted, None otherwise.
Promotion rules:
working → session: importance >= 0.7
session → relational: accessed in 3+ separate sessions (access_count >= 3)
relational → glacier: stable 90+ days AND accessed 5+ times
"""
config = TIER_CONFIG.get(tier)
if not config or config["promote_to"] is None:
return None
now = time.time()
if tier == "working":
if memory.get("importance", 0) >= config["promote_threshold"]:
return config["promote_to"]
elif tier == "session":
if memory.get("access_count", 0) >= config["promote_threshold"]:
return config["promote_to"]
elif tier == "relational":
created = memory.get("created_at", now)
age_days = (now - created) / 86400
is_stable = memory.get("shadow_load", 0) < 0.1 and not memory.get("is_shadowed", False)
accessed_enough = memory.get("access_count", 0) >= 5
if age_days >= config["promote_threshold"] and is_stable and accessed_enough:
return config["promote_to"]
return None
def should_expire(tier: str, memory: dict) -> bool:
"""
Check if a memory has expired based on its tier's TTL.
Only session and relational tiers have TTLs.
Additional condition: access_count < 2 (unused memories expire faster).
"""
config = TIER_CONFIG.get(tier)
if not config or config["ttl_days"] is None:
return False
now = time.time()
created = memory.get("created_at", now)
age_days = (now - created) / 86400
if age_days > config["ttl_days"] and memory.get("access_count", 0) < 2:
return True
return False
def get_working_memories_to_promote(memories: list[dict]) -> tuple[list[dict], list[dict]]:
"""
At session close, evaluate working-tier memories.
Returns (to_promote, to_discard):
- to_promote: memories with importance >= 0.7 → promote to session
- to_discard: remaining working memories → delete
"""
to_promote = []
to_discard = []
for mem in memories:
if mem.get("tier") != "working":
continue
target = should_promote("working", mem)
if target:
to_promote.append(mem)
else:
to_discard.append(mem)
return to_promote, to_discard
def compute_shadow_load(contradiction_count: int) -> float:
"""Simple shadow load: number of contradictions / 10, capped at 1.0."""
return min(1.0, contradiction_count / 10.0)
