feat: Tool investigation report + Mem0 local provider (#842 )

## Investigation Report - docs/tool-investigation-2026-04-15.md: Full report analyzing 414 tools from awesome-ai-tools. Top 5 recommendations with integration paths. - docs/plans/awesome-ai-tools-integration.md: Implementation tracking plan. ## Mem0 Local Provider (P1) - plugins/memory/mem0_local/: New ChromaDB-backed memory provider. No API key required - fully sovereign. Compatible tool schemas with cloud Mem0 (mem0_profile, mem0_search, mem0_conclude). - Pattern-based fact extraction from conversations. - Deterministic dedup via content hashing. - Circuit breaker for resilience. - tests/plugins/memory/test_mem0_local.py: Full test coverage. ## Issues Filed - #857: LightRAG integration (P2) - #858: n8n workflow orchestration (P3) - #859: RAGFlow document understanding (P4) - #860: tensorzero LLMOps evaluation (P3) Closes #842
Merge pull request 'feat(security): implement Vitalik's secure LLM patterns — privacy filter + confirmation daemon [resolves merge conflict]' (#830 ) from feat/vitalik-secure-llm-1776303263 into main
2026-04-15 23:04:41 -04:00 · 2026-04-16 01:36:58 +00:00 · 2026-04-16 01:35:49 +00:00 · 2026-04-16 01:35:24 +00:00 · 2026-04-16 01:35:21 +00:00 · 2026-04-16 01:35:18 +00:00
11 changed files with 2187 additions and 0 deletions
--- a/agent/privacy_filter.py
+++ b/agent/privacy_filter.py
@@ -0,0 +1,353 @@
+"""Privacy Filter — strip PII from context before remote API calls.
+
+Implements Vitalik's Pattern 2: "A local model can strip out private data
+before passing the query along to a remote LLM."
+
+When Hermes routes a request to a cloud provider (Anthropic, OpenRouter, etc.),
+this module sanitizes the message context to remove personally identifiable
+information before it leaves the user's machine.
+
+Threat model (from Vitalik's secure LLM architecture):
+- Privacy (other): Non-LLM data leakage via search queries, API calls
+- LLM accidents: LLM accidentally leaking private data in prompts
+- LLM jailbreaks: Remote content extracting private context
+
+Usage:
+    from agent.privacy_filter import PrivacyFilter, sanitize_messages
+
+    pf = PrivacyFilter()
+    safe_messages = pf.sanitize_messages(messages)
+    # safe_messages has PII replaced with [REDACTED] tokens
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass, field
+from enum import Enum, auto
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+class Sensitivity(Enum):
+    """Classification of content sensitivity."""
+    PUBLIC = auto()       # No PII detected
+    LOW = auto()          # Generic references (e.g., city names)
+    MEDIUM = auto()       # Personal identifiers (name, email, phone)
+    HIGH = auto()         # Secrets, keys, financial data, medical info
+    CRITICAL = auto()     # Crypto keys, passwords, SSN patterns
+
+
+@dataclass
+class RedactionReport:
+    """Summary of what was redacted from a message batch."""
+    total_messages: int = 0
+    redacted_messages: int = 0
+    redactions: List[Dict[str, Any]] = field(default_factory=list)
+    max_sensitivity: Sensitivity = Sensitivity.PUBLIC
+
+    @property
+    def had_redactions(self) -> bool:
+        return self.redacted_messages > 0
+
+    def summary(self) -> str:
+        if not self.had_redactions:
+            return "No PII detected — context is clean for remote query."
+        parts = [f"Redacted {self.redacted_messages}/{self.total_messages} messages:"]
+        for r in self.redactions[:10]:
+            parts.append(f"  - {r['type']}: {r['count']} occurrence(s)")
+        if len(self.redactions) > 10:
+            parts.append(f"  ... and {len(self.redactions) - 10} more types")
+        return "\n".join(parts)
+
+
+# =========================================================================
+# PII pattern definitions
+# =========================================================================
+
+# Each pattern is (compiled_regex, redaction_type, sensitivity_level, replacement)
+_PII_PATTERNS: List[Tuple[re.Pattern, str, Sensitivity, str]] = []
+
+
+def _compile_patterns() -> None:
+    """Compile PII detection patterns. Called once at module init."""
+    global _PII_PATTERNS
+    if _PII_PATTERNS:
+        return
+
+    raw_patterns = [
+        # --- CRITICAL: secrets and credentials ---
+        (
+            r'(?:api[_-]?key|apikey|secret[_-]?key|access[_-]?token)\s*[:=]\s*["\']?([A-Za-z0-9_\-\.]{20,})["\']?',
+            "api_key_or_token",
+            Sensitivity.CRITICAL,
+            "[REDACTED-API-KEY]",
+        ),
+        (
+            r'\b(?:sk-|sk_|pk_|rk_|ak_)[A-Za-z0-9]{20,}\b',
+            "prefixed_secret",
+            Sensitivity.CRITICAL,
+            "[REDACTED-SECRET]",
+        ),
+        (
+            r'\b(?:ghp_|gho_|ghu_|ghs_|ghr_)[A-Za-z0-9]{36,}\b',
+            "github_token",
+            Sensitivity.CRITICAL,
+            "[REDACTED-GITHUB-TOKEN]",
+        ),
+        (
+            r'\b(?:xox[bposa]-[A-Za-z0-9\-]+)\b',
+            "slack_token",
+            Sensitivity.CRITICAL,
+            "[REDACTED-SLACK-TOKEN]",
+        ),
+        (
+            r'(?:password|passwd|pwd)\s*[:=]\s*["\']?([^\s"\']{4,})["\']?',
+            "password",
+            Sensitivity.CRITICAL,
+            "[REDACTED-PASSWORD]",
+        ),
+        (
+            r'(?:-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----)',
+            "private_key_block",
+            Sensitivity.CRITICAL,
+            "[REDACTED-PRIVATE-KEY]",
+        ),
+        # Ethereum / crypto addresses (42-char hex starting with 0x)
+        (
+            r'\b0x[a-fA-F0-9]{40}\b',
+            "ethereum_address",
+            Sensitivity.HIGH,
+            "[REDACTED-ETH-ADDR]",
+        ),
+        # Bitcoin addresses (base58, 25-34 chars starting with 1/3/bc1)
+        (
+            r'\b[13][a-km-zA-HJ-NP-Z1-9]{25,34}\b',
+            "bitcoin_address",
+            Sensitivity.HIGH,
+            "[REDACTED-BTC-ADDR]",
+        ),
+        (
+            r'\bbc1[a-zA-HJ-NP-Z0-9]{39,59}\b',
+            "bech32_address",
+            Sensitivity.HIGH,
+            "[REDACTED-BTC-ADDR]",
+        ),
+        # --- HIGH: financial ---
+        (
+            r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
+            "credit_card_number",
+            Sensitivity.HIGH,
+            "[REDACTED-CC]",
+        ),
+        (
+            r'\b\d{3}-\d{2}-\d{4}\b',
+            "us_ssn",
+            Sensitivity.HIGH,
+            "[REDACTED-SSN]",
+        ),
+        # --- MEDIUM: personal identifiers ---
+        # Email addresses
+        (
+            r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b',
+            "email_address",
+            Sensitivity.MEDIUM,
+            "[REDACTED-EMAIL]",
+        ),
+        # Phone numbers (US/international patterns)
+        (
+            r'\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b',
+            "phone_number_us",
+            Sensitivity.MEDIUM,
+            "[REDACTED-PHONE]",
+        ),
+        (
+            r'\b\+\d{1,3}[-.\s]?\d{4,14}\b',
+            "phone_number_intl",
+            Sensitivity.MEDIUM,
+            "[REDACTED-PHONE]",
+        ),
+        # Filesystem paths that reveal user identity
+        (
+            r'(?:/Users/|/home/|C:\\Users\\)([A-Za-z0-9_\-]+)',
+            "user_home_path",
+            Sensitivity.MEDIUM,
+            r"/Users/[REDACTED-USER]",
+        ),
+        # --- LOW: environment / system info ---
+        # Internal IPs
+        (
+            r'\b(?:10\.\d{1,3}\.\d{1,3}\.\d{1,3}|172\.(?:1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3}|192\.168\.\d{1,3}\.\d{1,3})\b',
+            "internal_ip",
+            Sensitivity.LOW,
+            "[REDACTED-IP]",
+        ),
+    ]
+
+    _PII_PATTERNS = [
+        (re.compile(pattern, re.IGNORECASE), rtype, sensitivity, replacement)
+        for pattern, rtype, sensitivity, replacement in raw_patterns
+    ]
+
+
+_compile_patterns()
+
+
+# =========================================================================
+# Sensitive file path patterns (context-aware)
+# =========================================================================
+
+_SENSITIVE_PATH_PATTERNS = [
+    re.compile(r'\.(?:env|pem|key|p12|pfx|jks|keystore)\b', re.IGNORECASE),
+    re.compile(r'(?:\.ssh/|\.gnupg/|\.aws/|\.config/gcloud/)', re.IGNORECASE),
+    re.compile(r'(?:wallet|keystore|seed|mnemonic)', re.IGNORECASE),
+    re.compile(r'(?:\.hermes/\.env)', re.IGNORECASE),
+]
+
+
+def _classify_path_sensitivity(path: str) -> Sensitivity:
+    """Check if a file path references sensitive material."""
+    for pat in _SENSITIVE_PATH_PATTERNS:
+        if pat.search(path):
+            return Sensitivity.HIGH
+    return Sensitivity.PUBLIC
+
+
+# =========================================================================
+# Core filtering
+# =========================================================================
+
+class PrivacyFilter:
+    """Strip PII from message context before remote API calls.
+
+    Integrates with the agent's message pipeline. Call sanitize_messages()
+    before sending context to any cloud LLM provider.
+    """
+
+    def __init__(
+        self,
+        min_sensitivity: Sensitivity = Sensitivity.MEDIUM,
+        aggressive_mode: bool = False,
+    ):
+        """
+        Args:
+            min_sensitivity: Only redact PII at or above this level.
+                Default MEDIUM — redacts emails, phones, paths but not IPs.
+            aggressive_mode: If True, also redact file paths and internal IPs.
+        """
+        self.min_sensitivity = (
+            Sensitivity.LOW if aggressive_mode else min_sensitivity
+        )
+        self.aggressive_mode = aggressive_mode
+
+    def sanitize_text(self, text: str) -> Tuple[str, List[Dict[str, Any]]]:
+        """Sanitize a single text string. Returns (cleaned_text, redaction_list)."""
+        redactions = []
+        cleaned = text
+
+        for pattern, rtype, sensitivity, replacement in _PII_PATTERNS:
+            if sensitivity.value < self.min_sensitivity.value:
+                continue
+
+            matches = pattern.findall(cleaned)
+            if matches:
+                count = len(matches) if isinstance(matches[0], str) else sum(
+                    1 for m in matches if m
+                )
+                if count > 0:
+                    cleaned = pattern.sub(replacement, cleaned)
+                    redactions.append({
+                        "type": rtype,
+                        "sensitivity": sensitivity.name,
+                        "count": count,
+                    })
+
+        return cleaned, redactions
+
+    def sanitize_messages(
+        self, messages: List[Dict[str, Any]]
+    ) -> Tuple[List[Dict[str, Any]], RedactionReport]:
+        """Sanitize a list of OpenAI-format messages.
+
+        Returns (safe_messages, report). System messages are NOT sanitized
+        (they're typically static prompts). Only user and assistant messages
+        with string content are processed.
+
+        Args:
+            messages: List of {"role": ..., "content": ...} dicts.
+
+        Returns:
+            Tuple of (sanitized_messages, redaction_report).
+        """
+        report = RedactionReport(total_messages=len(messages))
+        safe_messages = []
+
+        for msg in messages:
+            role = msg.get("role", "")
+            content = msg.get("content", "")
+
+            # Only sanitize user/assistant string content
+            if role in ("user", "assistant") and isinstance(content, str) and content:
+                cleaned, redactions = self.sanitize_text(content)
+                if redactions:
+                    report.redacted_messages += 1
+                    report.redactions.extend(redactions)
+                    # Track max sensitivity
+                    for r in redactions:
+                        s = Sensitivity[r["sensitivity"]]
+                        if s.value > report.max_sensitivity.value:
+                            report.max_sensitivity = s
+                    safe_msg = {**msg, "content": cleaned}
+                    safe_messages.append(safe_msg)
+                    logger.info(
+                        "Privacy filter: redacted %d PII type(s) from %s message",
+                        len(redactions), role,
+                    )
+                else:
+                    safe_messages.append(msg)
+            else:
+                safe_messages.append(msg)
+
+        return safe_messages, report
+
+    def should_use_local_only(self, text: str) -> Tuple[bool, str]:
+        """Determine if content is too sensitive for any remote call.
+
+        Returns (should_block, reason). If True, the content should only
+        be processed by a local model.
+        """
+        _, redactions = self.sanitize_text(text)
+
+        critical_count = sum(
+            1 for r in redactions
+            if Sensitivity[r["sensitivity"]] == Sensitivity.CRITICAL
+        )
+        high_count = sum(
+            1 for r in redactions
+            if Sensitivity[r["sensitivity"]] == Sensitivity.HIGH
+        )
+
+        if critical_count > 0:
+            return True, f"Contains {critical_count} critical-secret pattern(s) — local-only"
+        if high_count >= 3:
+            return True, f"Contains {high_count} high-sensitivity pattern(s) — local-only"
+        return False, ""
+
+
+def sanitize_messages(
+    messages: List[Dict[str, Any]],
+    min_sensitivity: Sensitivity = Sensitivity.MEDIUM,
+    aggressive: bool = False,
+) -> Tuple[List[Dict[str, Any]], RedactionReport]:
+    """Convenience function: sanitize messages with default settings."""
+    pf = PrivacyFilter(min_sensitivity=min_sensitivity, aggressive_mode=aggressive)
+    return pf.sanitize_messages(messages)
+
+
+def quick_sanitize(text: str) -> str:
+    """Quick sanitize a single string — returns cleaned text only."""
+    pf = PrivacyFilter()
+    cleaned, _ = pf.sanitize_text(text)
+    return cleaned
--- a/docs/plans/awesome-ai-tools-integration.md
+++ b/docs/plans/awesome-ai-tools-integration.md
@@ -0,0 +1,44 @@
+# awesome-ai-tools Integration Plan
+
+**Tracking:** #842
+**Source report:** docs/tool-investigation-2026-04-15.md
+**Date:** 2026-04-16
+
+---
+
+## Status Dashboard
+
+| # | Tool | Category | Impact | Effort | Status | Issue |
+|---|------|----------|--------|--------|--------|-------|
+| 1 | Mem0 | Memory | 5/5 | 3/5 | Cloud + Local done | #842 |
+| 2 | LightRAG | RAG | 4/5 | 3/5 | Not started | #857 |
+| 3 | n8n | Orchestration | 5/5 | 4/5 | Not started | #858 |
+| 4 | RAGFlow | RAG | 4/5 | 4/5 | Not started | #859 |
+| 5 | tensorzero | LLMOps | 4/5 | 3/5 | Not started | #860 |
+
+---
+
+## #1: Mem0 — DONE
+
+Cloud: `plugins/memory/mem0/` (MEM0_API_KEY required)
+Local: `plugins/memory/mem0_local/` (ChromaDB, no API key)
+
+## #2: LightRAG (P2)
+
+Create `plugins/rag/lightrag/` plugin. Index skill docs. Use local Ollama embeddings.
+
+## #3: n8n (P3)
+
+Deploy as Docker service. Create workflow templates for Hermes patterns.
+
+## #4: RAGFlow (P4)
+
+Deploy as Docker service. Integrate via HTTP API for document understanding.
+
+## #5: tensorzero (P3)
+
+Evaluate as provider routing replacement. Canary migration (10% traffic first).
+
+---
+
+*Last updated: 2026-04-16*
--- a/docs/tool-investigation-2026-04-15.md
+++ b/docs/tool-investigation-2026-04-15.md
@@ -0,0 +1,151 @@
+## Tool Investigation Report: Top 5 Recommendations from awesome-ai-tools
+
+**Source:** [formatho/awesome-ai-tools](https://github.com/formatho/awesome-ai-tools)
+**Date:** 2026-04-15
+**Tools Analyzed:** 414 across 9 categories
+**Agent:** Timmy
+
+---
+
+## Analysis Summary
+
+Scanned 414 tools from the awesome-ai-tools repository. Evaluated each against Hermes integration potential across five categories: Memory/Context, Inference Optimization, Agent Orchestration, Workflow Automation, and Retrieval/RAG.
+
+### Evaluation Criteria
+- **Stars:** GitHub community validation (stability signal)
+- **Freshness:** Active development (Fresh = updated <=7 days)
+- **Integration Fit:** How well it complements Hermes' existing architecture (skills, memory, tools)
+- **Integration Effort:** 1 (trivial drop-in) to 5 (major refactor required)
+- **Impact:** 1 (incremental) to 5 (transformative)
+
+---
+
+## Top 5 Recommended Tools
+
+### #1: Mem0 — Universal Memory Layer for AI Agents
+
+| Metric | Value |
+|--------|-------|
+| **Category** | Memory/Context |
+| **GitHub** | [mem0ai/mem0](https://github.com/mem0ai/mem0) |
+| **Stars** | 53.1k |
+| **Freshness** | Fresh |
+| **Integration Effort** | 3/5 |
+| **Impact** | 5/5 |
+| **Hermes Status** | IMPLEMENTED (plugins/memory/mem0/) + LOCAL MODE (plugins/memory/mem0_local/) |
+
+**Why it fits Hermes:**
+Hermes currently has session_search (transcript recall) and memory (persistent facts), but lacks a unified memory layer that bridges sessions with semantic understanding. Mem0 provides exactly this: automatic memory extraction from conversations, deduplication, and cross-session retrieval with semantic search.
+
+**Integration path:**
+- Cloud: plugins/memory/mem0/ (requires MEM0_API_KEY)
+- Local: plugins/memory/mem0_local/ (ChromaDB-backed, no API key)
+- Auto-extract facts from session transcripts
+- Query before session_search for richer contextual recall
+
+**Key risk:** Mem0 is freemium — core is open-source but advanced features require paid tier. Local mode mitigates this entirely.
+
+---
+
+### #2: LightRAG — Simple and Fast Retrieval-Augmented Generation
+
+| Metric | Value |
+|--------|-------|
+| **Category** | Retrieval/RAG |
+| **GitHub** | [HKUDS/LightRAG](https://github.com/HKUDS/LightRAG) |
+| **Stars** | 33.1k |
+| **Freshness** | Fresh |
+| **Integration Effort** | 3/5 |
+| **Impact** | 4/5 |
+| **Hermes Status** | NOT IMPLEMENTED — Issue #857 |
+
+**Why it fits Hermes:**
+Hermes has 190+ skills but no unified knowledge retrieval system. LightRAG adds graph-based RAG that understands relationships between concepts, not just keyword matches. It's lightweight, runs locally, and has a simple API.
+
+**Integration path:**
+- LightRAG as a local knowledge base for skill references
+- Index GENOME.md files, README.md, and key codebase files
+- Use local Ollama models for embeddings
+- Complements existing search_files without replacing it
+
+---
+
+### #3: n8n — Workflow Automation Platform
+
+| Metric | Value |
+|--------|-------|
+| **Category** | Workflow Automation / Agent Orchestration |
+| **GitHub** | [n8n-io/n8n](https://github.com/n8n-io/n8n) |
+| **Stars** | 183.9k |
+| **Freshness** | Fresh |
+| **Integration Effort** | 4/5 |
+| **Impact** | 5/5 |
+| **Hermes Status** | NOT IMPLEMENTED — Issue #858 |
+
+**Why it fits Hermes:**
+n8n provides a self-hosted, fair-code workflow platform with 400+ integrations. Rather than replacing Hermes' agent loop, n8n sits above it: trigger Hermes agents from external events, chain multi-agent workflows, and visualize execution.
+
+---
+
+### #4: RAGFlow — Open-Source RAG Engine
+
+| Metric | Value |
+|--------|-------|
+| **Category** | Retrieval/RAG |
+| **GitHub** | [infiniflow/ragflow](https://github.com/infiniflow/ragflow) |
+| **Stars** | 77.9k |
+| **Freshness** | Fresh |
+| **Integration Effort** | 4/5 |
+| **Impact** | 4/5 |
+| **Hermes Status** | NOT IMPLEMENTED — Issue #859 |
+
+**Why it fits Hermes:**
+RAGFlow handles document parsing (PDF, Word, images via OCR), chunking, embedding, and retrieval with a web UI. Enables "document understanding" as a first-class capability.
+
+---
+
+### #5: tensorzero — LLMOps Platform
+
+| Metric | Value |
+|--------|-------|
+| **Category** | Inference Optimization / LLMOps |
+| **GitHub** | [tensorzero/tensorzero](https://github.com/tensorzero/tensorzero) |
+| **Stars** | 11.2k |
+| **Freshness** | Fresh |
+| **Integration Effort** | 3/5 |
+| **Impact** | 4/5 |
+| **Hermes Status** | NOT IMPLEMENTED — Issue #860 |
+
+**Why it fits Hermes:**
+TensorZero unifies LLM gateway, observability, evaluation, and optimization. Replaces custom provider routing with a maintained, battle-tested platform.
+
+---
+
+## Honorable Mentions
+
+| Tool | Stars | Category | Why Not Top 5 |
+|------|-------|----------|---------------|
+| memvid | 14.9k | Memory | Newer; Mem0 is more mature |
+| mempalace | 44.8k | Memory | Already evaluated; Mem0 has broader API |
+| Everything Claude Code | 154.3k | Agent | Too Claude-specific |
+| Portkey AI Gateway | 11.3k | Gateway | TensorZero is OSS; Portkey is freemium |
+
+---
+
+## Implementation Priority
+
+| Priority | Tool | Action | Status | Issue |
+|----------|------|--------|--------|-------|
+| P1 | Mem0 | Local-only mode (ChromaDB) | DONE | #842 |
+| P2 | LightRAG | Set up local instance, index skills | Not started | #857 |
+| P3 | tensorzero | Evaluate as provider routing | Not started | #860 |
+| P4 | RAGFlow | Deploy Docker, test docs | Not started | #859 |
+| P5 | n8n | Deploy for workflow viz | Not started | #858 |
+
+---
+
+## References
+- Source: https://github.com/formatho/awesome-ai-tools
+- Total tools: 414 across 9 categories
+- Last updated: April 16, 2026
+- Tracking issue: Timmy_Foundation/hermes-agent#842
--- a/plugins/memory/mem0_local/README.md
+++ b/plugins/memory/mem0_local/README.md
@@ -0,0 +1,60 @@
+# Mem0 Local - Sovereign Memory Provider
+
+Local-only memory provider using ChromaDB. No API key required - all data stays on your machine.
+
+## How It Differs from Cloud Mem0
+
+| Feature | Cloud Mem0 | Local Mem0 |
+|---------|-----------|------------|
+| API key | Required | Not needed |
+| Data location | Mem0 servers | Your machine |
+| Fact extraction | Server-side LLM | Pattern-based heuristics |
+| Reranking | Yes | No |
+| Cost | Freemium | Free forever |
+
+## Setup
+
+```bash
+pip install chromadb
+hermes config set memory.provider mem0-local
+```
+
+Or manually in ~/.hermes/config.yaml:
+```yaml
+memory:
+  provider: mem0-local
+```
+
+## Config
+
+Config file: $HERMES_HOME/mem0-local.json
+
+| Key | Default | Description |
+|-----|---------|-------------|
+| storage_path | ~/.hermes/mem0-local/ | ChromaDB storage directory |
+| collection_prefix | mem0 | Collection name prefix |
+| max_memories | 10000 | Maximum stored memories |
+
+## Tools
+
+Same interface as cloud Mem0:
+
+| Tool | Description |
+|------|-------------|
+| mem0_profile | All stored memories about the user |
+| mem0_search | Semantic search by meaning |
+| mem0_conclude | Store a fact verbatim |
+
+## Data Sovereignty
+
+All data is stored in $HERMES_HOME/mem0-local/ as a ChromaDB persistent database. No network calls are made.
+
+To back up: copy the mem0-local/ directory.
+To reset: delete the mem0-local/ directory.
+
+## Limitations
+
+- Fact extraction is pattern-based (not LLM-powered)
+- No reranking - results ranked by embedding similarity only
+- No cross-device sync (by design)
+- Requires chromadb pip dependency (~50MB)
--- a/plugins/memory/mem0_local/init.py
+++ b/plugins/memory/mem0_local/init.py
@@ -0,0 +1,381 @@
+"""Mem0 Local memory provider - ChromaDB-backed, no API key required.
+
+Sovereign deployment: all data stays on the user's machine. Uses ChromaDB
+for vector storage and simple heuristic fact extraction (no server-side LLM).
+
+Compatible tool schemas with the cloud Mem0 provider:
+  mem0_profile  - retrieve all stored memories
+  mem0_search   - semantic search by meaning
+  mem0_conclude - store a fact verbatim
+
+Config via $HERMES_HOME/mem0-local.json or environment variables:
+  MEM0_LOCAL_PATH  - storage directory (default: $HERMES_HOME/mem0-local/)
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import re
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
+
+logger = logging.getLogger(__name__)
+
+# Circuit breaker
+_BREAKER_THRESHOLD = 5
+_BREAKER_COOLDOWN_SECS = 120
+
+
+def _load_config() -> dict:
+    """Load local config from env vars, with $HERMES_HOME/mem0-local.json overrides."""
+    from hermes_constants import get_hermes_home
+
+    config = {
+        "storage_path": os.environ.get("MEM0_LOCAL_PATH", ""),
+        "collection_prefix": "mem0",
+        "max_memories": 10000,
+    }
+
+    config_path = get_hermes_home() / "mem0-local.json"
+    if config_path.exists():
+        try:
+            file_cfg = json.loads(config_path.read_text(encoding="utf-8"))
+            config.update({k: v for k, v in file_cfg.items()
+                           if v is not None and v != ""})
+        except Exception:
+            pass
+
+    if not config["storage_path"]:
+        config["storage_path"] = str(get_hermes_home() / "mem0-local")
+
+    return config
+
+
+# Simple fact extraction patterns (no LLM required)
+_FACT_PATTERNS = [
+    (r"(?:my|the user'?s?)\s+(?:name|username)\s+(?:is|=)\s+(.+?)(?:\.|$)", "user.name"),
+    (r"(?:i|user)\s+(?:prefer|like|use|want|need)s?\s+(.+?)(?:\.|$)", "preference"),
+    (r"(?:i|user)\s+(?:work|am)\s+(?:at|as|on|with)\s+(.+?)(?:\.|$)", "context"),
+    (r"(?:remember|note|save|store)[:\s]+(.+?)(?:\.|$)", "explicit"),
+    (r"(?:my|the)\s+(?:timezone|tz)\s+(?:is|=)\s+(.+?)(?:\.|$)", "user.timezone"),
+    (r"(?:my|the)\s+(?:project|repo|codebase)\s+(?:is|=|called)\s+(.+?)(?:\.|$)", "project"),
+    (r"(?:actually|correction|instead)[:\s]+(.+?)(?:\.|$)", "correction"),
+]
+
+
+def _extract_facts(text: str) -> List[Dict[str, str]]:
+    """Extract structured facts from conversation text using pattern matching."""
+    facts = []
+    if not text or len(text) < 10:
+        return facts
+    text_lower = text.lower().strip()
+
+    for pattern, category in _FACT_PATTERNS:
+        matches = re.findall(pattern, text_lower, re.IGNORECASE)
+        for match in matches:
+            fact_text = match.strip() if isinstance(match, str) else match[0].strip()
+            if len(fact_text) > 3 and len(fact_text) < 500:
+                facts.append({
+                    "content": fact_text,
+                    "category": category,
+                    "source_text": text[:200],
+                })
+
+    return facts
+
+
+# Tool schemas (compatible with cloud Mem0)
+PROFILE_SCHEMA = {
+    "name": "mem0_profile",
+    "description": (
+        "Retrieve all stored memories about the user - preferences, facts, "
+        "project context. Fast, no reranking. Use at conversation start."
+    ),
+    "parameters": {"type": "object", "properties": {}, "required": []},
+}
+
+SEARCH_SCHEMA = {
+    "name": "mem0_search",
+    "description": (
+        "Search memories by meaning. Returns relevant facts ranked by similarity. "
+        "Local-only - no API calls."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "query": {"type": "string", "description": "What to search for."},
+            "top_k": {"type": "integer", "description": "Max results (default: 10, max: 50)."},
+        },
+        "required": ["query"],
+    },
+}
+
+CONCLUDE_SCHEMA = {
+    "name": "mem0_conclude",
+    "description": (
+        "Store a durable fact about the user. Stored verbatim (no LLM extraction). "
+        "Use for explicit preferences, corrections, or decisions. Local-only."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "conclusion": {"type": "string", "description": "The fact to store."},
+        },
+        "required": ["conclusion"],
+    },
+}
+
+
+class Mem0LocalProvider(MemoryProvider):
+    """Local ChromaDB-backed memory provider. No API key required."""
+
+    def __init__(self):
+        self._config = None
+        self._client = None
+        self._collection = None
+        self._client_lock = threading.Lock()
+        self._user_id = "hermes-user"
+        self._storage_path = ""
+        self._max_memories = 10000
+        self._consecutive_failures = 0
+        self._breaker_open_until = 0.0
+
+    @property
+    def name(self) -> str:
+        return "mem0-local"
+
+    def is_available(self) -> bool:
+        try:
+            import chromadb
+            return True
+        except ImportError:
+            return False
+
+    def save_config(self, values, hermes_home):
+        config_path = Path(hermes_home) / "mem0-local.json"
+        existing = {}
+        if config_path.exists():
+            try:
+                existing = json.loads(config_path.read_text())
+            except Exception:
+                pass
+        existing.update(values)
+        config_path.write_text(json.dumps(existing, indent=2))
+
+    def get_config_schema(self):
+        return [
+            {"key": "storage_path", "description": "Storage directory for ChromaDB", "default": "~/.hermes/mem0-local/"},
+            {"key": "collection_prefix", "description": "Collection name prefix", "default": "mem0"},
+            {"key": "max_memories", "description": "Maximum stored memories", "default": "10000"},
+        ]
+
+    def _get_collection(self):
+        """Thread-safe ChromaDB collection accessor with lazy init."""
+        with self._client_lock:
+            if self._collection is not None:
+                return self._collection
+
+            try:
+                import chromadb
+                from chromadb.config import Settings
+            except ImportError:
+                raise RuntimeError("chromadb package not installed. Run: pip install chromadb")
+
+            Path(self._storage_path).mkdir(parents=True, exist_ok=True)
+
+            self._client = chromadb.PersistentClient(
+                path=self._storage_path,
+                settings=Settings(anonymized_telemetry=False),
+            )
+
+            collection_name = f"{self._config.get('collection_prefix', 'mem0')}_memories"
+            self._collection = self._client.get_or_create_collection(
+                name=collection_name,
+                metadata={"hnsw:space": "cosine"},
+            )
+
+            logger.info(
+                "Mem0 local: ChromaDB collection '%s' at %s (%d docs)",
+                collection_name, self._storage_path, self._collection.count(),
+            )
+
+            return self._collection
+
+    def _doc_id(self, content: str) -> str:
+        """Deterministic ID from content hash (for dedup)."""
+        return hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
+
+    def _is_breaker_open(self) -> bool:
+        if self._consecutive_failures < _BREAKER_THRESHOLD:
+            return False
+        if time.monotonic() >= self._breaker_open_until:
+            self._consecutive_failures = 0
+            return False
+        return True
+
+    def _record_success(self):
+        self._consecutive_failures = 0
+
+    def _record_failure(self):
+        self._consecutive_failures += 1
+        if self._consecutive_failures >= _BREAKER_THRESHOLD:
+            self._breaker_open_until = time.monotonic() + _BREAKER_COOLDOWN_SECS
+
+    def initialize(self, session_id: str, **kwargs) -> None:
+        self._config = _load_config()
+        self._storage_path = self._config.get("storage_path", "")
+        self._max_memories = int(self._config.get("max_memories", 10000))
+        self._user_id = kwargs.get("user_id") or self._config.get("user_id", "hermes-user")
+
+    def system_prompt_block(self) -> str:
+        count = 0
+        try:
+            col = self._get_collection()
+            count = col.count()
+        except Exception:
+            pass
+        return (
+            "# Mem0 Local Memory\n"
+            f"Active. {count} memories stored locally. "
+            "Use mem0_search to find memories, mem0_conclude to store facts, "
+            "mem0_profile for a full overview."
+        )
+
+    def prefetch(self, query: str, *, session_id: str = "") -> str:
+        return ""
+
+    def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
+        pass
+
+    def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
+        """Extract and store facts from the conversation turn."""
+        if self._is_breaker_open():
+            return
+        try:
+            col = self._get_collection()
+        except Exception:
+            return
+
+        for content in [user_content, assistant_content]:
+            if not content or len(content) < 10:
+                continue
+            facts = _extract_facts(content)
+            for fact in facts:
+                doc_id = self._doc_id(fact["content"])
+                try:
+                    col.upsert(
+                        ids=[doc_id],
+                        documents=[fact["content"]],
+                        metadatas=[{
+                            "category": fact["category"],
+                            "user_id": self._user_id,
+                            "timestamp": datetime.now(timezone.utc).isoformat(),
+                            "source": "extracted",
+                        }],
+                    )
+                    self._record_success()
+                except Exception as e:
+                    self._record_failure()
+                    logger.debug("Mem0 local: failed to upsert fact: %s", e)
+
+    def get_tool_schemas(self) -> List[Dict[str, Any]]:
+        return [PROFILE_SCHEMA, SEARCH_SCHEMA, CONCLUDE_SCHEMA]
+
+    def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
+        if self._is_breaker_open():
+            return json.dumps({"error": "Local memory temporarily unavailable. Will retry automatically."})
+
+        try:
+            col = self._get_collection()
+        except Exception as e:
+            return tool_error(f"ChromaDB not available: {e}")
+
+        if tool_name == "mem0_profile":
+            try:
+                results = col.get(
+                    where={"user_id": self._user_id} if self._user_id else None,
+                    limit=500,
+                )
+                documents = results.get("documents", [])
+                if not documents:
+                    return json.dumps({"result": "No memories stored yet."})
+                lines = [d for d in documents if d]
+                self._record_success()
+                return json.dumps({"result": "\n".join(f"- {l}" for l in lines), "count": len(lines)})
+            except Exception as e:
+                self._record_failure()
+                return tool_error(f"Failed to fetch profile: {e}")
+
+        elif tool_name == "mem0_search":
+            query = args.get("query", "")
+            if not query:
+                return tool_error("Missing required parameter: query")
+            top_k = min(int(args.get("top_k", 10)), 50)
+
+            try:
+                results = col.query(
+                    query_texts=[query],
+                    n_results=top_k,
+                    where={"user_id": self._user_id} if self._user_id else None,
+                )
+
+                documents = results.get("documents", [[]])[0]
+                distances = results.get("distances", [[]])[0]
+
+                if not documents:
+                    return json.dumps({"result": "No relevant memories found."})
+
+                items = []
+                for doc, dist in zip(documents, distances):
+                    score = max(0, 1 - (dist / 2))
+                    items.append({"memory": doc, "score": round(score, 3)})
+
+                self._record_success()
+                return json.dumps({"results": items, "count": len(items)})
+            except Exception as e:
+                self._record_failure()
+                return tool_error(f"Search failed: {e}")
+
+        elif tool_name == "mem0_conclude":
+            conclusion = args.get("conclusion", "")
+            if not conclusion:
+                return tool_error("Missing required parameter: conclusion")
+
+            try:
+                doc_id = self._doc_id(conclusion)
+                col.upsert(
+                    ids=[doc_id],
+                    documents=[conclusion],
+                    metadatas=[{
+                        "category": "explicit",
+                        "user_id": self._user_id,
+                        "timestamp": datetime.now(timezone.utc).isoformat(),
+                        "source": "conclude",
+                    }],
+                )
+                self._record_success()
+                return json.dumps({"result": "Fact stored locally.", "id": doc_id})
+            except Exception as e:
+                self._record_failure()
+                return tool_error(f"Failed to store: {e}")
+
+        return tool_error(f"Unknown tool: {tool_name}")
+
+    def shutdown(self) -> None:
+        with self._client_lock:
+            self._collection = None
+            self._client = None
+
+
+def register(ctx) -> None:
+    """Register Mem0 Local as a memory provider plugin."""
+    ctx.register_memory_provider(Mem0LocalProvider())
--- a/plugins/memory/mem0_local/plugin.yaml
+++ b/plugins/memory/mem0_local/plugin.yaml
@@ -0,0 +1,5 @@
+name: mem0_local
+version: 1.0.0
+description: "Mem0 local mode — ChromaDB-backed memory with no API key required. Sovereign deployment."
+pip_dependencies:
+  - chromadb
--- a/tests/agent/test_privacy_filter.py
+++ b/tests/agent/test_privacy_filter.py
@@ -0,0 +1,202 @@
+"""Tests for agent.privacy_filter — PII stripping before remote API calls."""
+
+import pytest
+from agent.privacy_filter import (
+    PrivacyFilter,
+    RedactionReport,
+    Sensitivity,
+    sanitize_messages,
+    quick_sanitize,
+)
+
+
+class TestPrivacyFilterSanitizeText:
+    """Test single-text sanitization."""
+
+    def test_no_pii_returns_clean(self):
+        pf = PrivacyFilter()
+        text = "The weather in Paris is nice today."
+        cleaned, redactions = pf.sanitize_text(text)
+        assert cleaned == text
+        assert redactions == []
+
+    def test_email_redacted(self):
+        pf = PrivacyFilter()
+        text = "Send report to alice@example.com by Friday."
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "alice@example.com" not in cleaned
+        assert "[REDACTED-EMAIL]" in cleaned
+        assert any(r["type"] == "email_address" for r in redactions)
+
+    def test_phone_redacted(self):
+        pf = PrivacyFilter()
+        text = "Call me at 555-123-4567 when ready."
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "555-123-4567" not in cleaned
+        assert "[REDACTED-PHONE]" in cleaned
+
+    def test_api_key_redacted(self):
+        pf = PrivacyFilter()
+        text = 'api_key = "sk-proj-abcdefghij1234567890abcdefghij1234567890"'
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "sk-proj-" not in cleaned
+        assert any(r["sensitivity"] == "CRITICAL" for r in redactions)
+
+    def test_github_token_redacted(self):
+        pf = PrivacyFilter()
+        text = "Use ghp_1234567890abcdefghijklmnopqrstuvwxyz1234 for auth"
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "ghp_" not in cleaned
+        assert any(r["type"] == "github_token" for r in redactions)
+
+    def test_ethereum_address_redacted(self):
+        pf = PrivacyFilter()
+        text = "Send to 0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18 please"
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "0x742d" not in cleaned
+        assert any(r["type"] == "ethereum_address" for r in redactions)
+
+    def test_user_home_path_redacted(self):
+        pf = PrivacyFilter()
+        text = "Read file at /Users/alice/Documents/secret.txt"
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "alice" not in cleaned
+        assert "[REDACTED-USER]" in cleaned
+
+    def test_multiple_pii_types(self):
+        pf = PrivacyFilter()
+        text = (
+            "Contact john@test.com or call 555-999-1234. "
+            "The API key is sk-abcdefghijklmnopqrstuvwxyz1234567890."
+        )
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "john@test.com" not in cleaned
+        assert "555-999-1234" not in cleaned
+        assert "sk-abcd" not in cleaned
+        assert len(redactions) >= 3
+
+
+class TestPrivacyFilterSanitizeMessages:
+    """Test message-list sanitization."""
+
+    def test_sanitize_user_message(self):
+        pf = PrivacyFilter()
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "Email me at bob@test.com with results."},
+        ]
+        safe, report = pf.sanitize_messages(messages)
+        assert report.redacted_messages == 1
+        assert "bob@test.com" not in safe[1]["content"]
+        assert "[REDACTED-EMAIL]" in safe[1]["content"]
+        # System message unchanged
+        assert safe[0]["content"] == "You are helpful."
+
+    def test_no_redaction_needed(self):
+        pf = PrivacyFilter()
+        messages = [
+            {"role": "user", "content": "What is 2+2?"},
+            {"role": "assistant", "content": "4"},
+        ]
+        safe, report = pf.sanitize_messages(messages)
+        assert report.redacted_messages == 0
+        assert not report.had_redactions
+
+    def test_assistant_messages_also_sanitized(self):
+        pf = PrivacyFilter()
+        messages = [
+            {"role": "assistant", "content": "Your email admin@corp.com was found."},
+        ]
+        safe, report = pf.sanitize_messages(messages)
+        assert report.redacted_messages == 1
+        assert "admin@corp.com" not in safe[0]["content"]
+
+    def test_tool_messages_not_sanitized(self):
+        pf = PrivacyFilter()
+        messages = [
+            {"role": "tool", "content": "Result: user@test.com found"},
+        ]
+        safe, report = pf.sanitize_messages(messages)
+        assert report.redacted_messages == 0
+        assert safe[0]["content"] == "Result: user@test.com found"
+
+
+class TestShouldUseLocalOnly:
+    """Test the local-only routing decision."""
+
+    def test_normal_text_allows_remote(self):
+        pf = PrivacyFilter()
+        block, reason = pf.should_use_local_only("Summarize this article about Python.")
+        assert not block
+
+    def test_critical_secret_blocks_remote(self):
+        pf = PrivacyFilter()
+        text = "Here is the API key: sk-abcdefghijklmnopqrstuvwxyz1234567890"
+        block, reason = pf.should_use_local_only(text)
+        assert block
+        assert "critical" in reason.lower()
+
+    def test_multiple_high_sensitivity_blocks(self):
+        pf = PrivacyFilter()
+        # 3+ high-sensitivity patterns
+        text = (
+            "Card: 4111-1111-1111-1111, "
+            "SSN: 123-45-6789, "
+            "BTC: 1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa, "
+            "ETH: 0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18"
+        )
+        block, reason = pf.should_use_local_only(text)
+        assert block
+
+
+class TestAggressiveMode:
+    """Test aggressive filtering mode."""
+
+    def test_aggressive_redacts_internal_ips(self):
+        pf = PrivacyFilter(aggressive_mode=True)
+        text = "Server at 192.168.1.100 is responding."
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "192.168.1.100" not in cleaned
+        assert any(r["type"] == "internal_ip" for r in redactions)
+
+    def test_normal_does_not_redact_ips(self):
+        pf = PrivacyFilter(aggressive_mode=False)
+        text = "Server at 192.168.1.100 is responding."
+        cleaned, redactions = pf.sanitize_text(text)
+        assert "192.168.1.100" in cleaned  # IP preserved in normal mode
+
+
+class TestConvenienceFunctions:
+    """Test module-level convenience functions."""
+
+    def test_quick_sanitize(self):
+        text = "Contact alice@example.com for details"
+        result = quick_sanitize(text)
+        assert "alice@example.com" not in result
+        assert "[REDACTED-EMAIL]" in result
+
+    def test_sanitize_messages_convenience(self):
+        messages = [{"role": "user", "content": "Call 555-000-1234"}]
+        safe, report = sanitize_messages(messages)
+        assert report.redacted_messages == 1
+
+
+class TestRedactionReport:
+    """Test the reporting structure."""
+
+    def test_summary_no_redactions(self):
+        report = RedactionReport(total_messages=3, redacted_messages=0)
+        assert "No PII" in report.summary()
+
+    def test_summary_with_redactions(self):
+        report = RedactionReport(
+            total_messages=2,
+            redacted_messages=1,
+            redactions=[
+                {"type": "email_address", "sensitivity": "MEDIUM", "count": 2},
+                {"type": "phone_number_us", "sensitivity": "MEDIUM", "count": 1},
+            ],
+        )
+        summary = report.summary()
+        assert "1/2" in summary
+        assert "email_address" in summary
--- a/tests/plugins/memory/test_mem0_local.py
+++ b/tests/plugins/memory/test_mem0_local.py
@@ -0,0 +1,173 @@
+"""Tests for Mem0 Local memory provider - ChromaDB-backed, no API key."""
+
+import json
+import os
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# Fact extraction tests
+
+class TestFactExtraction:
+    """Test the regex-based fact extraction."""
+
+    def _extract(self, text):
+        from plugins.memory.mem0_local import _extract_facts
+        return _extract_facts(text)
+
+    def test_name_extraction(self):
+        facts = self._extract("My name is Alexander Whitestone.")
+        assert any("alexander whitestone" in f["content"].lower() for f in facts)
+
+    def test_preference_extraction(self):
+        facts = self._extract("I prefer using vim for editing.")
+        assert any("vim" in f["content"].lower() for f in facts)
+
+    def test_timezone_extraction(self):
+        facts = self._extract("My timezone is America/New_York.")
+        assert any("america/new_york" in f["content"].lower() for f in facts)
+
+    def test_explicit_remember(self):
+        facts = self._extract("Remember: always use f-strings in Python.")
+        assert len(facts) > 0
+
+    def test_correction_extraction(self):
+        facts = self._extract("Actually: the port is 8080, not 3000.")
+        assert len(facts) > 0
+
+    def test_empty_input(self):
+        facts = self._extract("")
+        assert facts == []
+
+    def test_short_input_ignored(self):
+        facts = self._extract("Hi")
+        assert facts == []
+
+    def test_no_crash_on_random_text(self):
+        facts = self._extract("The quick brown fox jumps over the lazy dog. " * 10)
+        assert isinstance(facts, list)
+
+
+# Config tests
+
+class TestConfig:
+    """Test configuration loading."""
+
+    def test_default_storage_path(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+        from plugins.memory.mem0_local import _load_config
+        config = _load_config()
+        assert "mem0-local" in config["storage_path"]
+
+    def test_env_override(self, tmp_path, monkeypatch):
+        custom_path = str(tmp_path / "custom-mem0")
+        monkeypatch.setenv("MEM0_LOCAL_PATH", custom_path)
+        from plugins.memory.mem0_local import _load_config
+        config = _load_config()
+        assert config["storage_path"] == custom_path
+
+
+# Provider interface tests
+
+class TestProviderInterface:
+    """Test provider interface methods."""
+
+    def test_name(self):
+        from plugins.memory.mem0_local import Mem0LocalProvider
+        provider = Mem0LocalProvider()
+        assert provider.name == "mem0-local"
+
+    def test_tool_schemas(self):
+        from plugins.memory.mem0_local import Mem0LocalProvider
+        provider = Mem0LocalProvider()
+        schemas = provider.get_tool_schemas()
+        names = {s["name"] for s in schemas}
+        assert names == {"mem0_profile", "mem0_search", "mem0_conclude"}
+
+    def test_schema_required_params(self):
+        from plugins.memory.mem0_local import Mem0LocalProvider
+        provider = Mem0LocalProvider()
+        schemas = {s["name"]: s for s in provider.get_tool_schemas()}
+        assert "query" in schemas["mem0_search"]["parameters"]["required"]
+        assert "conclusion" in schemas["mem0_conclude"]["parameters"]["required"]
+
+
+# ChromaDB integration tests
+
+chromadb = None
+try:
+    import chromadb
+except ImportError:
+    pass
+
+
+@pytest.mark.skipif(chromadb is None, reason="chromadb not installed")
+class TestChromaDBIntegration:
+    """Integration tests with real ChromaDB."""
+
+    @pytest.fixture
+    def provider(self, tmp_path, monkeypatch):
+        from plugins.memory.mem0_local import Mem0LocalProvider
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+        provider = Mem0LocalProvider()
+        provider.initialize("test-session")
+        provider._storage_path = str(tmp_path / "mem0-test")
+        return provider
+
+    def test_store_and_search(self, provider):
+        result = provider.handle_tool_call("mem0_conclude", {"conclusion": "User prefers Python over JavaScript"})
+        data = json.loads(result)
+        assert data.get("result") == "Fact stored locally."
+
+        result = provider.handle_tool_call("mem0_search", {"query": "programming language preference"})
+        data = json.loads(result)
+        assert data["count"] > 0
+        assert any("python" in item["memory"].lower() for item in data["results"])
+
+    def test_profile_empty(self, provider):
+        result = provider.handle_tool_call("mem0_profile", {})
+        data = json.loads(result)
+        assert "No memories" in data.get("result", "") or data.get("count", 0) == 0
+
+    def test_profile_after_store(self, provider):
+        provider.handle_tool_call("mem0_conclude", {"conclusion": "User name is Alexander"})
+        provider.handle_tool_call("mem0_conclude", {"conclusion": "User timezone is UTC"})
+
+        result = provider.handle_tool_call("mem0_profile", {})
+        data = json.loads(result)
+        assert data["count"] >= 2
+
+    def test_dedup(self, provider):
+        provider.handle_tool_call("mem0_conclude", {"conclusion": "Project uses SQLite"})
+        provider.handle_tool_call("mem0_conclude", {"conclusion": "Project uses SQLite"})
+
+        result = provider.handle_tool_call("mem0_profile", {})
+        data = json.loads(result)
+        assert data["count"] == 1
+
+    def test_search_no_results(self, provider):
+        result = provider.handle_tool_call("mem0_search", {"query": "nonexistent topic xyz123"})
+        data = json.loads(result)
+        assert data.get("result") == "No relevant memories found." or data.get("count", 0) == 0
+
+    def test_sync_turn_extraction(self, provider):
+        provider.sync_turn(
+            "My name is TestUser and I prefer dark mode.",
+            "Hello TestUser! I'll remember your preference.",
+        )
+        result = provider.handle_tool_call("mem0_profile", {})
+        data = json.loads(result)
+        assert "count" in data
+
+    def test_conclude_missing_param(self, provider):
+        result = provider.handle_tool_call("mem0_conclude", {})
+        data = json.loads(result)
+        assert "error" in data
+
+    def test_search_missing_query(self, provider):
+        result = provider.handle_tool_call("mem0_search", {})
+        data = json.loads(result)
+        assert "error" in data
--- a/tests/tools/test_confirmation_daemon.py
+++ b/tests/tools/test_confirmation_daemon.py
@@ -0,0 +1,190 @@
+"""Tests for tools.confirmation_daemon — Human Confirmation Firewall."""
+
+import pytest
+import time
+from tools.confirmation_daemon import (
+    ConfirmationDaemon,
+    ConfirmationRequest,
+    ConfirmationStatus,
+    RiskLevel,
+    classify_action,
+    _is_whitelisted,
+    _DEFAULT_WHITELIST,
+)
+
+
+class TestClassifyAction:
+    """Test action risk classification."""
+
+    def test_crypto_tx_is_critical(self):
+        assert classify_action("crypto_tx") == RiskLevel.CRITICAL
+
+    def test_sign_transaction_is_critical(self):
+        assert classify_action("sign_transaction") == RiskLevel.CRITICAL
+
+    def test_send_email_is_high(self):
+        assert classify_action("send_email") == RiskLevel.HIGH
+
+    def test_send_message_is_medium(self):
+        assert classify_action("send_message") == RiskLevel.MEDIUM
+
+    def test_access_calendar_is_low(self):
+        assert classify_action("access_calendar") == RiskLevel.LOW
+
+    def test_unknown_action_is_medium(self):
+        assert classify_action("unknown_action_xyz") == RiskLevel.MEDIUM
+
+
+class TestWhitelist:
+    """Test whitelist auto-approval."""
+
+    def test_self_email_is_whitelisted(self):
+        whitelist = dict(_DEFAULT_WHITELIST)
+        payload = {"from": "me@test.com", "to": "me@test.com"}
+        assert _is_whitelisted("send_email", payload, whitelist) is True
+
+    def test_non_whitelisted_recipient_not_approved(self):
+        whitelist = dict(_DEFAULT_WHITELIST)
+        payload = {"to": "random@stranger.com"}
+        assert _is_whitelisted("send_email", payload, whitelist) is False
+
+    def test_whitelisted_contact_approved(self):
+        whitelist = {
+            "send_message": {"targets": ["alice", "bob"]},
+        }
+        assert _is_whitelisted("send_message", {"to": "alice"}, whitelist) is True
+        assert _is_whitelisted("send_message", {"to": "charlie"}, whitelist) is False
+
+    def test_no_whitelist_entry_means_not_whitelisted(self):
+        whitelist = {}
+        assert _is_whitelisted("crypto_tx", {"amount": 1.0}, whitelist) is False
+
+
+class TestConfirmationRequest:
+    """Test the request data model."""
+
+    def test_defaults(self):
+        req = ConfirmationRequest(
+            request_id="test-1",
+            action="send_email",
+            description="Test email",
+            risk_level="high",
+            payload={},
+        )
+        assert req.status == ConfirmationStatus.PENDING.value
+        assert req.created_at > 0
+        assert req.expires_at > req.created_at
+
+    def test_is_pending(self):
+        req = ConfirmationRequest(
+            request_id="test-2",
+            action="send_email",
+            description="Test",
+            risk_level="high",
+            payload={},
+            expires_at=time.time() + 300,
+        )
+        assert req.is_pending is True
+
+    def test_is_expired(self):
+        req = ConfirmationRequest(
+            request_id="test-3",
+            action="send_email",
+            description="Test",
+            risk_level="high",
+            payload={},
+            expires_at=time.time() - 10,
+        )
+        assert req.is_expired is True
+        assert req.is_pending is False
+
+    def test_to_dict(self):
+        req = ConfirmationRequest(
+            request_id="test-4",
+            action="send_email",
+            description="Test",
+            risk_level="medium",
+            payload={"to": "a@b.com"},
+        )
+        d = req.to_dict()
+        assert d["request_id"] == "test-4"
+        assert d["action"] == "send_email"
+        assert "is_pending" in d
+
+
+class TestConfirmationDaemon:
+    """Test the daemon logic (without HTTP layer)."""
+
+    def test_auto_approve_low_risk(self):
+        daemon = ConfirmationDaemon()
+        req = daemon.request(
+            action="access_calendar",
+            description="Read today's events",
+            risk_level="low",
+        )
+        assert req.status == ConfirmationStatus.AUTO_APPROVED.value
+
+    def test_whitelisted_auto_approves(self):
+        daemon = ConfirmationDaemon()
+        daemon._whitelist = {"send_message": {"targets": ["alice"]}}
+        req = daemon.request(
+            action="send_message",
+            description="Message alice",
+            payload={"to": "alice"},
+        )
+        assert req.status == ConfirmationStatus.AUTO_APPROVED.value
+
+    def test_non_whitelisted_goes_pending(self):
+        daemon = ConfirmationDaemon()
+        daemon._whitelist = {}
+        req = daemon.request(
+            action="send_email",
+            description="Email to stranger",
+            payload={"to": "stranger@test.com"},
+            risk_level="high",
+        )
+        assert req.status == ConfirmationStatus.PENDING.value
+        assert req.is_pending is True
+
+    def test_approve_response(self):
+        daemon = ConfirmationDaemon()
+        daemon._whitelist = {}
+        req = daemon.request(
+            action="send_email",
+            description="Email test",
+            risk_level="high",
+        )
+        result = daemon.respond(req.request_id, approved=True, decided_by="human")
+        assert result.status == ConfirmationStatus.APPROVED.value
+        assert result.decided_by == "human"
+
+    def test_deny_response(self):
+        daemon = ConfirmationDaemon()
+        daemon._whitelist = {}
+        req = daemon.request(
+            action="crypto_tx",
+            description="Send 1 ETH",
+            risk_level="critical",
+        )
+        result = daemon.respond(
+            req.request_id, approved=False, decided_by="human", reason="Too risky"
+        )
+        assert result.status == ConfirmationStatus.DENIED.value
+        assert result.reason == "Too risky"
+
+    def test_get_pending(self):
+        daemon = ConfirmationDaemon()
+        daemon._whitelist = {}
+        daemon.request(action="send_email", description="Test 1", risk_level="high")
+        daemon.request(action="send_email", description="Test 2", risk_level="high")
+        pending = daemon.get_pending()
+        assert len(pending) >= 2
+
+    def test_get_history(self):
+        daemon = ConfirmationDaemon()
+        req = daemon.request(
+            action="access_calendar", description="Test", risk_level="low"
+        )
+        history = daemon.get_history()
+        assert len(history) >= 1
+        assert history[0]["action"] == "access_calendar"
--- a/tools/approval.py
+++ b/tools/approval.py
@@ -121,6 +121,19 @@ DANGEROUS_PATTERNS = [
    (r'\b(cp|mv|install)\b.*\s/etc/', "copy/move file into /etc/"),
    (r'\bsed\s+-[^\s]*i.*\s/etc/', "in-place edit of system config"),
    (r'\bsed\s+--in-place\b.*\s/etc/', "in-place edit of system config (long flag)"),
+    # --- Vitalik's threat model: crypto / financial ---
+    (r'\b(?:bitcoin-cli|ethers\.js|web3|ether\.sendTransaction)\b', "direct crypto transaction tool usage"),
+    (r'\bwget\b.*\b(?:mnemonic|seed\s*phrase|private[_-]?key)\b', "attempting to download crypto credentials"),
+    (r'\bcurl\b.*\b(?:mnemonic|seed\s*phrase|private[_-]?key)\b', "attempting to exfiltrate crypto credentials"),
+    # --- Vitalik's threat model: credential exfiltration ---
+    (r'\b(?:curl|wget|http|nc|ncat|socat)\b.*\b(?:\.env|\.ssh|credentials|secrets|token|api[_-]?key)\b',
+     "attempting to exfiltrate credentials via network"),
+    (r'\bbase64\b.*\|(?:\s*curl|\s*wget)', "base64-encode then network exfiltration"),
+    (r'\bcat\b.*\b(?:\.env|\.ssh/id_rsa|credentials)\b.*\|(?:\s*curl|\s*wget)',
+     "reading secrets and piping to network tool"),
+    # --- Vitalik's threat model: data exfiltration ---
+    (r'\bcurl\b.*-d\s.*\$(?:HOME|USER)', "sending user home directory data to remote"),
+    (r'\bwget\b.*--post-data\s.*\$(?:HOME|USER)', "posting user data to remote"),
    # Script execution via heredoc — bypasses the -e/-c flag patterns above.
    # `python3 << 'EOF'` feeds arbitrary code via stdin without -c/-e flags.
    (r'\b(python[23]?|perl|ruby|node)\s+<<', "script execution via heredoc"),
--- a/tools/confirmation_daemon.py
+++ b/tools/confirmation_daemon.py
@@ -0,0 +1,615 @@
+"""Human Confirmation Daemon — HTTP server for two-factor action approval.
+
+Implements Vitalik's Pattern 1: "The new 'two-factor confirmation' is that
+the two factors are the human and the LLM."
+
+This daemon runs on localhost:6000 and provides a simple HTTP API for the
+agent to request human approval before executing high-risk actions.
+
+Threat model:
+- LLM jailbreaks: Remote content "hacking" the LLM to perform malicious actions
+- LLM accidents: LLM accidentally performing dangerous operations
+- The human acts as the second factor — the agent proposes, the human disposes
+
+Architecture:
+- Agent detects high-risk action → POST /confirm with action details
+- Daemon stores pending request, sends notification to user
+- User approves/denies via POST /respond (Telegram, CLI, or direct HTTP)
+- Agent receives decision and proceeds or aborts
+
+Usage:
+    # Start daemon (usually managed by gateway)
+    from tools.confirmation_daemon import ConfirmationDaemon
+    daemon = ConfirmationDaemon(port=6000)
+    daemon.start()
+
+    # Request approval (from agent code)
+    from tools.confirmation_daemon import request_confirmation
+    approved = request_confirmation(
+        action="send_email",
+        description="Send email to alice@example.com",
+        risk_level="high",
+        payload={"to": "alice@example.com", "subject": "Meeting notes"},
+        timeout=300,
+    )
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import threading
+import time
+import uuid
+from dataclasses import dataclass, field, asdict
+from enum import Enum, auto
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+class RiskLevel(Enum):
+    """Risk classification for actions requiring confirmation."""
+    LOW = "low"           # Log only, no confirmation needed
+    MEDIUM = "medium"     # Confirm for non-whitelisted targets
+    HIGH = "high"         # Always confirm
+    CRITICAL = "critical" # Always confirm + require explicit reason
+
+
+class ConfirmationStatus(Enum):
+    """Status of a pending confirmation request."""
+    PENDING = "pending"
+    APPROVED = "approved"
+    DENIED = "denied"
+    EXPIRED = "expired"
+    AUTO_APPROVED = "auto_approved"
+
+
+@dataclass
+class ConfirmationRequest:
+    """A request for human confirmation of a high-risk action."""
+    request_id: str
+    action: str               # Action type: send_email, send_message, crypto_tx, etc.
+    description: str          # Human-readable description of what will happen
+    risk_level: str           # low, medium, high, critical
+    payload: Dict[str, Any]   # Action-specific data (sanitized)
+    session_key: str = ""     # Session that initiated the request
+    created_at: float = 0.0
+    expires_at: float = 0.0
+    status: str = ConfirmationStatus.PENDING.value
+    decided_at: float = 0.0
+    decided_by: str = ""      # "human", "auto", "whitelist"
+    reason: str = ""          # Optional reason for denial
+
+    def __post_init__(self):
+        if not self.created_at:
+            self.created_at = time.time()
+        if not self.expires_at:
+            self.expires_at = self.created_at + 300  # 5 min default
+        if not self.request_id:
+            self.request_id = str(uuid.uuid4())[:12]
+
+    @property
+    def is_expired(self) -> bool:
+        return time.time() > self.expires_at
+
+    @property
+    def is_pending(self) -> bool:
+        return self.status == ConfirmationStatus.PENDING.value and not self.is_expired
+
+    def to_dict(self) -> Dict[str, Any]:
+        d = asdict(self)
+        d["is_expired"] = self.is_expired
+        d["is_pending"] = self.is_pending
+        return d
+
+
+# =========================================================================
+# Action categories (Vitalik's threat model)
+# =========================================================================
+
+ACTION_CATEGORIES = {
+    # Messaging — outbound communication to external parties
+    "send_email": RiskLevel.HIGH,
+    "send_message": RiskLevel.MEDIUM,     # Depends on recipient
+    "send_signal": RiskLevel.HIGH,
+    "send_telegram": RiskLevel.MEDIUM,
+    "send_discord": RiskLevel.MEDIUM,
+    "post_social": RiskLevel.HIGH,
+
+    # Financial / crypto
+    "crypto_tx": RiskLevel.CRITICAL,
+    "sign_transaction": RiskLevel.CRITICAL,
+    "access_wallet": RiskLevel.CRITICAL,
+    "modify_balance": RiskLevel.CRITICAL,
+
+    # System modification
+    "install_software": RiskLevel.HIGH,
+    "modify_system_config": RiskLevel.HIGH,
+    "modify_firewall": RiskLevel.CRITICAL,
+    "add_ssh_key": RiskLevel.CRITICAL,
+    "create_user": RiskLevel.CRITICAL,
+
+    # Data access
+    "access_contacts": RiskLevel.MEDIUM,
+    "access_calendar": RiskLevel.LOW,
+    "read_private_files": RiskLevel.MEDIUM,
+    "upload_data": RiskLevel.HIGH,
+    "share_credentials": RiskLevel.CRITICAL,
+
+    # Network
+    "open_port": RiskLevel.HIGH,
+    "modify_dns": RiskLevel.HIGH,
+    "expose_service": RiskLevel.CRITICAL,
+}
+
+# Default: any unrecognized action is MEDIUM risk
+DEFAULT_RISK_LEVEL = RiskLevel.MEDIUM
+
+
+def classify_action(action: str) -> RiskLevel:
+    """Classify an action by its risk level."""
+    return ACTION_CATEGORIES.get(action, DEFAULT_RISK_LEVEL)
+
+
+# =========================================================================
+# Whitelist configuration
+# =========================================================================
+
+_DEFAULT_WHITELIST = {
+    "send_message": {
+        "targets": [],   # Contact names/IDs that don't need confirmation
+    },
+    "send_email": {
+        "targets": [],   # Email addresses that don't need confirmation
+        "self_only": True,  # send-to-self always allowed
+    },
+}
+
+
+def _load_whitelist() -> Dict[str, Any]:
+    """Load action whitelist from config."""
+    config_path = Path.home() / ".hermes" / "approval_whitelist.json"
+    if config_path.exists():
+        try:
+            with open(config_path) as f:
+                return json.load(f)
+        except Exception as e:
+            logger.warning("Failed to load approval whitelist: %s", e)
+    return dict(_DEFAULT_WHITELIST)
+
+
+def _is_whitelisted(action: str, payload: Dict[str, Any], whitelist: Dict) -> bool:
+    """Check if an action is pre-approved by the whitelist."""
+    action_config = whitelist.get(action, {})
+    if not action_config:
+        return False
+
+    # Check target-based whitelist
+    targets = action_config.get("targets", [])
+    target = payload.get("to") or payload.get("recipient") or payload.get("target", "")
+    if target and target in targets:
+        return True
+
+    # Self-only email
+    if action_config.get("self_only") and action == "send_email":
+        sender = payload.get("from", "")
+        recipient = payload.get("to", "")
+        if sender and recipient and sender.lower() == recipient.lower():
+            return True
+
+    return False
+
+
+# =========================================================================
+# Confirmation daemon
+# =========================================================================
+
+class ConfirmationDaemon:
+    """HTTP daemon for human confirmation of high-risk actions.
+
+    Runs on localhost:PORT (default 6000). Provides:
+    - POST /confirm   — agent requests human approval
+    - POST /respond   — human approves/denies
+    - GET  /pending   — list pending requests
+    - GET  /health    — health check
+    """
+
+    def __init__(
+        self,
+        host: str = "127.0.0.1",
+        port: int = 6000,
+        default_timeout: int = 300,
+        notify_callback: Optional[Callable] = None,
+    ):
+        self.host = host
+        self.port = port
+        self.default_timeout = default_timeout
+        self.notify_callback = notify_callback
+        self._pending: Dict[str, ConfirmationRequest] = {}
+        self._history: List[ConfirmationRequest] = []
+        self._lock = threading.Lock()
+        self._whitelist = _load_whitelist()
+        self._app = None
+        self._runner = None
+
+    def request(
+        self,
+        action: str,
+        description: str,
+        payload: Optional[Dict[str, Any]] = None,
+        risk_level: Optional[str] = None,
+        session_key: str = "",
+        timeout: Optional[int] = None,
+    ) -> ConfirmationRequest:
+        """Create a confirmation request.
+
+        Returns the request. Check .status to see if it was immediately
+        auto-approved (whitelisted) or is pending human review.
+        """
+        payload = payload or {}
+
+        # Classify risk if not specified
+        if risk_level is None:
+            risk_level = classify_action(action).value
+
+        # Check whitelist
+        if risk_level in ("low",) or _is_whitelisted(action, payload, self._whitelist):
+            req = ConfirmationRequest(
+                request_id=str(uuid.uuid4())[:12],
+                action=action,
+                description=description,
+                risk_level=risk_level,
+                payload=payload,
+                session_key=session_key,
+                expires_at=time.time() + (timeout or self.default_timeout),
+                status=ConfirmationStatus.AUTO_APPROVED.value,
+                decided_at=time.time(),
+                decided_by="whitelist",
+            )
+            with self._lock:
+                self._history.append(req)
+            logger.info("Auto-approved whitelisted action: %s", action)
+            return req
+
+        # Create pending request
+        req = ConfirmationRequest(
+            request_id=str(uuid.uuid4())[:12],
+            action=action,
+            description=description,
+            risk_level=risk_level,
+            payload=payload,
+            session_key=session_key,
+            expires_at=time.time() + (timeout or self.default_timeout),
+        )
+
+        with self._lock:
+            self._pending[req.request_id] = req
+
+        # Notify human
+        if self.notify_callback:
+            try:
+                self.notify_callback(req.to_dict())
+            except Exception as e:
+                logger.warning("Confirmation notify callback failed: %s", e)
+
+        logger.info(
+            "Confirmation request %s: %s (%s risk) — waiting for human",
+            req.request_id, action, risk_level,
+        )
+        return req
+
+    def respond(
+        self,
+        request_id: str,
+        approved: bool,
+        decided_by: str = "human",
+        reason: str = "",
+    ) -> Optional[ConfirmationRequest]:
+        """Record a human decision on a pending request."""
+        with self._lock:
+            req = self._pending.get(request_id)
+            if not req:
+                logger.warning("Confirmation respond: unknown request %s", request_id)
+                return None
+            if not req.is_pending:
+                logger.warning("Confirmation respond: request %s already decided", request_id)
+                return req
+
+            req.status = (
+                ConfirmationStatus.APPROVED.value if approved
+                else ConfirmationStatus.DENIED.value
+            )
+            req.decided_at = time.time()
+            req.decided_by = decided_by
+            req.reason = reason
+
+            # Move to history
+            del self._pending[request_id]
+            self._history.append(req)
+
+        logger.info(
+            "Confirmation %s: %s by %s",
+            request_id, "APPROVED" if approved else "DENIED", decided_by,
+        )
+        return req
+
+    def wait_for_decision(
+        self, request_id: str, timeout: Optional[float] = None
+    ) -> ConfirmationRequest:
+        """Block until a decision is made or timeout expires."""
+        deadline = time.time() + (timeout or self.default_timeout)
+        while time.time() < deadline:
+            with self._lock:
+                req = self._pending.get(request_id)
+                if req and not req.is_pending:
+                    return req
+                if req and req.is_expired:
+                    req.status = ConfirmationStatus.EXPIRED.value
+                    del self._pending[request_id]
+                    self._history.append(req)
+                    return req
+            time.sleep(0.5)
+
+        # Timeout
+        with self._lock:
+            req = self._pending.pop(request_id, None)
+            if req:
+                req.status = ConfirmationStatus.EXPIRED.value
+                self._history.append(req)
+                return req
+
+        # Shouldn't reach here
+        return ConfirmationRequest(
+            request_id=request_id,
+            action="unknown",
+            description="Request not found",
+            risk_level="high",
+            payload={},
+            status=ConfirmationStatus.EXPIRED.value,
+        )
+
+    def get_pending(self) -> List[Dict[str, Any]]:
+        """Return list of pending confirmation requests."""
+        self._expire_old()
+        with self._lock:
+            return [r.to_dict() for r in self._pending.values() if r.is_pending]
+
+    def get_history(self, limit: int = 50) -> List[Dict[str, Any]]:
+        """Return recent confirmation history."""
+        with self._lock:
+            return [r.to_dict() for r in self._history[-limit:]]
+
+    def _expire_old(self) -> None:
+        """Move expired requests to history."""
+        now = time.time()
+        with self._lock:
+            expired = [
+                rid for rid, req in self._pending.items()
+                if now > req.expires_at
+            ]
+            for rid in expired:
+                req = self._pending.pop(rid)
+                req.status = ConfirmationStatus.EXPIRED.value
+                self._history.append(req)
+
+    # --- aiohttp HTTP API ---
+
+    async def _handle_health(self, request):
+        from aiohttp import web
+        return web.json_response({
+            "status": "ok",
+            "service": "hermes-confirmation-daemon",
+            "pending": len(self._pending),
+        })
+
+    async def _handle_confirm(self, request):
+        from aiohttp import web
+        try:
+            body = await request.json()
+        except Exception:
+            return web.json_response({"error": "invalid JSON"}, status=400)
+
+        action = body.get("action", "")
+        description = body.get("description", "")
+        if not action or not description:
+            return web.json_response(
+                {"error": "action and description required"}, status=400
+            )
+
+        req = self.request(
+            action=action,
+            description=description,
+            payload=body.get("payload", {}),
+            risk_level=body.get("risk_level"),
+            session_key=body.get("session_key", ""),
+            timeout=body.get("timeout"),
+        )
+
+        # If auto-approved, return immediately
+        if req.status != ConfirmationStatus.PENDING.value:
+            return web.json_response({
+                "request_id": req.request_id,
+                "status": req.status,
+                "decided_by": req.decided_by,
+            })
+
+        # Otherwise, wait for human decision (with timeout)
+        timeout = min(body.get("timeout", self.default_timeout), 600)
+        result = self.wait_for_decision(req.request_id, timeout=timeout)
+
+        return web.json_response({
+            "request_id": result.request_id,
+            "status": result.status,
+            "decided_by": result.decided_by,
+            "reason": result.reason,
+        })
+
+    async def _handle_respond(self, request):
+        from aiohttp import web
+        try:
+            body = await request.json()
+        except Exception:
+            return web.json_response({"error": "invalid JSON"}, status=400)
+
+        request_id = body.get("request_id", "")
+        approved = body.get("approved")
+        if not request_id or approved is None:
+            return web.json_response(
+                {"error": "request_id and approved required"}, status=400
+            )
+
+        result = self.respond(
+            request_id=request_id,
+            approved=bool(approved),
+            decided_by=body.get("decided_by", "human"),
+            reason=body.get("reason", ""),
+        )
+
+        if not result:
+            return web.json_response({"error": "unknown request"}, status=404)
+
+        return web.json_response({
+            "request_id": result.request_id,
+            "status": result.status,
+        })
+
+    async def _handle_pending(self, request):
+        from aiohttp import web
+        return web.json_response({"pending": self.get_pending()})
+
+    def _build_app(self):
+        """Build the aiohttp application."""
+        from aiohttp import web
+
+        app = web.Application()
+        app.router.add_get("/health", self._handle_health)
+        app.router.add_post("/confirm", self._handle_confirm)
+        app.router.add_post("/respond", self._handle_respond)
+        app.router.add_get("/pending", self._handle_pending)
+        self._app = app
+        return app
+
+    async def start_async(self) -> None:
+        """Start the daemon as an async server."""
+        from aiohttp import web
+
+        app = self._build_app()
+        self._runner = web.AppRunner(app)
+        await self._runner.setup()
+        site = web.TCPSite(self._runner, self.host, self.port)
+        await site.start()
+        logger.info("Confirmation daemon listening on %s:%d", self.host, self.port)
+
+    async def stop_async(self) -> None:
+        """Stop the daemon."""
+        if self._runner:
+            await self._runner.cleanup()
+            self._runner = None
+
+    def start(self) -> None:
+        """Start daemon in a background thread (blocking caller)."""
+        def _run():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            loop.run_until_complete(self.start_async())
+            loop.run_forever()
+
+        t = threading.Thread(target=_run, daemon=True, name="confirmation-daemon")
+        t.start()
+        logger.info("Confirmation daemon started in background thread")
+
+    def start_blocking(self) -> None:
+        """Start daemon and block (for standalone use)."""
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.start_async())
+        try:
+            loop.run_forever()
+        except KeyboardInterrupt:
+            pass
+        finally:
+            loop.run_until_complete(self.stop_async())
+
+
+# =========================================================================
+# Convenience API for agent integration
+# =========================================================================
+
+# Global singleton — initialized by gateway or CLI at startup
+_daemon: Optional[ConfirmationDaemon] = None
+
+
+def get_daemon() -> Optional[ConfirmationDaemon]:
+    """Get the global confirmation daemon instance."""
+    return _daemon
+
+
+def init_daemon(
+    host: str = "127.0.0.1",
+    port: int = 6000,
+    notify_callback: Optional[Callable] = None,
+) -> ConfirmationDaemon:
+    """Initialize the global confirmation daemon."""
+    global _daemon
+    _daemon = ConfirmationDaemon(
+        host=host, port=port, notify_callback=notify_callback
+    )
+    return _daemon
+
+
+def request_confirmation(
+    action: str,
+    description: str,
+    payload: Optional[Dict[str, Any]] = None,
+    risk_level: Optional[str] = None,
+    session_key: str = "",
+    timeout: int = 300,
+) -> bool:
+    """Request human confirmation for a high-risk action.
+
+    This is the primary integration point for agent code. It:
+    1. Classifies the action risk level
+    2. Checks the whitelist
+    3. If confirmation needed, blocks until human responds
+    4. Returns True if approved, False if denied/expired
+
+    Args:
+        action: Action type (send_email, crypto_tx, etc.)
+        description: Human-readable description
+        payload: Action-specific data
+        risk_level: Override auto-classification
+        session_key: Session requesting approval
+        timeout: Seconds to wait for human response
+
+    Returns:
+        True if approved, False if denied or expired.
+    """
+    daemon = get_daemon()
+    if not daemon:
+        logger.warning(
+            "No confirmation daemon running — DENYING action %s by default. "
+            "Start daemon with init_daemon() or --confirmation-daemon flag.",
+            action,
+        )
+        return False
+
+    req = daemon.request(
+        action=action,
+        description=description,
+        payload=payload,
+        risk_level=risk_level,
+        session_key=session_key,
+        timeout=timeout,
+    )
+
+    # Auto-approved (whitelisted)
+    if req.status == ConfirmationStatus.AUTO_APPROVED.value:
+        return True
+
+    # Wait for human
+    result = daemon.wait_for_decision(req.request_id, timeout=timeout)
+    return result.status == ConfirmationStatus.APPROVED.value
Author	SHA1	Message	Date
Alexander Whitestone	3238cf4eb1	feat: Tool investigation report + Mem0 local provider (#842 ) Some checks failed Contributor Attribution Check / check-attribution (pull_request) Successful in 38s Details Docker Build and Publish / build-and-push (pull_request) Has been skipped Details Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 32s Details Tests / test (pull_request) Failing after 43m54s Details Tests / e2e (pull_request) Successful in 2m5s Details ## Investigation Report - docs/tool-investigation-2026-04-15.md: Full report analyzing 414 tools from awesome-ai-tools. Top 5 recommendations with integration paths. - docs/plans/awesome-ai-tools-integration.md: Implementation tracking plan. ## Mem0 Local Provider (P1) - plugins/memory/mem0_local/: New ChromaDB-backed memory provider. No API key required - fully sovereign. Compatible tool schemas with cloud Mem0 (mem0_profile, mem0_search, mem0_conclude). - Pattern-based fact extraction from conversations. - Deterministic dedup via content hashing. - Circuit breaker for resilience. - tests/plugins/memory/test_mem0_local.py: Full test coverage. ## Issues Filed - #857: LightRAG integration (P2) - #858: n8n workflow orchestration (P3) - #859: RAGFlow document understanding (P4) - #860: tensorzero LLMOps evaluation (P3) Closes #842	2026-04-15 23:04:41 -04:00
Alexander Whitestone	db72e908f7	Merge pull request 'feat(security): implement Vitalik's secure LLM patterns — privacy filter + confirmation daemon [resolves merge conflict]' (#830 ) from feat/vitalik-secure-llm-1776303263 into main Vitalik's secure LLM patterns — privacy filter + confirmation daemon Clean rebase of #397 onto current main. Resolves merge conflicts in tools/approval.py.	2026-04-16 01:36:58 +00:00
Alexander Whitestone	b82b760d5d	feat: add Vitalik's threat model patterns to DANGEROUS_PATTERNS Some checks failed Docker Build and Publish / build-and-push (pull_request) Has been skipped Details Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 41s Details Contributor Attribution Check / check-attribution (pull_request) Successful in 51s Details Tests / e2e (pull_request) Successful in 5m21s Details Tests / test (pull_request) Failing after 45m7s Details	2026-04-16 01:35:49 +00:00
Alexander Whitestone	d8d7846897	feat: add tests/tools/test_confirmation_daemon.py from PR #397	2026-04-16 01:35:24 +00:00
Alexander Whitestone	6840d05554	feat: add tests/agent/test_privacy_filter.py from PR #397	2026-04-16 01:35:21 +00:00
Alexander Whitestone	8abe59ed95	feat: add tools/confirmation_daemon.py from PR #397	2026-04-16 01:35:18 +00:00
Alexander Whitestone	435d790201	feat: add agent/privacy_filter.py from PR #397	2026-04-16 01:35:14 +00:00