Add policy drift memory benchmark

examples/benchmarks/policy_drift_memory/README.md

@@ -0,0 +1,75 @@
+# Policy Drift Memory Benchmark
+
+This benchmark stress-tests the core 2026 agent-memory tradeoff from Memanto
+issue #639: retrieval accuracy versus resource footprint when instructions mutate
+across sessions.
+
+The dataset models support and operations agents that receive policy updates,
+privacy reversals, launch-train changes, and escalation changes. The benchmark
+then asks current-state questions where stale memories are actively harmful.
+
+## Compared Backends
+
+- `memanto_style_active_digest`: an offline Memanto-style typed digest that
+  stores the current fact per key and returns compact current-state evidence.
+- `episode_graph_baseline`: a graph-like episode memory baseline that groups
+  values by key but keeps historical values in retrieval.
+- `recent_window_3`: a recency-only baseline using the last three events for an
+  entity.
+- `append_only_log`: a passive memory baseline that replays every event for an
+  entity.
+
+The default suite is credential-free so reviewers can reproduce it without a
+Moorcheh API key. It is intentionally structured so a live Memanto CLI adapter
+can be added without changing the dataset or scoring contract.
+
+## Metrics
+
+- `accuracy`: fraction of queries that include all required current facts and no
+  forbidden stale facts.
+- `total_retrieved_tokens`: approximate token footprint returned to the agent
+  across all benchmark queries.
+- `avg_retrieved_tokens`: average retrieved context size per query.
+- `p95_latency_ms`: p95 retrieval latency over repeated in-process retrievals.
+
+## Run
+
+```bash
+python examples/benchmarks/policy_drift_memory/benchmark.py --repeats 200
+```
+
+Optional full detail output:
+
+```bash
+python examples/benchmarks/policy_drift_memory/benchmark.py \
+  --repeats 200 \
+  --output-json examples/benchmarks/policy_drift_memory/results.json
+```
+
+## Expected Shape
+
+On the included dataset, the active digest should preserve current facts while
+dropping stale instructions, so it should reach full accuracy with a much
+smaller retrieval footprint than passive baselines. The exact latency values are
+machine-dependent, but relative token footprint and stale-fact failures should
+be stable.
+
+Example output from a local run:
+
+```text
+backend,accuracy,total_retrieved_tokens,avg_retrieved_tokens,p95_latency_ms
+memanto_style_active_digest,1.0,282,56.4,0.0168
+episode_graph_baseline,0.4,261,52.2,0.0082
+recent_window_3,0.0,315,63,0.0007
+append_only_log,0.0,361,72.2,0.0004
+```
+
+## Reproducibility Notes
+
+- Python 3.10 or newer.
+- No third-party packages are required for the offline benchmark.
+- The source dataset is `dataset.json`.
+- Token counts are deterministic approximations based on word and punctuation
+  boundaries, so they are comparable across backends without requiring a model
+  tokenizer.
+- The benchmark does not call external APIs by default.

examples/benchmarks/policy_drift_memory/benchmark.py

