feat: llama-server health monitor with auto-restart (#713)

llama-server on port 8081 was down with no monitoring or auto-restart.

New scripts/llama_health_monitor.py:
- check_server_health(): HTTP health check for any inference server
- check_ollama() / check_llama_server(): typed health checks
- find_llama_server_process(): find running process via ps
- restart_llama_server(): kill existing + start new with params
- _find_hermes3_model(): auto-discover model file path
- check_all_servers(): check Ollama + llama-server together
- format_status(): human-readable health report
- --watch mode: continuous monitoring with down alerts

CLI:
  python3 llama_health_monitor.py --check
  python3 llama_health_monitor.py --check --port 8081
  python3 llama_health_monitor.py --restart 8081
  python3 llama_health_monitor.py --watch
  python3 llama_health_monitor.py --report

Closes #713

agent/privacy_filter.py

@@ -1,353 +0,0 @@
-"""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

scripts/llama_health_monitor.py

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""llama-server health monitor — check, restart, and report on local inference.
+
+Monitors local inference servers (llama-server, Ollama) and can auto-restart
+them when they go down.
+
+Usage:
+    python3 scripts/llama_health_monitor.py --check              # check all
+    python3 scripts/llama_health_monitor.py --check --port 8081  # check specific
+    python3 scripts/llama_health_monitor.py --restart 8081       # restart server
+    python3 scripts/llama_health_monitor.py --watch              # continuous monitor
+    python3 scripts/llama_health_monitor.py --report             # JSON status report
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, asdict
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Default servers to monitor
+_DEFAULT_SERVERS = [
+    {"name": "ollama", "port": 11434, "type": "ollama", "health_path": "/api/tags"},
+    {"name": "llama-server", "port": 8081, "type": "llama-server", "health_path": "/health"},
+]
+
+
+@dataclass
+class ServerStatus:
+    """Status of a single inference server."""
+    name: str
+    port: int
+    server_type: str
+    reachable: bool
+    health_ok: bool
+    latency_ms: int
+    models: List[str]
+    error: str
+    checked_at: str
+
+
+def check_server_health(host: str = "localhost", port: int = 8081, health_path: str = "/health", timeout: int = 5) -> dict:
+    """Check if a server is healthy.
+
+    Returns dict with reachable, health_ok, latency_ms, models, error.
+    """
+    import urllib.request
+    import urllib.error
+
+    result = {
+        "reachable": False,
+        "health_ok": False,
+        "latency_ms": 0,
+        "models": [],
+        "error": "",
+    }
+
+    url = f"http://{host}:{port}{health_path}"
+    t0 = time.monotonic()
+
+    try:
+        req = urllib.request.Request(url, method="GET")
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            result["latency_ms"] = int((time.monotonic() - t0) * 1000)
+            result["reachable"] = True
+
+            if resp.status == 200:
+                result["health_ok"] = True
+                try:
+                    data = json.loads(resp.read())
+                    if isinstance(data, dict):
+                        result["models"] = [
+                            m.get("name", m.get("id", ""))
+                            for m in data.get("data", data.get("models", []))
+                        ]
+                except Exception:
+                    pass
+    except urllib.error.URLError as e:
+        result["latency_ms"] = int((time.monotonic() - t0) * 1000)
+        result["error"] = f"Connection refused or unreachable: {e}"
+    except Exception as e:
+        result["latency_ms"] = int((time.monotonic() - t0) * 1000)
+        result["error"] = str(e)
+
+    return result
+
+
+def check_ollama(port: int = 11434) -> ServerStatus:
+    """Check Ollama server status."""
+    import datetime
+    health = check_server_health(port=port, health_path="/api/tags")
+    return ServerStatus(
+        name="ollama",
+        port=port,
+        server_type="ollama",
+        reachable=health["reachable"],
+        health_ok=health["health_ok"],
+        latency_ms=health["latency_ms"],
+        models=health["models"],
+        error=health["error"],
+        checked_at=datetime.datetime.now().isoformat(),
+    )
+
+
+def check_llama_server(port: int = 8081) -> ServerStatus:
+    """Check llama-server status."""
+    import datetime
+    health = check_server_health(port=port, health_path="/health")
+    return ServerStatus(
+        name="llama-server",
+        port=port,
+        server_type="llama-server",
+        reachable=health["reachable"],
+        health_ok=health["health_ok"],
+        latency_ms=health["latency_ms"],
+        models=health.get("models", []),
+        error=health["error"],
+        checked_at=datetime.datetime.now().isoformat(),
+    )
+
+
+def find_llama_server_process() -> Optional[dict]:
+    """Find running llama-server process."""
+    try:
+        result = subprocess.run(
+            ["ps", "aux"],
+            capture_output=True, text=True, timeout=5,
+        )
+        for line in result.stdout.split("\n"):
+            if "llama-server" in line and "grep" not in line:
+                parts = line.split()
+                if len(parts) >= 11:
+                    return {
+                        "pid": int(parts[1]),
+                        "cpu": parts[2],
+                        "mem": parts[3],
+                        "command": " ".join(parts[10:]),
+                    }
+    except Exception:
+        pass
+    return None
+
+
+def restart_llama_server(
+    model_path: str = "",
+    port: int = 8081,
+    ctx_size: int = 8192,
+    gpu_layers: int = 99,
+    alias: str = "hermes3",
+) -> dict:
+    """Restart llama-server with specified parameters."""
+    # Kill existing process
+    existing = find_llama_server_process()
+    if existing:
+        try:
+            os.kill(existing["pid"], 15)  # SIGTERM
+            time.sleep(2)
+            logger.info("Killed existing llama-server (PID %d)", existing["pid"])
+        except ProcessLookupError:
+            pass
+        except Exception as e:
+            return {"success": False, "error": f"Failed to kill existing: {e}"}
+
+    # Find model path if not specified
+    if not model_path:
+        model_path = _find_hermes3_model()
+        if not model_path:
+            return {"success": False, "error": "Could not find hermes3 model path"}
+
+    # Build command
+    cmd = [
+        "llama-server",
+        "--model", model_path,
+        "--port", str(port),
+        "--host", "127.0.0.1",
+        "--n-gpu-layers", str(gpu_layers),
+        "--flash-attn", "on",
+        "--ctx-size", str(ctx_size),
+        "--alias", alias,
+    ]
+
+    try:
+        # Start in background
+        proc = subprocess.Popen(
+            cmd,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        time.sleep(3)  # Wait for startup
+
+        # Verify it's running
+        health = check_server_health(port=port)
+        if health["reachable"]:
+            return {
+                "success": True,
+                "pid": proc.pid,
+                "port": port,
+                "model": model_path,
+            }
+        else:
+            return {
+                "success": False,
+                "error": f"Started but not reachable: {health['error']}",
+                "pid": proc.pid,
+            }
+    except FileNotFoundError:
+        return {"success": False, "error": "llama-server binary not found in PATH"}
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+def _find_hermes3_model() -> str:
+    """Try to find the hermes3 model file."""
+    import glob
+    search_paths = [
+        os.path.expanduser("~/.ollama/models/blobs/sha256-*"),
+        os.path.expanduser("~/.cache/llama.cpp/*.gguf"),
+        "/opt/models/*.gguf",
+    ]
+    for pattern in search_paths:
+        matches = glob.glob(pattern)
+        if matches:
+            return matches[0]
+    return ""
+
+
+def check_all_servers() -> List[ServerStatus]:
+    """Check all configured servers."""
+    results = []
+    results.append(check_ollama())
+    results.append(check_llama_server())
+    return results
+
+
+def format_status(statuses: List[ServerStatus]) -> str:
+    """Format server statuses as a report."""
+    lines = ["Local Inference Health", "=" * 40, ""]
+
+    for s in statuses:
+        icon = "\u2705" if s.reachable and s.health_ok else "\u274c"
+        lines.append(f"{icon} {s.name} (port {s.port})")
+        lines.append(f"   Type:      {s.server_type}")
+        lines.append(f"   Reachable: {s.reachable}")
+        lines.append(f"   Healthy:   {s.health_ok}")
+        lines.append(f"   Latency:   {s.latency_ms}ms")
+        if s.models:
+            lines.append(f"   Models:    {', '.join(s.models[:5])}")
+        if s.error:
+            lines.append(f"   Error:     {s.error[:100]}")
+        lines.append("")
+
+    # llama-server process
+    proc = find_llama_server_process()
+    if proc:
+        lines.append(f"llama-server process: PID {proc['pid']}, CPU {proc['cpu']}%, MEM {proc['mem']}%")
+    else:
+        lines.append("llama-server process: NOT RUNNING")
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Local inference health monitor")
+    parser.add_argument("--check", action="store_true", help="Check server health")
+    parser.add_argument("--port", type=int, default=0, help="Check specific port")
+    parser.add_argument("--restart", type=int, default=0, metavar="PORT", help="Restart server on port")
+    parser.add_argument("--watch", action="store_true", help="Continuous monitoring")
+    parser.add_argument("--report", action="store_true", help="JSON status report")
+    parser.add_argument("--interval", type=int, default=30, help="Watch interval in seconds")
+    args = parser.parse_args()
+
+    if args.restart:
+        print(f"Restarting llama-server on port {args.restart}...")
+        result = restart_llama_server(port=args.restart)
+        print(json.dumps(result, indent=2))
+        return 0 if result["success"] else 1
+
+    if args.watch:
+        print(f"Monitoring every {args.interval}s. Press Ctrl+C to stop.")
+        while True:
+            statuses = check_all_servers()
+            down = [s for s in statuses if not s.reachable]
+            if down:
+                for s in down:
+                    print(f"[{time.strftime('%H:%M:%S')}] DOWN: {s.name} port {s.port} — {s.error}")
+            else:
+                print(f"[{time.strftime('%H:%M:%S')}] All servers OK")
+            time.sleep(args.interval)
+        return 0
+
+    if args.port:
+        status = check_llama_server(port=args.port)
+        statuses = [status]
+    else:
+        statuses = check_all_servers()
+
+    if args.report:
+        print(json.dumps([asdict(s) for s in statuses], indent=2))
+    else:
+        print(format_status(statuses))
+
+    down = [s for s in statuses if not s.reachable]
+    return 1 if down else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/agent/test_privacy_filter.py

@@ -1,202 +0,0 @@
-"""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

tests/test_parallel_tool_calling.py

@@ -1,418 +0,0 @@
-#!/usr/bin/env python3
-"""
-test_parallel_tool_calling.py — Tests for parallel tool calling (2+ tools per response).
-
-Verifies that hermes-agent correctly handles multiple tool calls in a single
-response, including ordering, dependency resolution, and parallel safety.
-
-Issue #798: Gemma 4 Tool Calling Hardening
-"""
-
-import json
-import os
-import sys
-import pytest
-from dataclasses import dataclass
-from pathlib import Path
-from unittest.mock import MagicMock, patch, call
-
-# Add project root to path
-sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
-
-from run_agent import (
-    _should_parallelize_tool_batch,
-    _extract_parallel_scope_path,
-    _is_destructive_command,
-    _PARALLEL_SAFE_TOOLS,
-    _NEVER_PARALLEL_TOOLS,
-    _PATH_SCOPED_TOOLS,
-)
-
-
-# ── Mock Tool Call Structure ──────────────────────────────────────────────────
-
-@dataclass
-class MockFunction:
-    name: str
-    arguments: str
-
-@dataclass
-class MockToolCall:
-    id: str
-    function: MockFunction
-
-    @classmethod
-    def make(cls, name: str, args: dict, idx: int = 0):
-        return cls(
-            id=f"call_{idx}",
-            function=MockFunction(name=name, arguments=json.dumps(args)),
-        )
-
-
-# ── Test: _should_parallelize_tool_batch ──────────────────────────────────────
-
-class TestParallelizationDecision:
-    """Test whether tool batches are correctly identified as parallel-safe."""
-
-    def test_single_tool_not_parallel(self):
-        """A single tool call should never be parallelized."""
-        calls = [MockToolCall.make("read_file", {"path": "a.txt"})]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_two_read_files_different_paths(self):
-        """Two read_file calls on different paths should parallelize."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "b.txt"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_two_read_files_same_path(self):
-        """Two read_file calls on the same path should NOT parallelize."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "a.txt"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_read_plus_search_parallel(self):
-        """read_file + search_files should parallelize (both safe, different scopes)."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("search_files", {"pattern": "foo"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_clarify_never_parallel(self):
-        """clarify tool should block parallelization."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("clarify", {"question": "what?"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_three_read_files_all_different(self):
-        """Three read_file calls on different paths should parallelize."""
-        calls = [
-            MockToolCall.make("read_file", {"path": f"file{i}.txt"}, i)
-            for i in range(3)
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_write_plus_read_same_path(self):
-        """write_file + read_file on same path should NOT parallelize."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("write_file", {"path": "a.txt", "content": "new"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_write_plus_read_different_paths(self):
-        """write_file + read_file on different paths should parallelize."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("write_file", {"path": "b.txt", "content": "new"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_unsafe_tool_blocks_parallel(self):
-        """A tool not in _PARALLEL_SAFE_TOOLS or _PATH_SCOPED_TOOLS blocks parallel."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("some_unknown_tool", {"param": "value"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_all_safe_tools(self):
-        """All tools in _PARALLEL_SAFE_TOOLS should parallelize together."""
-        calls = [
-            MockToolCall.make("web_search", {"query": "test"}, 0),
-            MockToolCall.make("session_search", {"query": "test"}, 1),
-            MockToolCall.make("skills_list", {}, 2),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_malformed_json_args(self):
-        """Malformed JSON arguments should block parallelization."""
-        tc = MockToolCall(id="call_0", function=MockFunction(
-            name="read_file", arguments="not json"
-        ))
-        calls = [MockToolCall.make("read_file", {"path": "a.txt"}, 1), tc]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_non_dict_args(self):
-        """Non-dict arguments should block parallelization."""
-        tc = MockToolCall(id="call_0", function=MockFunction(
-            name="read_file", arguments='"just a string"'
-        ))
-        calls = [MockToolCall.make("read_file", {"path": "a.txt"}, 1), tc]
-        assert _should_parallelize_tool_batch(calls) is False
-
-
-# ── Test: Path Scope Extraction ──────────────────────────────────────────────
-
-class TestPathScopeExtraction:
-    """Test path extraction for scoped parallel tools."""
-
-    def test_relative_path(self):
-        result = _extract_parallel_scope_path("read_file", {"path": "foo/bar.txt"})
-        assert result is not None
-        assert "bar.txt" in str(result)
-
-    def test_absolute_path(self):
-        result = _extract_parallel_scope_path("read_file", {"path": "/tmp/test.txt"})
-        assert result == Path("/tmp/test.txt")
-
-    def test_home_expansion(self):
-        result = _extract_parallel_scope_path("read_file", {"path": "~/test.txt"})
-        assert result is not None
-        assert str(result).endswith("test.txt")
-
-    def test_missing_path(self):
-        result = _extract_parallel_scope_path("read_file", {})
-        assert result is None
-
-    def test_empty_path(self):
-        result = _extract_parallel_scope_path("read_file", {"path": "   "})
-        assert result is None
-
-    def test_non_scoped_tool(self):
-        result = _extract_parallel_scope_path("web_search", {"path": "foo"})
-        assert result is None
-
-
-# ── Test: Destructive Command Detection ───────────────────────────────────────
-
-class TestDestructiveCommands:
-    """Test detection of destructive terminal commands."""
-
-    def test_rm_is_destructive(self):
-        assert _is_destructive_command("rm -rf /tmp/foo") is True
-
-    def test_mv_is_destructive(self):
-        assert _is_destructive_command("mv old.txt new.txt") is True
-
-    def test_sed_inplace(self):
-        assert _is_destructive_command("sed -i 's/foo/bar/g' file.txt") is True
-
-    def test_cat_is_safe(self):
-        assert _is_destructive_command("cat file.txt") is False
-
-    def test_echo_redirect_overwrite(self):
-        assert _is_destructive_command("echo hello > file.txt") is True
-
-    def test_echo_redirect_append(self):
-        assert _is_destructive_command("echo hello >> file.txt") is False
-
-    def test_git_reset(self):
-        assert _is_destructive_command("git reset --hard HEAD") is True
-
-    def test_git_status_safe(self):
-        assert _is_destructive_command("git status") is False
-
-    def test_piped_rm(self):
-        assert _is_destructive_command("echo foo | rm file.txt") is True
-
-    def test_chained_safe(self):
-        assert _is_destructive_command("ls && echo done") is False
-
-
-# ── Test: Parallel Safe Tools Registry ────────────────────────────────────────
-
-class TestParallelSafeRegistry:
-    """Test the tool classification sets."""
-
-    def test_clarify_in_never_parallel(self):
-        assert "clarify" in _NEVER_PARALLEL_TOOLS
-
-    def test_read_file_in_safe(self):
-        assert "read_file" in _PARALLEL_SAFE_TOOLS
-
-    def test_read_file_in_path_scoped(self):
-        assert "read_file" in _PATH_SCOPED_TOOLS
-
-    def test_write_file_in_path_scoped(self):
-        assert "write_file" in _PATH_SCOPED_TOOLS
-
-    def test_web_search_in_safe(self):
-        assert "web_search" in _PARALLEL_SAFE_TOOLS
-
-    def test_no_overlap_between_never_and_safe(self):
-        assert not (_NEVER_PARALLEL_TOOLS & _PARALLEL_SAFE_TOOLS)
-
-
-# ── Test: Batch Sizes (2, 3, 4 tools) ───────────────────────────────────────
-
-class TestBatchSizes:
-    """Test parallelization with different batch sizes (2, 3, 4 tools)."""
-
-    def test_two_tool_batch(self):
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "b.txt"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_three_tool_batch(self):
-        calls = [
-            MockToolCall.make("read_file", {"path": f"f{i}.txt"}, i)
-            for i in range(3)
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_four_tool_batch(self):
-        calls = [
-            MockToolCall.make("web_search", {"query": f"q{i}"}, i)
-            for i in range(4)
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_four_tool_batch_with_one_collision(self):
-        """4 tools where 2 collide on the same path."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "b.txt"}, 1),
-            MockToolCall.make("read_file", {"path": "a.txt"}, 2),  # collision
-            MockToolCall.make("read_file", {"path": "c.txt"}, 3),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-
-# ── Test: Gemma 4 Specific Patterns ──────────────────────────────────────────
-
-class TestGemma4Patterns:
-    """
-    Test patterns specific to Gemma 4 tool calling behavior.
-
-    Gemma 4 may issue tool calls in specific ordering patterns that
-    need to be handled correctly by the parallel execution layer.
-    """
-
-    def test_gemma4_typical_2tool_pattern(self):
-        """Gemma 4 typically issues read+search as a pair."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "config.yaml"}, 0),
-            MockToolCall.make("search_files", {"pattern": "provider"}, 1),
-        ]
-        # These should parallelize — different tools, no path conflict
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_gemma4_typical_3tool_pattern(self):
-        """Gemma 4 may issue 3 reads for different files."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.py"}, 0),
-            MockToolCall.make("read_file", {"path": "b.py"}, 1),
-            MockToolCall.make("read_file", {"path": "c.py"}, 2),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_gemma4_sequential_dependency(self):
-        """
-        Gemma 4 may issue: search_files then read_file on search result.
-        These have implicit dependency but are issued as a batch.
-        The agent should handle this — search first, then read.
-        This test verifies the batch IS marked as parallel-safe
-        (ordering is the agent loop's responsibility, not this function's).
-        """
-        calls = [
-            MockToolCall.make("search_files", {"pattern": "import"}, 0),
-            MockToolCall.make("read_file", {"path": "main.py"}, 1),
-        ]
-        # Both tools are in safe/scoped sets with no path conflict
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_gemma4_mixed_safe_unsafe(self):
-        """Gemma 4 may mix read (safe) with write (path-scoped)."""
-        calls = [
-            MockToolCall.make("read_file", {"path": "input.txt"}, 0),
-            MockToolCall.make("write_file", {"path": "output.txt", "content": "x"}, 1),
-            MockToolCall.make("read_file", {"path": "config.txt"}, 2),
-        ]
-        # All path-scoped on different paths, no unsafe tools
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_gemma4_terminal_parallel(self):
-        """
-        Terminal commands are NOT in _PARALLEL_SAFE_TOOLS.
-        If Gemma 4 issues 2 terminal calls, they should NOT parallelize.
-        """
-        calls = [
-            MockToolCall.make("terminal", {"command": "ls"}, 0),
-            MockToolCall.make("terminal", {"command": "pwd"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-
-# ── Test: Integration-style (mocked) ─────────────────────────────────────────
-
-class TestParallelExecutionMocked:
-    """Test the parallel execution path with mocked tool handlers."""
-
-    def test_parallel_results_collected(self):
-        """Simulate parallel execution and verify results are collected."""
-        # Mock two tool calls returning different results
-        results = {}
-
-        def mock_handler(name, args):
-            return f"result_{name}_{args.get('path', 'x')}"
-
-        calls = [
-            MockToolCall.make("read_file", {"path": "a.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "b.txt"}, 1),
-        ]
-
-        # Simulate parallel execution
-        for tc in calls:
-            results[tc.id] = mock_handler(tc.function.name,
-                                          json.loads(tc.function.arguments))
-
-        assert results["call_0"] == "result_read_file_a.txt"
-        assert results["call_1"] == "result_read_file_b.txt"
-
-    def test_parallel_results_order_preserved(self):
-        """Results should be ordered by tool call ID, not completion time."""
-        import time
-        results = {}
-
-        calls = [
-            MockToolCall.make("read_file", {"path": "slow.txt"}, 0),
-            MockToolCall.make("read_file", {"path": "fast.txt"}, 1),
-        ]
-
-        # Simulate out-of-order completion
-        results["call_1"] = "fast_result"
-        results["call_0"] = "slow_result"
-
-        # Verify we can reconstruct in order
-        ordered = [results[tc.id] for tc in calls]
-        assert ordered == ["slow_result", "fast_result"]
-
-
-# ── Test: Edge Cases ──────────────────────────────────────────────────────────
-
-class TestEdgeCases:
-    """Edge cases for parallel tool calling."""
-
-    def test_empty_batch(self):
-        assert _should_parallelize_tool_batch([]) is False
-
-    def test_patch_with_same_path(self):
-        """Two patch calls on the same file should NOT parallelize."""
-        calls = [
-            MockToolCall.make("patch", {"path": "a.py", "old_string": "x", "new_string": "y"}, 0),
-            MockToolCall.make("patch", {"path": "a.py", "old_string": "a", "new_string": "b"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is False
-
-    def test_patch_different_paths(self):
-        """patch on different files should parallelize."""
-        calls = [
-            MockToolCall.make("patch", {"path": "a.py", "old_string": "x", "new_string": "y"}, 0),
-            MockToolCall.make("patch", {"path": "b.py", "old_string": "a", "new_string": "b"}, 1),
-        ]
-        assert _should_parallelize_tool_batch(calls) is True
-
-    def test_max_workers_defined(self):
-        """Verify max workers constant exists and is reasonable."""
-        from run_agent import _MAX_TOOL_WORKERS
-        assert 1 <= _MAX_TOOL_WORKERS <= 32

tests/tools/test_confirmation_daemon.py

@@ -1,190 +0,0 @@
-"""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"

tools/approval.py

@@ -121,19 +121,6 @@ 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"),

tools/confirmation_daemon.py

@@ -1,615 +0,0 @@
-"""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