Temporal

"""Bitemporal evidence, permission-aware lookup, and freshness diagnostics.

RAG systems need at least two clocks:

* **valid time** — when a statement was true in the represented world; and
* **transaction time** — when the retrieval system learned or indexed it.

Conflating these clocks creates hindsight leakage in evaluation and stale or
anachronistic answers in production.  This module provides a small immutable
fact model and an append-only bitemporal store suitable for notebook-scale
experiments.
"""

import math
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, Iterable, List, Optional, Sequence, Set, Tuple


def parse_instant(value: str) -> datetime:
    """Parse an ISO-8601 timestamp and require an explicit timezone."""

    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
    parsed = datetime.fromisoformat(normalized)
    if parsed.tzinfo is None:
        raise ValueError("timestamps must include an explicit timezone")
    return parsed.astimezone(timezone.utc)


@dataclass(frozen=True)
class TemporalFact:
    """One append-only version of a source claim."""

    source_id: str
    version_id: str
    key: str
    value: str
    valid_from: str
    observed_at: str
    valid_to: str = ""
    retracted_at: str = ""
    principals: Tuple[str, ...] = ()
    trust_domain: str = "unverified"
    provenance: str = ""

    def __post_init__(self) -> None:
        if not all((self.source_id.strip(), self.version_id.strip(), self.key.strip())):
            raise ValueError("source_id, version_id, and key must be non-empty")
        start = parse_instant(self.valid_from)
        observed = parse_instant(self.observed_at)
        if self.valid_to and parse_instant(self.valid_to) <= start:
            raise ValueError("valid_to must be after valid_from")
        if self.retracted_at and parse_instant(self.retracted_at) < observed:
            raise ValueError("retracted_at cannot precede observed_at")

    def valid_at(self, instant: datetime) -> bool:
        """Whether the claim describes the world at ``instant``."""

        start = parse_instant(self.valid_from)
        end = parse_instant(self.valid_to) if self.valid_to else None
        return start <= instant and (end is None or instant < end)

    def known_at(self, instant: datetime) -> bool:
        """Whether the store knew the version at ``instant``."""

        observed = parse_instant(self.observed_at)
        retracted = parse_instant(self.retracted_at) if self.retracted_at else None
        return observed <= instant and (retracted is None or instant < retracted)

    def accessible_to(self, principals: Iterable[str]) -> bool:
        """Public facts have no ACL; restricted facts require an intersection."""

        allowed = set(self.principals)
        return not allowed or bool(allowed & set(principals))


@dataclass(frozen=True)
class TemporalLookup:
    """Lookup result including ambiguity instead of silently choosing a value."""

    key: str
    valid_at: str
    known_at: str
    facts: Tuple[TemporalFact, ...]
    ambiguous: bool
    values: Tuple[str, ...]


class BitemporalStore:
    """Append-only, permission-aware bitemporal fact store."""

    def __init__(self, facts: Sequence[TemporalFact] = ()) -> None:
        self._facts: List[TemporalFact] = []
        self._version_ids: Set[str] = set()
        for fact in facts:
            self.append(fact)

    @property
    def facts(self) -> Tuple[TemporalFact, ...]:
        return tuple(self._facts)

    def append(self, fact: TemporalFact) -> None:
        """Append a unique source version; history is never overwritten."""

        if fact.version_id in self._version_ids:
            raise ValueError("duplicate temporal version: " + fact.version_id)
        self._facts.append(fact)
        self._version_ids.add(fact.version_id)

    def lookup(
        self,
        key: str,
        valid_at: str,
        known_at: str,
        principals: Iterable[str] = (),
        trust_domains: Iterable[str] = (),
    ) -> TemporalLookup:
        """Return every fact simultaneously valid, known, authorized, and trusted."""

        valid_instant = parse_instant(valid_at)
        known_instant = parse_instant(known_at)
        trusted = set(trust_domains)
        matches = [
            fact
            for fact in self._facts
            if fact.key == key
            and fact.valid_at(valid_instant)
            and fact.known_at(known_instant)
            and fact.accessible_to(principals)
            and (not trusted or fact.trust_domain in trusted)
        ]
        matches.sort(
            key=lambda fact: (
                parse_instant(fact.observed_at),
                parse_instant(fact.valid_from),
                fact.version_id,
            ),
            reverse=True,
        )
        values = tuple(sorted({fact.value for fact in matches}))
        return TemporalLookup(
            key=key,
            valid_at=valid_at,
            known_at=known_at,
            facts=tuple(matches),
            ambiguous=len(values) > 1,
            values=values,
        )

    def latest_known(
        self,
        key: str,
        known_at: str,
        principals: Iterable[str] = (),
    ) -> Optional[TemporalFact]:
        """Latest non-retracted version by observation time, regardless of valid time."""

        instant = parse_instant(known_at)
        matches = [
            fact
            for fact in self._facts
            if fact.key == key and fact.known_at(instant) and fact.accessible_to(principals)
        ]
        if not matches:
            return None
        return max(matches, key=lambda fact: (parse_instant(fact.observed_at), fact.version_id))

    def change_log(self, source_id: str = "") -> Tuple[TemporalFact, ...]:
        """Return versions in transaction-time order for replay and audit."""

        facts = [fact for fact in self._facts if not source_id or fact.source_id == source_id]
        facts.sort(key=lambda fact: (parse_instant(fact.observed_at), fact.version_id))
        return tuple(facts)


def age_seconds(observed_at: str, query_time: str) -> float:
    """Non-negative age of indexed evidence at query time."""

    delta = (parse_instant(query_time) - parse_instant(observed_at)).total_seconds()
    return max(0.0, delta)


def exponential_time_decay(
    score: float,
    observed_at: str,
    query_time: str,
    half_life_seconds: float,
) -> float:
    """Apply exponential recency decay without changing the sign of a score."""

    if half_life_seconds <= 0:
        raise ValueError("half_life_seconds must be positive")
    decay = math.exp(-math.log(2.0) * age_seconds(observed_at, query_time) / half_life_seconds)
    return score * decay


def freshness_sla_met(observed_at: str, query_time: str, maximum_age_seconds: float) -> bool:
    """Whether evidence age stays inside an explicit ingestion/query SLA."""

    if maximum_age_seconds < 0:
        raise ValueError("maximum_age_seconds must be non-negative")
    return age_seconds(observed_at, query_time) <= maximum_age_seconds


def stale_rate(
    observed_times: Sequence[str],
    query_time: str,
    maximum_age_seconds: float,
) -> float:
    """Fraction of returned evidence units violating the freshness SLA."""

    if not observed_times:
        return 0.0
    stale = sum(
        not freshness_sla_met(observed, query_time, maximum_age_seconds)
        for observed in observed_times
    )
    return stale / len(observed_times)


def cache_identity(
    query: str,
    corpus_snapshot: str,
    permission_fingerprint: str,
    valid_at: str,
    model_revision: str,
) -> Tuple[str, ...]:
    """Fields that must separate temporal and authorization-sensitive caches."""

    if not all((query, corpus_snapshot, permission_fingerprint, valid_at, model_revision)):
        raise ValueError("all cache identity fields must be non-empty")
    return (query, corpus_snapshot, permission_fingerprint, valid_at, model_revision)