@@ -0,0 +1,281 @@
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import re
+import statistics
+import time
+from collections import defaultdict
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+
+TOKEN_RE = re.compile(r"\w+|[^\w\s]", re.UNICODE)
+
+
+def approx_tokens(text: str) -> int:
+    return len(TOKEN_RE.findall(text))
+
+
+def p95(values: list[float]) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = max(0, math.ceil(0.95 * len(ordered)) - 1)
+    return ordered[index]
+
+
+def normalize(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip().lower()
+
+
+@dataclass
+class QueryResult:
+    answer: str
+    latency_ms: float
+    retrieved_tokens: int
+
+
+class Backend:
+    name = "backend"
+
+    def ingest(self, event: dict[str, Any]) -> None:
+        raise NotImplementedError
+
+    def answer(self, query: dict[str, Any]) -> QueryResult:
+        raise NotImplementedError
+
+
+@dataclass
+class AppendOnlyLogBackend(Backend):
+    name: str = "append_only_log"
+    events: dict[str, list[str]] = field(default_factory=lambda: defaultdict(list))
+
+    def ingest(self, event: dict[str, Any]) -> None:
+        self.events[event["entity"]].append(event["text"])
+
+    def answer(self, query: dict[str, Any]) -> QueryResult:
+        start = time.perf_counter()
+        answer = "\n".join(self.events[query["entity"]])
+        latency = (time.perf_counter() - start) * 1000
+        return QueryResult(answer, latency, approx_tokens(answer))
+
+
+@dataclass
+class RecentWindowBackend(Backend):
+    name: str = "recent_window_3"
+    events: dict[str, list[str]] = field(default_factory=lambda: defaultdict(list))
+    window: int = 3
+
+    def ingest(self, event: dict[str, Any]) -> None:
+        self.events[event["entity"]].append(event["text"])
+
+    def answer(self, query: dict[str, Any]) -> QueryResult:
+        start = time.perf_counter()
+        answer = "\n".join(self.events[query["entity"]][-self.window :])
+        latency = (time.perf_counter() - start) * 1000
+        return QueryResult(answer, latency, approx_tokens(answer))
+
+
+@dataclass
+class EpisodeGraphBaseline(Backend):
+    """Small dependency-free approximation of episode graph memory behavior.
+
+    The backend preserves all observed values by key. It retrieves by entity and
+    key relevance, so it is more selective than a raw log but still exposes
+    superseded values when preferences drift.
+    """
+
+    name: str = "episode_graph_baseline"
+    facts: dict[str, dict[str, list[tuple[str, str]]]] = field(
+        default_factory=lambda: defaultdict(lambda: defaultdict(list))
+    )
+
+    def ingest(self, event: dict[str, Any]) -> None:
+        for fact in event.get("facts", []):
+            self.facts[event["entity"]][fact["key"]].append(
+                (event["timestamp"], fact["value"])
+            )
+
+    def answer(self, query: dict[str, Any]) -> QueryResult:
+        start = time.perf_counter()
+        question = normalize(query["question"])
+        rows: list[str] = []
+        for key, values in self.facts[query["entity"]].items():
+            key_terms = key.replace("_", " ")
+            if any(term in question for term in key_terms.split()):
+                rendered = " | ".join(f"{ts}: {value}" for ts, value in values)
+                rows.append(f"{key}: {rendered}")
+        if not rows:
+            for key, values in self.facts[query["entity"]].items():
+                rows.append(f"{key}: {values[-1][1]}")
+        answer = "\n".join(rows)
+        latency = (time.perf_counter() - start) * 1000
+        return QueryResult(answer, latency, approx_tokens(answer))
+
+
+@dataclass
+class ActiveDigestBackend(Backend):
+    """Memanto-style active memory digest for the offline control.
+
+    Each new typed fact replaces the current value for its key while retaining
+    evidence timestamps. Retrieval returns only the active state relevant to the
+    query entity. This mirrors the benchmarked Memanto property: current,
+    compact, conflict-aware memory instead of passive log replay.
+    """
+
+    name: str = "memanto_style_active_digest"
+    current: dict[str, dict[str, tuple[str, str]]] = field(
+        default_factory=lambda: defaultdict(dict)
+    )
+
+    def ingest(self, event: dict[str, Any]) -> None:
+        for fact in event.get("facts", []):
+            self.current[event["entity"]][fact["key"]] = (
+                event["timestamp"],
+                fact["value"],
+            )
+
+    def answer(self, query: dict[str, Any]) -> QueryResult:
+        start = time.perf_counter()
+        question = normalize(query["question"])
+        rows: list[str] = []
+        for key, (timestamp, value) in self.current[query["entity"]].items():
+            key_terms = key.replace("_", " ")
+            value_terms = normalize(value)
+            if (
+                any(term in question for term in key_terms.split())
+                or any(term in question for term in value_terms.split())
+                or key in {"pii_policy", "report_privacy", "launch_train"}
+            ):
+                rows.append(f"{key}: {value} (current since {timestamp})")
+        if not rows:
+            rows = [
+                f"{key}: {value} (current since {timestamp})"
+                for key, (timestamp, value) in self.current[query["entity"]].items()
+            ]
+        answer = "\n".join(rows)
+        latency = (time.perf_counter() - start) * 1000
+        return QueryResult(answer, latency, approx_tokens(answer))
+
+
+def score_answer(answer: str, query: dict[str, Any]) -> tuple[float, list[str]]:
+    lowered = normalize(answer)
+    missing = [item for item in query["required"] if normalize(item) not in lowered]
+    stale = [item for item in query["forbidden"] if normalize(item) in lowered]
+    if not missing and not stale:
+        return 1.0, []
+    penalties = [f"missing={item}" for item in missing]
+    penalties.extend(f"stale={item}" for item in stale)
+    return 0.0, penalties
+
+
+def build_backends() -> list[Backend]:
+    return [
+        ActiveDigestBackend(),
+        EpisodeGraphBaseline(),
+        RecentWindowBackend(),
+        AppendOnlyLogBackend(),
+    ]
+
+
+def run_once(dataset: dict[str, Any], repeats: int) -> dict[str, Any]:
+    backends = build_backends()
+    for event in dataset["events"]:
+        for backend in backends:
+            backend.ingest(event)
+
+    summary: list[dict[str, Any]] = []
+    details: list[dict[str, Any]] = []
+    for backend in backends:
+        scores: list[float] = []
+        latencies: list[float] = []
+        tokens: list[int] = []
+        for query in dataset["queries"]:
+            result: QueryResult | None = None
+            for _ in range(repeats):
+                result = backend.answer(query)
+                latencies.append(result.latency_ms)
+            assert result is not None
+            score, failures = score_answer(result.answer, query)
+            scores.append(score)
+            tokens.append(result.retrieved_tokens)
+            details.append(
+                {
+                    "backend": backend.name,
+                    "entity": query["entity"],
+                    "question": query["question"],
+                    "score": score,
+                    "retrieved_tokens": result.retrieved_tokens,
+                    "failures": failures,
+                    "answer": result.answer,
+                }
+            )
+        summary.append(
+            {
+                "backend": backend.name,
+                "accuracy": round(statistics.mean(scores), 4),
+                "total_retrieved_tokens": sum(tokens),
+                "avg_retrieved_tokens": round(statistics.mean(tokens), 2),
+                "p95_latency_ms": round(p95(latencies), 4),
+            }
+        )
+
+    return {
+        "dataset": dataset["name"],
+        "query_count": len(dataset["queries"]),
+        "repeat_count": repeats,
+        "summary": summary,
+        "details": details,
+    }
+
+
+def main() -> None:
+    def positive_int(value: str) -> int:
+        parsed = int(value)
+        if parsed < 1:
+            raise argparse.ArgumentTypeError("--repeats must be >= 1")
+        return parsed
+
+    parser = argparse.ArgumentParser(
+        description="Run the policy drift memory benchmark."
+    )
+    parser.add_argument(
+        "--dataset",
+        default=str(Path(__file__).with_name("dataset.json")),
+        help="Path to the benchmark dataset JSON.",
+    )
+    parser.add_argument(
+        "--repeats",
+        type=positive_int,
+        default=200,
+        help="Latency repeats per query. Default: 200.",
+    )
+    parser.add_argument(
+        "--output-json",
+        default="",
+        help="Optional path for the full JSON result.",
+    )
+    args = parser.parse_args()
+
+    dataset = json.loads(Path(args.dataset).read_text(encoding="utf-8"))
+    result = run_once(dataset, args.repeats)
+
+    print("backend,accuracy,total_retrieved_tokens,avg_retrieved_tokens,p95_latency_ms")
+    for row in result["summary"]:
+        print(
+            f"{row['backend']},{row['accuracy']},"
+            f"{row['total_retrieved_tokens']},{row['avg_retrieved_tokens']},"
+            f"{row['p95_latency_ms']}"
+        )
+
+    if args.output_json:
+        Path(args.output_json).write_text(
+            json.dumps(result, indent=2), encoding="utf-8"
+        )
+
+
+if __name__ == "__main__":
+    main()