V2 merge: purpose_agent/memory.py

purpose_agent/memory.py

@@ -0,0 +1,301 @@
+"""
+memory.py — Typed, versioned, scoped, reversible memory system.
+
+Replaces the flat heuristic library with a structured memory store where
+every memory has a kind, status, scope, trust score, and full provenance.
+
+Memory lifecycle:
+    candidate → quarantined (immune scan) → promoted (replay-tested) → archived
+                                         → rejected (if scan/test fails)
+
+Memory kinds:
+    purpose_contract   — the user's stated goal and constraints
+    user_preference    — learned user-specific preferences ("always cite sources")
+    skill_card         — reusable procedure extracted from successful trajectories
+    episodic_case      — a specific (state, action, outcome) triple worth remembering
+    failure_pattern    — a pattern that led to failure (negative heuristic)
+    critic_calibration — learned adjustments to the Purpose Function's scoring
+    tool_policy        — usage constraints and tips for specific tools
+"""
+from __future__ import annotations
+
+import json
+import logging
+import math
+import time
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+from purpose_agent.v2_types import MemoryScope
+
+logger = logging.getLogger(__name__)
+
+
+class MemoryKind(Enum):
+    PURPOSE_CONTRACT = "purpose_contract"
+    USER_PREFERENCE = "user_preference"
+    SKILL_CARD = "skill_card"
+    EPISODIC_CASE = "episodic_case"
+    FAILURE_PATTERN = "failure_pattern"
+    CRITIC_CALIBRATION = "critic_calibration"
+    TOOL_POLICY = "tool_policy"
+
+
+class MemoryStatus(Enum):
+    CANDIDATE = "candidate"
+    QUARANTINED = "quarantined"
+    PROMOTED = "promoted"
+    REJECTED = "rejected"
+    ARCHIVED = "archived"
+
+
+@dataclass
+class MemoryCard:
+    """
+    A single unit of agent memory. Every field is tracked for provenance.
+    """
+    id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
+    kind: MemoryKind = MemoryKind.SKILL_CARD
+    status: MemoryStatus = MemoryStatus.CANDIDATE
+    scope: MemoryScope = field(default_factory=MemoryScope)
+
+    # Content
+    content: str = ""               # The actual knowledge (human readable)
+    pattern: str = ""               # When does this apply?
+    strategy: str = ""              # What to do?
+    steps: list[str] = field(default_factory=list)
+
+    # Trust & utility
+    trust_score: float = 0.5        # 0=untrusted, 1=fully trusted
+    utility_score: float = 0.5      # how useful is this when retrieved?
+    times_retrieved: int = 0
+    times_helped: int = 0
+    times_hurt: int = 0
+
+    # Provenance
+    source_trace_id: str = ""
+    source_step: int = 0
+    created_at: float = field(default_factory=time.time)
+    created_by: str = ""            # agent name or "human" or "system"
+    version: int = 1
+    parent_id: str = ""             # if this is a revision of another memory
+    rejection_reason: str = ""
+    immune_scan_result: dict[str, Any] = field(default_factory=dict)
+
+    # Embedding for retrieval
+    embedding: list[float] | None = None
+
+    def update_utility(self, helped: bool, alpha: float = 0.1) -> None:
+        """Monte Carlo utility update: U_new = U + α(reward - U)."""
+        self.times_retrieved += 1
+        if helped:
+            self.times_helped += 1
+            self.utility_score += alpha * (1.0 - self.utility_score)
+        else:
+            self.times_hurt += 1
+            self.utility_score += alpha * (0.0 - self.utility_score)
+        self.utility_score = max(0.0, min(1.0, self.utility_score))
+
+    @property
+    def empirical_help_rate(self) -> float:
+        if self.times_retrieved == 0:
+            return 0.5
+        return self.times_helped / self.times_retrieved
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "kind": self.kind.value,
+            "status": self.status.value,
+            "scope": {
+                "agent_roles": self.scope.agent_roles,
+                "tool_names": self.scope.tool_names,
+                "task_categories": self.scope.task_categories,
+                "user_id": self.scope.user_id,
+            },
+            "content": self.content,
+            "pattern": self.pattern,
+            "strategy": self.strategy,
+            "steps": self.steps,
+            "trust_score": self.trust_score,
+            "utility_score": self.utility_score,
+            "times_retrieved": self.times_retrieved,
+            "times_helped": self.times_helped,
+            "times_hurt": self.times_hurt,
+            "source_trace_id": self.source_trace_id,
+            "created_at": self.created_at,
+            "created_by": self.created_by,
+            "version": self.version,
+            "parent_id": self.parent_id,
+            "status_detail": self.rejection_reason,
+            "immune_scan": self.immune_scan_result,
+        }
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "MemoryCard":
+        scope_d = d.get("scope", {})
+        return cls(
+            id=d.get("id", uuid.uuid4().hex[:12]),
+            kind=MemoryKind(d.get("kind", "skill_card")),
+            status=MemoryStatus(d.get("status", "candidate")),
+            scope=MemoryScope(
+                agent_roles=scope_d.get("agent_roles", []),
+                tool_names=scope_d.get("tool_names", []),
+                task_categories=scope_d.get("task_categories", []),
+                user_id=scope_d.get("user_id", ""),
+            ),
+            content=d.get("content", ""),
+            pattern=d.get("pattern", ""),
+            strategy=d.get("strategy", ""),
+            steps=d.get("steps", []),
+            trust_score=d.get("trust_score", 0.5),
+            utility_score=d.get("utility_score", 0.5),
+            times_retrieved=d.get("times_retrieved", 0),
+            times_helped=d.get("times_helped", 0),
+            times_hurt=d.get("times_hurt", 0),
+            source_trace_id=d.get("source_trace_id", ""),
+            created_at=d.get("created_at", time.time()),
+            created_by=d.get("created_by", ""),
+            version=d.get("version", 1),
+            parent_id=d.get("parent_id", ""),
+            rejection_reason=d.get("status_detail", ""),
+            immune_scan_result=d.get("immune_scan", {}),
+        )
+
+
+class MemoryStore:
+    """
+    Persistent, queryable store of MemoryCards.
+
+    Supports:
+    - Add/retrieve/update/reject memories
+    - Filter by kind, status, scope
+    - Ranked retrieval (relevance × trust × utility)
+    - Persistence to JSON
+    """
+
+    def __init__(self, persistence_path: str | None = None):
+        self._cards: dict[str, MemoryCard] = {}
+        self._path = Path(persistence_path) if persistence_path else None
+        if self._path and self._path.exists():
+            self._load()
+
+    def add(self, card: MemoryCard) -> MemoryCard:
+        """Add a memory card. Returns the card (with id assigned)."""
+        self._cards[card.id] = card
+        self._save()
+        return card
+
+    def get(self, card_id: str) -> MemoryCard | None:
+        return self._cards.get(card_id)
+
+    def update_status(self, card_id: str, status: MemoryStatus, reason: str = "") -> None:
+        card = self._cards.get(card_id)
+        if card:
+            card.status = status
+            if reason:
+                card.rejection_reason = reason
+            self._save()
+
+    def retrieve(
+        self,
+        query_text: str = "",
+        scope: MemoryScope | None = None,
+        kinds: list[MemoryKind] | None = None,
+        statuses: list[MemoryStatus] | None = None,
+        top_k: int = 10,
+    ) -> list[MemoryCard]:
+        """
+        Retrieve memories ranked by composite score.
+        Default: only promoted memories.
+        """
+        statuses = statuses or [MemoryStatus.PROMOTED]
+        candidates = []
+        query_emb = self._embed(query_text) if query_text else None
+
+        for card in self._cards.values():
+            if card.status not in statuses:
+                continue
+            if kinds and card.kind not in kinds:
+                continue
+            if scope and not card.scope.matches(scope):
+                continue
+
+            # Composite score: relevance × trust × utility
+            relevance = 0.5
+            if query_emb and card.embedding:
+                relevance = self._cosine(query_emb, card.embedding)
+            elif query_emb:
+                card.embedding = self._embed(card.content or card.pattern)
+                relevance = self._cosine(query_emb, card.embedding)
+
+            score = 0.4 * relevance + 0.3 * card.trust_score + 0.3 * card.utility_score
+            candidates.append((score, card))
+
+        candidates.sort(key=lambda x: -x[0])
+        return [c for _, c in candidates[:top_k]]
+
+    def get_by_status(self, status: MemoryStatus) -> list[MemoryCard]:
+        return [c for c in self._cards.values() if c.status == status]
+
+    def get_all(self) -> list[MemoryCard]:
+        return list(self._cards.values())
+
+    @property
+    def size(self) -> int:
+        return len(self._cards)
+
+    def stats(self) -> dict[str, Any]:
+        by_status = {}
+        by_kind = {}
+        for c in self._cards.values():
+            by_status[c.status.value] = by_status.get(c.status.value, 0) + 1
+            by_kind[c.kind.value] = by_kind.get(c.kind.value, 0) + 1
+        return {"total": self.size, "by_status": by_status, "by_kind": by_kind}
+
+    # --- Persistence ---
+
+    def _save(self) -> None:
+        if not self._path:
+            return
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        data = [c.to_dict() for c in self._cards.values()]
+        with open(self._path, "w") as f:
+            json.dump(data, f, indent=2, default=str)
+
+    def _load(self) -> None:
+        if not self._path or not self._path.exists():
+            return
+        try:
+            with open(self._path) as f:
+                data = json.load(f)
+            for d in data:
+                card = MemoryCard.from_dict(d)
+                self._cards[card.id] = card
+            logger.info(f"MemoryStore: loaded {len(self._cards)} cards")
+        except Exception as e:
+            logger.error(f"MemoryStore: load failed: {e}")
+
+    # --- Embedding (lightweight, swappable) ---
+
+    @staticmethod
+    def _embed(text: str) -> list[float]:
+        dim = 128
+        vec = [0.0] * dim
+        for i in range(len(text) - 2):
+            h = hash(text[i:i+3].lower()) % dim
+            vec[h] += 1.0
+        mag = math.sqrt(sum(x*x for x in vec))
+        return [x/mag for x in vec] if mag > 0 else vec
+
+    @staticmethod
+    def _cosine(a: list[float], b: list[float]) -> float:
+        if not a or not b or len(a) != len(b):
+            return 0.0
+        dot = sum(x*y for x, y in zip(a, b))
+        ma = math.sqrt(sum(x*x for x in a))
+        mb = math.sqrt(sum(x*x for x in b))
+        return dot / (ma * mb) if ma > 0 and mb > 0 else 0.0