"""Trust memory: verified vs unverified, provenance, confidence decay for AGI.""" from datetime import datetime, timezone from typing import Any from fusionagi._logger import logger def _utc_now() -> datetime: return datetime.now(timezone.utc) class TrustMemory: """ Tracks verified vs unverified claims, source provenance, and optional confidence decay over time. """ def __init__(self, decay_enabled: bool = False) -> None: self._entries: dict[str, dict[str, Any]] = {} # claim_id -> {verified, source, confidence, created_at} self._decay_enabled = decay_enabled def add( self, claim_id: str, verified: bool, source: str = "", confidence: float = 1.0, metadata: dict[str, Any] | None = None, ) -> None: """Record a claim with verification status and provenance.""" self._entries[claim_id] = { "verified": verified, "source": source, "confidence": confidence, "created_at": _utc_now(), "metadata": metadata or {}, } logger.debug("Trust memory: claim added", extra={"claim_id": claim_id, "verified": verified}) def get(self, claim_id: str) -> dict[str, Any] | None: """Return trust entry or None. Applies decay if enabled.""" e = self._entries.get(claim_id) if not e: return None if self._decay_enabled: # Simple decay: reduce confidence by 0.01 per day (placeholder) from datetime import timedelta age_days = (_utc_now() - e["created_at"]).total_seconds() / 86400 e = dict(e) e["confidence"] = max(0.0, e["confidence"] - 0.01 * age_days) return e def is_verified(self, claim_id: str) -> bool: """Return True if claim is marked verified.""" e = self._entries.get(claim_id) return e.get("verified", False) if e else False def set_verified(self, claim_id: str, verified: bool) -> bool: """Update verified status. Returns True if claim existed.""" if claim_id not in self._entries: return False self._entries[claim_id]["verified"] = verified return True