/headroom/proxy/memory_rank_policy.py

"""Pure memory ranking policy helpers.

This module owns timestamp parsing and recency score math for proxy memory
ranking. It deliberately avoids backend objects and ranker classes so the
formula can be tested, ported, and reused independently of retrieval adapters.
"""

from __future__ import annotations

import math
from datetime import datetime, timezone

UTC = timezone.utc


def parse_memory_created_at(value: object) -> datetime | None:
    """Best-effort parse of a memory timestamp into a UTC-aware datetime."""
    if value is None:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
    if isinstance(value, str):
        try:
            return datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError:
            return None
    return None


def memory_recency_factor(
    *,
    now: datetime,
    created_at: datetime | None,
    decay_days: float,
) -> float:
    """Compute the recency multiplier for one memory candidate.

    Missing timestamps and future timestamps are neutral. For normal historical
    timestamps the multiplier is ``exp(-age_days / decay_days)``.
    """
    if created_at is None:
        return 1.0
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=UTC)
    if now.tzinfo is None:
        now = now.replace(tzinfo=UTC)

    age_days = (now - created_at).total_seconds() / 86400.0
    if age_days <= 0:
        return 1.0
    return math.exp(-age_days / decay_days)


def boost_memory_score(
    *,
    score: float,
    now: datetime,
    created_at: datetime | None,
    decay_days: float,
) -> float:
    """Apply the recency multiplier to a backend similarity score."""
    return score * memory_recency_factor(
        now=now,
        created_at=created_at,
        decay_days=decay_days,
    )