diff --git a/src/mnemosyne/__init__.py b/src/mnemosyne/__init__.py
new file mode 100644
index 0000000..6c6e2b8
--- /dev/null
+++ b/src/mnemosyne/__init__.py
@@ -0,0 +1,6 @@
+"""Mnemosyne — object-addressed context memory for LLM agents.
+
+Built on Pichay's demand paging foundation, extended with multi-fidelity
+compression, semantic object segmentation, and a helper LLM for goal-aware
+retrieval. Named after the Greek Titan of memory.
+"""
diff --git a/src/mnemosyne/__main__.py b/src/mnemosyne/__main__.py
new file mode 100644
index 0000000..9ff5313
--- /dev/null
+++ b/src/mnemosyne/__main__.py
@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""Pichay experiment runner.
+
+Single entry point: starts the proxy, launches Claude Code through it,
+captures all artifacts when done.
+
+Usage:
+ # Baseline run (observe only)
+ python -m pichay --treatment baseline --project /path/to/project --prompt "Build X"
+
+ # Compact run (dead tool eviction)
+ python -m pichay --treatment compact --compact --project /path/to/project --prompt "Build X"
+
+ # With temperature control
+ python -m pichay --treatment compact --compact --temperature 0 --project /path/to/project --prompt "Build X"
+
+Artifacts are saved to experiments/{treatment}_run{N}/ in the pichay project directory.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Transitional: proxy.py moved to deprecated/. Gateway is the primary path
+# but the experiment runner still needs Flask until fully ported.
+from mnemosyne.deprecated.proxy import create_app, find_free_port
+
+
+def find_project_claude_dir(project_dir: str) -> Path:
+ """Map a project directory to its ~/.claude/projects/ path."""
+ normalized = project_dir.replace("/", "-")
+ return Path.home() / ".claude" / "projects" / normalized
+
+
+# Environment variables that cause Claude Code to detect nested invocation.
+_CLAUDE_NESTED_VARS = [
+ "CLAUDECODE",
+ "CLAUDE_CODE_ENTRYPOINT",
+ "CLAUDE_CODE_SSE_PORT",
+ "CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS",
+]
+
+
+def clean_env_for_subprocess() -> dict[str, str]:
+ """Build environment for nested Claude Code, removing detection vars."""
+ env = os.environ.copy()
+ for var in _CLAUDE_NESTED_VARS:
+ env.pop(var, None)
+ return env
+
+
+def run_experiment(args: argparse.Namespace) -> None:
+ """Run a single experiment."""
+ pichay_dir = Path(__file__).resolve().parent.parent.parent
+ exp_dir = pichay_dir / "experiments" / f"{args.treatment}_run{args.run}"
+ exp_dir.mkdir(parents=True, exist_ok=True)
+ log_dir = exp_dir / "logs"
+ log_dir.mkdir(exist_ok=True)
+
+ project_dir = Path(args.project).resolve()
+ if not project_dir.is_dir():
+ print(f"Error: project directory not found: {project_dir}", file=sys.stderr)
+ sys.exit(1)
+
+ print("=" * 60, file=sys.stderr)
+ print(f"Pichay Experiment", file=sys.stderr)
+ print(f" Treatment: {args.treatment}", file=sys.stderr)
+ print(f" Run: {args.run}", file=sys.stderr)
+ print(f" Project: {project_dir}", file=sys.stderr)
+ print(f" Compact: {args.compact}", file=sys.stderr)
+ print(f" Trim: {args.trim}", file=sys.stderr)
+ print(f" Temperature: {args.temperature}", file=sys.stderr)
+ print(f" Output: {exp_dir}", file=sys.stderr)
+ print("=" * 60, file=sys.stderr)
+
+ # Save config
+ config = {
+ "treatment": args.treatment,
+ "run": args.run,
+ "project_dir": str(project_dir),
+ "branch": args.branch,
+ "compact": args.compact,
+ "trim": args.trim,
+ "age_threshold": args.age_threshold,
+ "min_size": args.min_size,
+ "temperature": args.temperature,
+ "prompt": args.prompt,
+ "started_at": datetime.now(timezone.utc).isoformat(),
+ }
+ (exp_dir / "config.json").write_text(json.dumps(config, indent=2))
+
+ # Step 1: Reset project
+ if args.branch:
+ print(f"\n[1/5] Resetting project to {args.branch}...", file=sys.stderr)
+ subprocess.run(
+ ["git", "checkout", args.branch],
+ cwd=project_dir, capture_output=True,
+ )
+ subprocess.run(
+ ["git", "clean", "-fd"],
+ cwd=project_dir, capture_output=True,
+ )
+ else:
+ print(f"\n[1/5] No branch specified, using current state.", file=sys.stderr)
+
+ # Clear previous session data
+ claude_dir = find_project_claude_dir(str(project_dir))
+ if args.clear_session and claude_dir.is_dir():
+ for jsonl in claude_dir.glob("*.jsonl"):
+ jsonl.unlink()
+ print(f" Cleared session data in {claude_dir}", file=sys.stderr)
+
+ # Step 2: Start proxy
+ print(f"\n[2/5] Starting proxy...", file=sys.stderr)
+ port = find_free_port()
+
+ create_kwargs = dict(
+ log_dir=log_dir,
+ compact=args.compact,
+ trim=args.trim,
+ age_threshold=args.age_threshold,
+ min_size=args.min_size,
+ )
+ if args.upstream:
+ create_kwargs["upstream"] = args.upstream
+ app = create_app(**create_kwargs)
+ if args.temperature is not None:
+ app.config["temperature_override"] = args.temperature
+
+ # Run Flask in a thread
+ server_thread = threading.Thread(
+ target=lambda: app.run(
+ host="127.0.0.1", port=port, threaded=True, use_reloader=False,
+ ),
+ daemon=True,
+ )
+ server_thread.start()
+ time.sleep(1) # Let server bind
+ print(f" Proxy on http://localhost:{port}", file=sys.stderr)
+
+ # Step 3: Run Claude Code
+ print(f"\n[3/5] Launching Claude Code...", file=sys.stderr)
+ print(f" Prompt: {args.prompt[:80]}...", file=sys.stderr)
+ print(file=sys.stderr)
+
+ env = clean_env_for_subprocess()
+ env["ANTHROPIC_BASE_URL"] = f"http://localhost:{port}"
+
+ claude_cmd = [
+ "claude",
+ "-p",
+ "--dangerously-skip-permissions",
+ "--max-budget-usd", str(args.max_budget),
+ ]
+ if args.system_prompt is not None:
+ claude_cmd.extend(["--system-prompt", args.system_prompt])
+ claude_cmd.append(args.prompt)
+ try:
+ subprocess.run(
+ claude_cmd,
+ cwd=project_dir,
+ env=env,
+ stdin=subprocess.DEVNULL,
+ )
+ except KeyboardInterrupt:
+ print("\n Interrupted by user.", file=sys.stderr)
+ except FileNotFoundError:
+ print(" Error: 'claude' not found in PATH.", file=sys.stderr)
+ sys.exit(1)
+
+ # Step 4: Capture artifacts
+ print(f"\n[4/5] Capturing artifacts...", file=sys.stderr)
+
+ # Copy session data
+ if claude_dir.is_dir():
+ session_dest = exp_dir / "session"
+ if session_dest.exists():
+ shutil.rmtree(session_dest)
+ shutil.copytree(claude_dir, session_dest, dirs_exist_ok=True)
+
+ # Git state
+ subprocess.run(
+ ["git", "log", "--oneline", "-20"],
+ cwd=project_dir,
+ stdout=open(exp_dir / "git_log.txt", "w"),
+ stderr=subprocess.DEVNULL,
+ )
+ if args.branch:
+ subprocess.run(
+ ["git", "diff", f"{args.branch}..HEAD"],
+ cwd=project_dir,
+ stdout=open(exp_dir / "git_diff.txt", "w"),
+ stderr=subprocess.DEVNULL,
+ )
+ subprocess.run(
+ ["git", "diff", "--stat", f"{args.branch}..HEAD"],
+ cwd=project_dir,
+ stdout=open(exp_dir / "git_diff_stat.txt", "w"),
+ stderr=subprocess.DEVNULL,
+ )
+
+ # Test results
+ pyproject = project_dir / "pyproject.toml"
+ if pyproject.exists():
+ subprocess.run(
+ ["uv", "run", "pytest", "tests/", "-v", "--tb=short"],
+ cwd=project_dir,
+ stdout=open(exp_dir / "test_results.txt", "w"),
+ stderr=subprocess.STDOUT,
+ )
+
+ # Record end time
+ config["ended_at"] = datetime.now(timezone.utc).isoformat()
+ (exp_dir / "config.json").write_text(json.dumps(config, indent=2))
+
+ # Step 5: Summary
+ print(f"\n[5/5] Run complete.", file=sys.stderr)
+ print(f" Artifacts: {exp_dir}", file=sys.stderr)
+
+ # Run eval if proxy log exists
+ proxy_logs = list(log_dir.glob("proxy_*.jsonl"))
+ if proxy_logs:
+ from mnemosyne.eval import analyze_run, print_run_summary
+ summary = analyze_run(proxy_logs[0], label=args.treatment)
+ print_run_summary(summary)
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description="Pichay — context paging experiment runner"
+ )
+ parser.add_argument(
+ "--treatment", required=True,
+ help="Treatment label (e.g., baseline, compact, trimmed)",
+ )
+ parser.add_argument(
+ "--run", type=int, default=1,
+ help="Run number (default: 1)",
+ )
+ parser.add_argument(
+ "--project", required=True,
+ help="Target project directory",
+ )
+ parser.add_argument(
+ "--prompt", required=True,
+ help="Starting prompt for Claude Code",
+ )
+ parser.add_argument(
+ "--branch", default=None,
+ help="Git branch to reset to before each run",
+ )
+ parser.add_argument(
+ "--compact", action="store_true",
+ help="Enable dead tool result eviction",
+ )
+ parser.add_argument(
+ "--trim", action="store_true",
+ help="Enable system prompt trimming (tool stubs, skill dedup)",
+ )
+ parser.add_argument(
+ "--age-threshold", type=int, default=4,
+ help="Eviction age threshold in user-turns (default: 4)",
+ )
+ parser.add_argument(
+ "--min-size", type=int, default=500,
+ help="Min tool result size for eviction (default: 500)",
+ )
+ parser.add_argument(
+ "--temperature", type=float, default=None,
+ help="Override temperature (e.g., 0 for deterministic)",
+ )
+ parser.add_argument(
+ "--max-budget", type=float, default=20.0,
+ help="Maximum dollar budget for Claude Code session (default: $20)",
+ )
+ parser.add_argument(
+ "--clear-session", action="store_true", default=True,
+ help="Clear previous Claude session data (default: yes)",
+ )
+ parser.add_argument(
+ "--no-clear-session", action="store_false", dest="clear_session",
+ help="Keep previous Claude session data",
+ )
+ parser.add_argument(
+ "--system-prompt", default=None,
+ help="Override Claude Code's system prompt (--system-prompt flag)",
+ )
+ parser.add_argument(
+ "--upstream", default=None,
+ help="Upstream API base URL (default: https://api.anthropic.com). "
+ "Use for OpenRouter, Kimi, or any Anthropic-compatible endpoint.",
+ )
+ args = parser.parse_args()
+ run_experiment(args)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/src/mnemosyne/blocks.py b/src/mnemosyne/blocks.py
new file mode 100644
index 0000000..05ec772
--- /dev/null
+++ b/src/mnemosyne/blocks.py
@@ -0,0 +1,377 @@
+"""Conversation block store: content-addressed KV for message blocks.
+
+Assigns short stable IDs to conversation message blocks based on
+content hashing. The model sees these labels and can reference them
+in cleanup operations.
+
+Phase 1: labeling — inject [block:xxxx] markers into message content.
+Phase 2: cleanup — drop, summarize, anchor blocks via inline tags.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass
+class BlockEntry:
+ """A tracked conversation block."""
+ block_id: str # Short hex ID (first 8 chars of content hash)
+ content_hash: str # Full SHA-256 of content
+ size: int # Byte size of original content
+ turn: int # Turn when first seen
+ role: str # "user" or "assistant"
+ preview: str # First 80 chars for logging
+ status: str = "resident" # resident | anchored | summarized | dropped
+ original_content: str | None = None # Full content for fault restoration
+ summary: str | None = None # Model-authored summary (if summarized)
+
+
+class BlockStore:
+ """Per-session content-addressed store for conversation blocks.
+
+ Maps content hashes to BlockEntries. Assigns short IDs for model
+ reference. Detects content changes (re-labels gracefully).
+ """
+
+ def __init__(self):
+ self._by_id: dict[str, BlockEntry] = {}
+ self._by_hash: dict[str, str] = {} # content_hash → block_id
+
+ _LABEL_RE = re.compile(r"^\[(?:tensor|block):([0-9a-f]+)[ (]")
+
+ def _has_our_label(self, text: str) -> bool:
+ """Check if text starts with a label WE placed.
+
+ Validates the ID against known block IDs. Foreign labels
+ (injected via file contents, tool results, or crafted messages)
+ will have unrecognized IDs and return False, preventing an
+ attacker from skipping labeling or spoofing block references.
+ """
+ m = self._LABEL_RE.match(text)
+ if m is None:
+ return False
+ return m.group(1) in self._by_id
+
+ def label_messages(self, messages: list[dict], current_turn: int) -> None:
+ """Inject [block:xxxx] labels into message content.
+
+ Modifies messages in-place. Each message gets a label based on
+ its content hash. Labels are stable across turns as long as
+ the content doesn't change.
+
+ Only labels user and assistant text messages. Tool_use and
+ tool_result blocks are managed by the PageStore, not here.
+ """
+ for msg in messages:
+ role = msg.get("role", "")
+ if role not in ("user", "assistant"):
+ continue
+
+ content = msg.get("content", "")
+
+ # String content (simple user messages)
+ if isinstance(content, str):
+ # Skip if already labeled by us (validated against known IDs)
+ if self._has_our_label(content):
+ continue
+ # Skip very short messages (not worth labeling)
+ if len(content) < 200:
+ continue
+
+ entry = self._get_or_create(content, role, current_turn)
+ if entry and entry.status == "resident":
+ size_k = entry.size / 1024
+ msg["content"] = f"[tensor:{entry.block_id} ({size_k:.1f}KB)]\n{content}"
+
+ # List content (structured blocks)
+ elif isinstance(content, list):
+ for block in content:
+ if not isinstance(block, dict):
+ continue
+ if block.get("type") != "text":
+ continue
+ text = block.get("text", "")
+ # Skip if already labeled by us (validated against known IDs)
+ if self._has_our_label(text):
+ continue
+ # Skip short blocks
+ if len(text) < 200:
+ continue
+
+ entry = self._get_or_create(text, role, current_turn)
+ if entry and entry.status == "resident":
+ size_k = entry.size / 1024
+ block["text"] = f"[tensor:{entry.block_id} ({size_k:.1f}KB)]\n{text}"
+
+ def _get_or_create(self, content: str, role: str,
+ turn: int) -> BlockEntry | None:
+ """Get existing entry by content hash, or create a new one."""
+ content_hash = hashlib.sha256(content.encode()).hexdigest()
+ short_id = content_hash[:8]
+
+ # Already tracked
+ if content_hash in self._by_hash:
+ return self._by_id[self._by_hash[content_hash]]
+
+ # Check for ID collision (different content, same short hash)
+ if short_id in self._by_id:
+ existing = self._by_id[short_id]
+ if existing.content_hash != content_hash:
+ # Collision — extend the ID
+ short_id = content_hash[:12]
+ if short_id in self._by_id:
+ return None # Extremely unlikely, skip labeling
+
+ preview = content[:80].replace("\n", " ")
+ entry = BlockEntry(
+ block_id=short_id,
+ content_hash=content_hash,
+ size=len(content.encode()),
+ turn=turn,
+ role=role,
+ preview=preview,
+ original_content=content,
+ )
+ self._by_id[short_id] = entry
+ self._by_hash[content_hash] = short_id
+ return entry
+
+ def get(self, block_id: str) -> BlockEntry | None:
+ """Look up a block by its short ID."""
+ return self._by_id.get(block_id)
+
+ def restore(self, block_id: str) -> str | None:
+ """Restore a compressed block's original content."""
+ entry = self._by_id.get(block_id)
+ if entry and entry.original_content:
+ entry.status = "resident"
+ return entry.original_content
+ return None
+
+ # --- Phase 2: Cleanup operations ---
+
+ def drop(self, block_id: str) -> bool:
+ """Mark a block as dropped. Content stays for audit but won't
+ appear in future messages."""
+ entry = self._by_id.get(block_id)
+ if not entry:
+ return False
+ entry.status = "dropped"
+ return True
+
+ def summarize(self, block_id: str, summary: str) -> bool:
+ """Replace a block's content with a model-authored summary."""
+ entry = self._by_id.get(block_id)
+ if not entry:
+ return False
+ entry.status = "summarized"
+ entry.summary = summary
+ return True
+
+ def anchor(self, block_id: str) -> bool:
+ """Mark a block for retention — hint to keep in working memory."""
+ entry = self._by_id.get(block_id)
+ if not entry:
+ return False
+ entry.status = "anchored"
+ return True
+
+ def collapse_range(self, start_turn: int, end_turn: int,
+ summary: str) -> list[str]:
+ """Replace all blocks in a turn range with a summary marker.
+
+ Marks all resident/anchored blocks in [start_turn, end_turn] as
+ dropped, then creates a synthetic summary block covering the range.
+ Returns list of block IDs that were collapsed.
+
+ The synthetic block gets a deterministic ID from the range + summary
+ so repeated collapse of the same range is idempotent.
+ """
+ collapsed_ids = []
+ for entry in self._by_id.values():
+ if (start_turn <= entry.turn <= end_turn
+ and entry.status in ("resident", "anchored")):
+ entry.status = "dropped"
+ collapsed_ids.append(entry.block_id)
+
+ if not collapsed_ids:
+ return []
+
+ # Create a synthetic summary block for the range
+ synthetic_content = (
+ f"[Turns {start_turn}-{end_turn} collapsed: {summary}]"
+ )
+ content_hash = hashlib.sha256(synthetic_content.encode()).hexdigest()
+ short_id = content_hash[:8]
+
+ # Avoid collision with existing blocks
+ if short_id in self._by_id:
+ short_id = content_hash[:12]
+
+ if short_id not in self._by_id:
+ entry = BlockEntry(
+ block_id=short_id,
+ content_hash=content_hash,
+ size=len(synthetic_content.encode()),
+ turn=start_turn,
+ role="assistant",
+ preview=synthetic_content[:80],
+ status="summarized",
+ summary=summary,
+ )
+ self._by_id[short_id] = entry
+ self._by_hash[content_hash] = short_id
+
+ return collapsed_ids
+
+ _BLOCK_LABEL_RE = re.compile(r"^\[(?:tensor|block):([a-f0-9]{8,12})(?:\s*\([^)]*\))?\]\n?")
+
+ def apply_to_messages(self, messages: list[dict]) -> dict:
+ """Apply block status to messages — replace dropped/summarized content.
+
+ Modifies messages in-place. Returns stats dict.
+ """
+ stats = {"dropped": 0, "summarized": 0, "anchored": 0}
+
+ for msg in messages:
+ content = msg.get("content", "")
+
+ if isinstance(content, str):
+ msg["content"] = self._apply_to_text(content, msg, stats)
+
+ elif isinstance(content, list):
+ for block in content:
+ if not isinstance(block, dict) or block.get("type") != "text":
+ continue
+ text = block.get("text", "")
+ block["text"] = self._apply_to_text(text, msg, stats)
+
+ return stats
+
+ def _apply_to_text(self, text: str, msg: dict, stats: dict) -> str:
+ """Apply block status to a single text content."""
+ m = self._BLOCK_LABEL_RE.match(text)
+ if not m:
+ return text
+
+ block_id = m.group(1)
+ entry = self._by_id.get(block_id)
+ if not entry:
+ return text
+
+ if entry.status == "dropped":
+ stats["dropped"] += 1
+ turn_info = f"message {entry.turn} in session log"
+ return (
+ f"[...archived {entry.size:,} chars, {turn_info}...]"
+ )
+
+ if entry.status == "summarized" and entry.summary:
+ stats["summarized"] += 1
+ return (
+ f"[tensor:{block_id} — summarized, was {entry.size:,} chars]\n"
+ f"{entry.summary}"
+ )
+
+ # resident or anchored — no change
+ if entry.status == "anchored":
+ stats["anchored"] += 1
+
+ return text
+
+ @property
+ def block_count(self) -> int:
+ return len(self._by_id)
+
+ @property
+ def total_bytes(self) -> int:
+ return sum(e.size for e in self._by_id.values()
+ if e.status == "resident")
+
+ def large_blocks(self, min_size: int = 2000) -> list[BlockEntry]:
+ """Return resident blocks larger than min_size, sorted by size."""
+ return sorted(
+ [e for e in self._by_id.values()
+ if e.status == "resident" and e.size >= min_size],
+ key=lambda e: e.size,
+ reverse=True,
+ )
+
+ def summary(self) -> dict:
+ """Return a summary for the health endpoint."""
+ by_status = {}
+ for e in self._by_id.values():
+ by_status[e.status] = by_status.get(e.status, 0) + 1
+ return {
+ "total_blocks": len(self._by_id),
+ "total_bytes": self.total_bytes,
+ "by_status": by_status,
+ }
+
+ # --- Checkpoint / Restart ---
+
+ def checkpoint(self, path: Path) -> int:
+ """Serialize block state to JSON. Returns count of entries saved.
+
+ Writes atomically (tmp + rename) so a crash mid-write can't
+ corrupt the checkpoint. Only saves metadata and summaries —
+ original_content is NOT checkpointed (too large, and the
+ messages array is the source of truth for content).
+ """
+ entries = []
+ for entry in self._by_id.values():
+ entries.append({
+ "block_id": entry.block_id,
+ "content_hash": entry.content_hash,
+ "size": entry.size,
+ "turn": entry.turn,
+ "role": entry.role,
+ "preview": entry.preview,
+ "status": entry.status,
+ "summary": entry.summary,
+ })
+
+ tmp = path.with_suffix(".tmp")
+ tmp.write_text(json.dumps(entries, indent=2))
+ tmp.rename(path)
+ return len(entries)
+
+ @classmethod
+ def from_checkpoint(cls, path: Path) -> "BlockStore":
+ """Restore a BlockStore from a checkpoint file.
+
+ Returns a new BlockStore with all tracked entries restored.
+ Blocks that were resident at checkpoint time stay resident
+ but without original_content (will be re-populated when
+ label_messages sees the same content again).
+ """
+ store = cls()
+ if not path.is_file():
+ return store
+
+ try:
+ entries = json.loads(path.read_text())
+ except (json.JSONDecodeError, OSError):
+ return store
+
+ for rec in entries:
+ entry = BlockEntry(
+ block_id=rec["block_id"],
+ content_hash=rec["content_hash"],
+ size=rec["size"],
+ turn=rec["turn"],
+ role=rec["role"],
+ preview=rec["preview"],
+ status=rec.get("status", "resident"),
+ summary=rec.get("summary"),
+ original_content=None,
+ )
+ store._by_id[entry.block_id] = entry
+ store._by_hash[entry.content_hash] = entry.block_id
+
+ return store
diff --git a/src/mnemosyne/config.py b/src/mnemosyne/config.py
new file mode 100644
index 0000000..f3cb3fc
--- /dev/null
+++ b/src/mnemosyne/config.py
@@ -0,0 +1,176 @@
+"""Paging policy configuration.
+
+Defaults derived from corpus analysis (68 sessions, 36K turns,
+427T attention units):
+ - Median avg context: 95K tokens
+ - 85% of sessions hit 100K+
+ - 96.3% of attention cost comes from sessions >80K avg
+ - Compaction events drop 70-80% of context (amnesia, not curation)
+ - Fault cost at 80K: 6.4G attn, at 165K: 27.2G (4.2x more expensive)
+
+Three-tier policy:
+ - Below floor: no eviction needed, let context grow
+ - Advisory: inform the model, suggest curation (cooperative)
+ - Involuntary: pager evicts by policy (assertive)
+ - Hard cap: aggressive eviction, only pinned content survives
+
+These are starting points. Each session that runs through the gateway
+generates data that can refine them. The long-term path is
+crowdsourced calibration across instances.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+
+
+@dataclass(frozen=True)
+class PagingPolicy:
+ """Thresholds for context paging behavior.
+
+ All token counts refer to effective input tokens
+ (input + cache_creation + cache_read).
+ """
+
+ # Context window size (tokens)
+ window_size: int = 200_000
+
+ # Below this: no eviction, let context grow freely
+ # Always send compact memory stats (fill %, tokens, block count)
+ floor_tokens: int = 0
+
+ # Above this: advisory — inform model of largest blocks + cleanup ops
+ # 50% of window gives ~50% runway before involuntary
+ advisory_tokens: int = 100_000
+
+ # Above this: involuntary eviction by age/size policy
+ # 70% of window — act now or the system will
+ involuntary_tokens: int = 140_000
+
+ # Above this: aggressive eviction, survival requires pins
+ # 85% of window — last resort before context death
+ hard_cap_tokens: int = 170_000
+
+ # Eviction parameters
+ age_threshold: int = 4 # evict tool results older than N turns
+ min_evict_size: int = 500 # don't evict results smaller than N bytes
+
+ @property
+ def floor_pct(self) -> float:
+ return self.floor_tokens / self.window_size
+
+ @property
+ def advisory_pct(self) -> float:
+ return self.advisory_tokens / self.window_size
+
+ @property
+ def involuntary_pct(self) -> float:
+ return self.involuntary_tokens / self.window_size
+
+ @property
+ def hard_cap_pct(self) -> float:
+ return self.hard_cap_tokens / self.window_size
+
+ def zone(self, context_tokens: int) -> str:
+ """Return the policy zone for a given context size."""
+ if context_tokens >= self.hard_cap_tokens:
+ return "aggressive"
+ if context_tokens >= self.involuntary_tokens:
+ return "involuntary"
+ if context_tokens >= self.advisory_tokens:
+ return "advisory"
+ return "normal"
+
+ def to_dict(self) -> dict:
+ return asdict(self)
+
+
+# Module-level default
+_default = PagingPolicy()
+
+
+def get_policy() -> PagingPolicy:
+ """Return the current paging policy."""
+ return _default
+
+
+def set_policy(policy: PagingPolicy) -> None:
+ """Replace the current paging policy."""
+ global _default
+ _default = policy
+
+
+def load_policy(
+ window_size: int | None = None,
+ floor_tokens: int | None = None,
+ advisory_tokens: int | None = None,
+ involuntary_tokens: int | None = None,
+ hard_cap_tokens: int | None = None,
+ age_threshold: int | None = None,
+ min_evict_size: int | None = None,
+) -> PagingPolicy:
+ """Build a policy from explicit overrides, falling back to defaults.
+
+ Call with no arguments to get the default policy. Pass any subset
+ of parameters to override specific thresholds. This is the
+ integration point for CLI args, env vars, config files, or
+ eventually a database of crowdsourced values.
+ """
+ defaults = PagingPolicy()
+ policy = PagingPolicy(
+ window_size=window_size if window_size is not None else defaults.window_size,
+ floor_tokens=floor_tokens if floor_tokens is not None else defaults.floor_tokens,
+ advisory_tokens=advisory_tokens if advisory_tokens is not None else defaults.advisory_tokens,
+ involuntary_tokens=involuntary_tokens if involuntary_tokens is not None else defaults.involuntary_tokens,
+ hard_cap_tokens=hard_cap_tokens if hard_cap_tokens is not None else defaults.hard_cap_tokens,
+ age_threshold=age_threshold if age_threshold is not None else defaults.age_threshold,
+ min_evict_size=min_evict_size if min_evict_size is not None else defaults.min_evict_size,
+ )
+ set_policy(policy)
+ return policy
+
+
+def add_policy_args(parser) -> None:
+ """Add paging policy arguments to an argparse parser."""
+ group = parser.add_argument_group("paging policy")
+ group.add_argument(
+ "--window-size", type=int, default=None,
+ help="Context window size in tokens (default: 200000)",
+ )
+ group.add_argument(
+ "--floor-tokens", type=int, default=None,
+ help="Below this: no eviction (default: 0, always send stats)",
+ )
+ group.add_argument(
+ "--advisory-tokens", type=int, default=None,
+ help="Above this: suggest curation to model (default: 100000, 50%%)",
+ )
+ group.add_argument(
+ "--involuntary-tokens", type=int, default=None,
+ help="Above this: auto-evict by policy (default: 140000, 70%%)",
+ )
+ group.add_argument(
+ "--hard-cap-tokens", type=int, default=None,
+ help="Above this: aggressive eviction (default: 170000, 85%%)",
+ )
+ group.add_argument(
+ "--age-threshold", type=int, default=None,
+ help="Evict tool results older than N turns (default: 4)",
+ )
+ group.add_argument(
+ "--min-evict-size", type=int, default=None,
+ help="Don't evict results smaller than N bytes (default: 500)",
+ )
+
+
+def policy_from_args(args) -> PagingPolicy:
+ """Build a PagingPolicy from parsed argparse args."""
+ return load_policy(
+ window_size=getattr(args, "window_size", None),
+ floor_tokens=getattr(args, "floor_tokens", None),
+ advisory_tokens=getattr(args, "advisory_tokens", None),
+ involuntary_tokens=getattr(args, "involuntary_tokens", None),
+ hard_cap_tokens=getattr(args, "hard_cap_tokens", None),
+ age_threshold=getattr(args, "age_threshold", None),
+ min_evict_size=getattr(args, "min_evict_size", None),
+ )
diff --git a/src/mnemosyne/core/__init__.py b/src/mnemosyne/core/__init__.py
new file mode 100644
index 0000000..33c64bb
--- /dev/null
+++ b/src/mnemosyne/core/__init__.py
@@ -0,0 +1 @@
+"""Core gateway abstractions: canonical model + policy pipeline."""
diff --git a/src/mnemosyne/core/models.py b/src/mnemosyne/core/models.py
new file mode 100644
index 0000000..1963a5a
--- /dev/null
+++ b/src/mnemosyne/core/models.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class CanonicalMessage:
+ role: str
+ content: list[dict[str, Any]]
+ raw: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class CanonicalRequest:
+ provider: str
+ model: str
+ max_tokens: int | None
+ stream: bool
+ messages: list[CanonicalMessage]
+ tools: list[dict[str, Any]] = field(default_factory=list)
+ system: Any = None
+ extensions: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class PolicyAction:
+ stage: str
+ action: str
+ target_id: str
+ message_index: int
+ block_index: int
+ replacement_text: str | None
+ bytes: int
+ duplication_score: float
+
+
+@dataclass
+class PolicyContext:
+ protected_targets: set[str] = field(default_factory=set)
diff --git a/src/mnemosyne/core/pipeline.py b/src/mnemosyne/core/pipeline.py
new file mode 100644
index 0000000..c832cb6
--- /dev/null
+++ b/src/mnemosyne/core/pipeline.py
@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from mnemosyne.core.models import CanonicalRequest, PolicyContext
+from mnemosyne.core.policy import (
+ PolicyConfig,
+ apply_action,
+ paging_stage,
+ phantom_stage,
+ trim_stage,
+)
+
+
+PRECEDENCE = {"phantom": 3, "paging": 2, "trim": 1}
+
+
+class Pipeline:
+ def __init__(self, cfg: PolicyConfig, emit_event):
+ self.cfg = cfg
+ self.emit_event = emit_event
+
+ def run(self, req: CanonicalRequest) -> CanonicalRequest:
+ ctx = PolicyContext()
+ ctx, phantom_actions = phantom_stage(req, ctx)
+ paging_actions = paging_stage(req, ctx, self.cfg)
+ trim_actions = trim_stage(req, ctx, self.cfg)
+
+ # Phantom only marks protections in v1.
+ _ = phantom_actions
+
+ taken_targets: set[str] = set()
+ ordered = sorted(
+ paging_actions + trim_actions,
+ key=lambda a: PRECEDENCE.get(a.stage, 0),
+ reverse=True,
+ )
+
+ for action in ordered:
+ if action.target_id in ctx.protected_targets:
+ self.emit_event(
+ "policy_conflict_resolved",
+ winner_stage="phantom",
+ loser_stage=action.stage,
+ loser_action=action.action,
+ target_id=action.target_id,
+ target_bytes=action.bytes,
+ duplication_score=action.duplication_score,
+ resolution_reason="phantom_protection",
+ )
+ continue
+ if action.target_id in taken_targets:
+ continue
+ applied = apply_action(req, action)
+ if applied:
+ taken_targets.add(action.target_id)
+ self.emit_event(
+ "policy_action_applied",
+ stage=action.stage,
+ action=action.action,
+ target_id=action.target_id,
+ target_bytes=action.bytes,
+ duplication_score=action.duplication_score,
+ )
+
+ return req
diff --git a/src/mnemosyne/core/policy.py b/src/mnemosyne/core/policy.py
new file mode 100644
index 0000000..55c1a0a
--- /dev/null
+++ b/src/mnemosyne/core/policy.py
@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+from collections import Counter
+from dataclasses import dataclass
+from typing import Any
+
+from mnemosyne.core.models import CanonicalRequest, PolicyAction, PolicyContext
+from mnemosyne.core.utils import content_bytes
+
+
+@dataclass
+class PolicyConfig:
+ enable_paging: bool = True
+ enable_trim: bool = True
+ min_evict_size: int = 500
+
+
+@dataclass
+class BlockRef:
+ target_id: str
+ message_index: int
+ block_index: int
+ block: dict[str, Any]
+ size_bytes: int
+ text: str
+
+
+def _block_text(block: dict[str, Any]) -> str:
+ if isinstance(block.get("text"), str):
+ return block["text"]
+ if isinstance(block.get("content"), str):
+ return block["content"]
+ return ""
+
+
+def collect_blocks(req: CanonicalRequest) -> list[BlockRef]:
+ refs: list[BlockRef] = []
+ for mi, msg in enumerate(req.messages):
+ for bi, block in enumerate(msg.content):
+ text = _block_text(block)
+ refs.append(
+ BlockRef(
+ target_id=f"m{mi}:b{bi}",
+ message_index=mi,
+ block_index=bi,
+ block=block,
+ size_bytes=content_bytes(block),
+ text=text,
+ )
+ )
+ return refs
+
+
+def phantom_stage(req: CanonicalRequest, ctx: PolicyContext) -> tuple[PolicyContext, list[PolicyAction]]:
+ actions: list[PolicyAction] = []
+ for ref in collect_blocks(req):
+ if ref.block.get("pichay_phantom_protected") is True:
+ ctx.protected_targets.add(ref.target_id)
+ return ctx, actions
+
+
+def paging_stage(req: CanonicalRequest, ctx: PolicyContext, cfg: PolicyConfig) -> list[PolicyAction]:
+ if not cfg.enable_paging:
+ return []
+
+ refs = collect_blocks(req)
+ counts = Counter(r.text for r in refs if r.text)
+ actions: list[PolicyAction] = []
+
+ for ref in refs:
+ if ref.size_bytes < cfg.min_evict_size:
+ continue
+ btype = ref.block.get("type")
+ if btype not in {"tool_result", "text"}:
+ continue
+ if counts.get(ref.text, 0) < 2:
+ continue
+ actions.append(
+ PolicyAction(
+ stage="paging",
+ action="evict",
+ target_id=ref.target_id,
+ message_index=ref.message_index,
+ block_index=ref.block_index,
+ replacement_text=(
+ f"[Paged out duplicate block: {ref.size_bytes} bytes]"
+ ),
+ bytes=ref.size_bytes,
+ duplication_score=float(counts[ref.text]),
+ )
+ )
+ return actions
+
+
+def trim_stage(req: CanonicalRequest, ctx: PolicyContext, cfg: PolicyConfig) -> list[PolicyAction]:
+ if not cfg.enable_trim:
+ return []
+
+ refs = collect_blocks(req)
+ seen: set[tuple[str, str]] = set()
+ actions: list[PolicyAction] = []
+ for ref in refs:
+ key = (req.messages[ref.message_index].role, ref.text)
+ if not ref.text:
+ continue
+ if key in seen:
+ actions.append(
+ PolicyAction(
+ stage="trim",
+ action="trim_duplicate",
+ target_id=ref.target_id,
+ message_index=ref.message_index,
+ block_index=ref.block_index,
+ replacement_text="[Trimmed duplicate block]",
+ bytes=ref.size_bytes,
+ duplication_score=1.0,
+ )
+ )
+ else:
+ seen.add(key)
+ return actions
+
+
+def apply_action(req: CanonicalRequest, action: PolicyAction) -> bool:
+ try:
+ msg = req.messages[action.message_index]
+ block = msg.content[action.block_index]
+ except (IndexError, KeyError):
+ return False
+
+ if action.replacement_text is None:
+ return False
+
+ block_type = block.get("type", "text")
+ if block_type == "tool_result":
+ # Anthropic tool_result blocks use `content`, not `text`.
+ block["type"] = block_type
+ block.pop("text", None)
+ block["content"] = action.replacement_text
+ return True
+ if block_type == "text":
+ block["type"] = block_type
+ block["text"] = action.replacement_text
+ return True
+
+ msg.content[action.block_index] = {
+ "type": "text",
+ "text": action.replacement_text,
+ }
+ return True
diff --git a/src/mnemosyne/core/utils.py b/src/mnemosyne/core/utils.py
new file mode 100644
index 0000000..48c971b
--- /dev/null
+++ b/src/mnemosyne/core/utils.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import re
+
+
+def parse_duration(value: str) -> int:
+ """Parse duration strings like 24h, 15m, 7d into seconds."""
+ m = re.fullmatch(r"\s*(\d+)\s*([smhd])\s*", value)
+ if not m:
+ raise ValueError(f"invalid duration: {value!r}")
+ n = int(m.group(1))
+ unit = m.group(2)
+ mult = {"s": 1, "m": 60, "h": 3600, "d": 86400}[unit]
+ return n * mult
+
+
+def content_bytes(content: object) -> int:
+ if isinstance(content, str):
+ return len(content.encode("utf-8"))
+ if isinstance(content, list):
+ return sum(content_bytes(x) for x in content)
+ if isinstance(content, dict):
+ return sum(content_bytes(v) for v in content.values())
+ return len(str(content).encode("utf-8"))
diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py
new file mode 100644
index 0000000..0428e08
--- /dev/null
+++ b/src/mnemosyne/gateway.py
@@ -0,0 +1,2222 @@
+"""Pichay gateway — cache-aware multi-provider proxy for LLM context management.
+
+Primary entry point. Supersedes deprecated/proxy.py (Flask).
+
+Architecture:
+ Claude Code → Gateway (FastAPI) → Provider (Anthropic, OpenAI)
+
+The gateway intercepts, transforms, and forwards API requests. Features:
+- Per-conversation session management with stable fingerprinting
+- Static system prompt injection (cache-friendly, no prefix mutation)
+- Dynamic status anchor (end-of-messages, after cache breakpoints)
+- Token cap enforcement with cache hit rate tracking
+- Policy pipeline: phantom protection → paging → trim
+- Multi-provider support via adapter pattern
+- Prometheus metrics + live HTML dashboard
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import concurrent.futures
+import copy
+import hashlib
+from contextlib import asynccontextmanager
+import json
+import os
+import socket
+import sys
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import httpx
+import uvicorn
+from fastapi import FastAPI, HTTPException, Request, Response
+from fastapi.responses import HTMLResponse, StreamingResponse
+
+from mnemosyne.blocks import BlockStore
+from mnemosyne.core.models import CanonicalRequest
+from mnemosyne.core.pipeline import Pipeline
+from mnemosyne.core.policy import PolicyConfig, collect_blocks
+from mnemosyne.core.utils import parse_duration
+from mnemosyne.launcher import LaunchSpec, launch
+from mnemosyne.message_ops import (
+ check_inbound_for_injected_tags,
+ get_system_prompt,
+ inject_system_status,
+ process_cleanup_tags,
+ PICHAY_STATUS_MARKER,
+)
+from mnemosyne.fidelity import FidelityLevel, FidelityManager, PressureZone, make_object
+from mnemosyne.object_store import ObjectStoreBackend
+from mnemosyne.pager import PageStore, compact_messages
+from mnemosyne.message_store import MessageStore
+from mnemosyne.providers import adapters
+from mnemosyne.telemetry import Telemetry
+
+
+# ANSI for stderr status lines
+_DIM = "\033[2m"
+_YELLOW = "\033[33m"
+_RED = "\033[31m"
+_RESET = "\033[0m"
+
+
+def _run_async(coro):
+ """Run an async coroutine from sync context within uvicorn.
+
+ When called from a sync function that's running inside an async
+ event loop (e.g. FastAPI route → sync _preprocess), we can't use
+ asyncio.run() (loop already running). Instead, dispatch to a
+ thread pool and run a fresh event loop there.
+ """
+ try:
+ asyncio.get_running_loop()
+ # We're inside an event loop — run in a thread with its own loop
+ with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+ future = pool.submit(asyncio.run, coro)
+ return future.result(timeout=10)
+ except RuntimeError:
+ # No event loop running — safe to use asyncio.run directly
+ return asyncio.run(coro)
+
+
+def find_free_port() -> int:
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(("127.0.0.1", 0))
+ return s.getsockname()[1]
+
+
+def _session_id(body: dict) -> str:
+ """Derive a stable session fingerprint from the conversation.
+
+ Uses the system prompt (stable across turns within a session) combined
+ with the first user message text (unique per conversation). Falls back
+ to first message hash if no system prompt is present.
+ """
+ parts: list[str] = []
+
+ # System prompt is the most stable identifier — same across all turns
+ system = body.get("system")
+ if isinstance(system, list):
+ # Extract text blocks only (ignore cache_control, etc.)
+ sys_text = "".join(
+ b.get("text", "") for b in system if isinstance(b, dict) and b.get("type") == "text"
+ )
+ if sys_text:
+ # Use first 500 chars — enough to distinguish sessions,
+ # avoids hashing megabytes of system prompt
+ parts.append(sys_text[:500])
+ elif isinstance(system, str) and system:
+ parts.append(system[:500])
+
+ # Add first user message text for uniqueness across concurrent sessions
+ # with the same system prompt
+ messages = body.get("messages", [])
+ for msg in messages:
+ if msg.get("role") == "user":
+ content = msg.get("content", "")
+ if isinstance(content, str):
+ parts.append(content[:200])
+ elif isinstance(content, list):
+ for block in content:
+ if isinstance(block, dict) and block.get("type") == "text":
+ parts.append(block.get("text", "")[:200])
+ break
+ break
+
+ fingerprint = (
+ "|".join(parts) if parts else json.dumps(messages[0], sort_keys=True) if messages else ""
+ )
+ return hashlib.sha256(fingerprint.encode()).hexdigest()[:8]
+
+
+def _duplication_score(req: CanonicalRequest) -> float:
+ texts = [b.text for b in collect_blocks(req) if b.text]
+ if not texts:
+ return 0.0
+ unique = len(set(texts))
+ return max(0.0, float(len(texts) - unique) / float(len(texts)))
+
+
+def _add_cache_control(msg: dict) -> None:
+ """Add cache_control marker to the last content block of a message."""
+ content = msg.get("content")
+ if isinstance(content, str):
+ msg["content"] = [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]
+ elif isinstance(content, list) and content:
+ last = content[-1]
+ if isinstance(last, dict):
+ last["cache_control"] = {"type": "ephemeral"}
+
+
+def _strip_all_cache_controls(payload: dict) -> None:
+ """Remove all existing cache_control markers from system and messages.
+
+ Must run before _place_cache_controls to prevent exceeding the 4-block
+ API limit. Claude Code places its own markers on the system prompt;
+ without stripping first we can end up with 5+.
+ """
+ system = payload.get("system")
+ if isinstance(system, list):
+ for block in system:
+ if isinstance(block, dict):
+ block.pop("cache_control", None)
+ for msg in payload.get("messages", []):
+ content = msg.get("content")
+ if isinstance(content, list):
+ for block in content:
+ if isinstance(block, dict):
+ block.pop("cache_control", None)
+
+
+def _place_cache_controls(payload: dict) -> None:
+ """Place up to 4 cache_control markers in the outbound payload.
+
+ Mutates payload in place (the ephemeral copy — never the physical store).
+ Strips all existing markers first, then places fresh ones at:
+ 1. System prompt (always stable)
+ 2. 25% through messages
+ 3. 75% through messages
+ 4. Second-to-last message (previous assistant response)
+ """
+ _strip_all_cache_controls(payload)
+
+ # 1. System prompt
+ system = payload.get("system")
+ if system is not None:
+ if isinstance(system, str):
+ payload["system"] = [
+ {"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}
+ ]
+ elif isinstance(system, list) and system:
+ last = system[-1]
+ if isinstance(last, dict):
+ last["cache_control"] = {"type": "ephemeral"}
+
+ messages = payload.get("messages", [])
+ n = len(messages)
+ if n < 2:
+ return
+
+ placed_indices: set[int] = set()
+
+ def _try_place(idx: int) -> None:
+ """Place cache_control at messages[idx] if valid (not last, not tool role)."""
+ if idx < 0 or idx >= n - 1:
+ return
+ if idx in placed_indices:
+ return
+ msg = messages[idx]
+ if msg.get("role") == "tool":
+ return
+ _add_cache_control(msg)
+ placed_indices.add(idx)
+
+ # 2. 25% of the way through
+ _try_place(n // 4)
+
+ # 3. 75% of the way through
+ _try_place(3 * n // 4)
+
+ # 4. Second-to-last message
+ _try_place(n - 2)
+
+
+def _copy_headers(headers: httpx.Headers) -> dict[str, str]:
+ dropped = {
+ "content-length",
+ "transfer-encoding",
+ "connection",
+ "content-encoding",
+ }
+ out: dict[str, str] = {}
+ for k, v in headers.items():
+ if k.lower() in dropped:
+ continue
+ out[k] = v
+ return out
+
+
+def _inspect_sse_chunk(
+ chunk: bytes,
+ *,
+ buffer: bytearray,
+ emit_event,
+ request_id: str,
+ session_id: str,
+ provider: str,
+ usage_accumulator: dict[str, Any] | None = None,
+) -> None:
+ """Best-effort SSE validation for telemetry; never mutates payload."""
+ buffer.extend(chunk)
+ while b"\n\n" in buffer:
+ raw_event, rest = buffer.split(b"\n\n", 1)
+ buffer[:] = rest
+ if not raw_event:
+ continue
+ try:
+ text = raw_event.decode("utf-8")
+ except UnicodeDecodeError as e:
+ emit_event(
+ "anomaly",
+ kind="malformed_stream_chunk",
+ request_id=request_id,
+ session_id=session_id,
+ provider=provider,
+ error=f"invalid_utf8:{e}",
+ )
+ continue
+
+ for line in text.splitlines():
+ if not line.startswith("data: "):
+ continue
+ payload = line[6:]
+ if payload == "[DONE]":
+ continue
+ try:
+ evt = json.loads(payload)
+ if usage_accumulator is not None and isinstance(evt, dict):
+ etype = evt.get("type")
+ if etype == "message_start":
+ msg = evt.get("message", {})
+ if isinstance(msg, dict):
+ usage = msg.get("usage", {})
+ if isinstance(usage, dict):
+ usage_accumulator.update(usage)
+ elif etype == "message_delta":
+ usage = evt.get("usage", {})
+ if isinstance(usage, dict):
+ usage_accumulator.update(usage)
+ except json.JSONDecodeError as e:
+ emit_event(
+ "anomaly",
+ kind="malformed_stream_chunk",
+ request_id=request_id,
+ session_id=session_id,
+ provider=provider,
+ error=f"invalid_json:{e}",
+ )
+
+
+def _dashboard_html() -> str:
+ return """
+
+Mnemosyne
+
+
+
+
+Mnemosyne
Context Memory Proxy
Context Reduction
+—
token savings
+Active Sessions
+0
connected clients
+Objects Stored
+0
in memory store
+Fidelity Degradations
+0
total downgrades
+
+
+
+Micro-Faults
+Attempts0
+Successes0
+Tokens Saved0
+Success Rate—
+Latency (ms)
+Average—
+Minimum—
+Maximum—
+P95—
+Sessions
+| Session | Requests | Objects | Fidelity | Ctx Reduction |
+Evictions | Faults | Fault Rate | Incoming | Outgoing |
|---|
+