fix(memory): filter noisy prefetch results before injection (#982)

Add line-level filtering to MemoryManager.prefetch_all() that removes:
- exact duplicate lines
- meta-instruction debris (prompts, templates like 'Focus on:')
- very short / content-free fragments (< 15 chars after markers)

This prevents noisy memory providers (e.g. MemPalace transcript recall)
from bloating the <memory-context> block with repeated instructions and
low-signal snippets.

Changes:
- agent/memory_manager.py: add _filter_prefetch_lines(),
  _is_meta_instruction_line(), _is_low_signal_line() helpers;
  wire filtering into prefetch_all()
- tests/agent/test_memory_provider.py: add TestPrefetchFiltering with
  8 tests covering dedup, meta-instruction removal, low-signal filtering,
  structured-fact preservation, and integration with prefetch_all()

Closes #982

.githooks/pre-commit-hardcoded-path.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""
+Pre-commit hook: Reject hardcoded home-directory paths.
+
+Install:
+    cp pre-commit-hardcoded-path.py .git/hooks/pre-commit-hardcoded-path
+    chmod +x .git/hooks/pre-commit-hardcoded-path
+    
+    Or add to .pre-commit-config.yaml
+"""
+
+import sys
+import subprocess
+import re
+
+PATTERNS = [
+    (r"/Users/[\w.\-]+/", "macOS home directory"),
+    (r"/home/[\w.\-]+/", "Linux home directory"),
+    (r"(?<![\w/])~/", "unexpanded tilde"),
+]
+
+NOQA = re.compile(r"#\s*noqa:?\s*hardcoded-path-ok")
+
+def get_staged_files():
+    result = subprocess.run(
+        ["git", "diff", "--cached", "--name-only", "--diff-filter=ACM"],
+        capture_output=True, text=True
+    )
+    return [f for f in result.stdout.strip().split("\n") if f.endswith(".py")]
+
+def check_file(filepath):
+    try:
+        result = subprocess.run(
+            ["git", "show", f":{filepath}"],
+            capture_output=True, text=True
+        )
+        content = result.stdout
+    except Exception:
+        return []
+    
+    violations = []
+    for i, line in enumerate(content.split("\n"), 1):
+        if line.strip().startswith("#"):
+            continue
+        if line.strip().startswith(("import ", "from ")):
+            continue
+        if NOQA.search(line):
+            continue
+        for pattern, desc in PATTERNS:
+            if re.search(pattern, line):
+                violations.append((filepath, i, line.strip(), desc))
+                break
+    return violations
+
+def main():
+    files = get_staged_files()
+    if not files:
+        sys.exit(0)
+    
+    all_violations = []
+    for f in files:
+        all_violations.extend(check_file(f))
+    
+    if all_violations:
+        print("ERROR: Hardcoded home directory paths detected:")
+        print()
+        for filepath, line_no, line, desc in all_violations:
+            print(f"  {filepath}:{line_no}: {desc}")
+            print(f"    {line[:100]}")
+            print()
+        print("Fix: Use $HOME, relative paths, or get_hermes_home().")
+        print("Override: Add '# noqa: hardcoded-path-ok' to the line.")
+        sys.exit(1)
+    
+    sys.exit(0)
+
+if __name__ == "__main__":
+    main()

agent/a2a_mtls.py

@@ -0,0 +1,443 @@
+"""
+A2A mutual-TLS server — secure agent-to-agent communication.
+
+Each fleet agent runs an A2A server that:
+  - Presents its own TLS certificate (signed by the fleet CA).
+  - Requires the connecting peer to present a valid client certificate
+    also signed by the fleet CA.
+  - Rejects connections from unknown / self-signed peers.
+
+Usage (standalone):
+    python -m agent.a2a_mtls \\
+        --cert ~/.hermes/pki/agents/timmy/timmy.crt \\
+        --key  ~/.hermes/pki/agents/timmy/timmy.key \\
+        --ca   ~/.hermes/pki/ca/fleet-ca.crt \\
+        --host 0.0.0.0 --port 9443
+
+Environment variables (alternative to CLI flags):
+    HERMES_A2A_CERT   path to agent certificate
+    HERMES_A2A_KEY    path to agent private key
+    HERMES_A2A_CA     path to fleet CA certificate
+
+Refs #806
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import ssl
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from typing import Any, Callable, Dict, Optional
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# mTLS SSL context helpers
+# ---------------------------------------------------------------------------
+
+def build_server_ssl_context(
+    cert: str | Path,
+    key: str | Path,
+    ca: str | Path,
+) -> ssl.SSLContext:
+    """Return an SSLContext that presents *cert/key* and requires a valid
+    client certificate signed by *ca*.
+
+    Raises ``FileNotFoundError`` if any path is missing.
+    Raises ``ssl.SSLError`` if the files are malformed.
+    """
+    cert, key, ca = Path(cert), Path(key), Path(ca)
+    for p in (cert, key, ca):
+        if not p.exists():
+            raise FileNotFoundError(f"mTLS: file not found: {p}")
+
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+    ctx.load_cert_chain(certfile=str(cert), keyfile=str(key))
+    ctx.load_verify_locations(cafile=str(ca))
+    # CERT_REQUIRED — reject peers that don't present a cert signed by *ca*.
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    return ctx
+
+
+def build_client_ssl_context(
+    cert: str | Path,
+    key: str | Path,
+    ca: str | Path,
+) -> ssl.SSLContext:
+    """Return an SSLContext for an outgoing mTLS connection.
+
+    Presents *cert/key* as the client identity and verifies the server
+    certificate against *ca*.
+    """
+    cert, key, ca = Path(cert), Path(key), Path(ca)
+    for p in (cert, key, ca):
+        if not p.exists():
+            raise FileNotFoundError(f"mTLS client: file not found: {p}")
+
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+    ctx.load_cert_chain(certfile=str(cert), keyfile=str(key))
+    ctx.load_verify_locations(cafile=str(ca))
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    ctx.check_hostname = True
+    return ctx
+
+
+# ---------------------------------------------------------------------------
+# Minimal A2A HTTP request handler
+# ---------------------------------------------------------------------------
+
+class A2AHandler(BaseHTTPRequestHandler):
+    """Handles A2A requests over a mutually-authenticated TLS connection.
+
+    GET /.well-known/agent-card.json  — returns the local agent card.
+    POST /a2a/task                    — dispatches an A2A task (stub).
+    """
+
+    log_message = logger.debug  # route access log to Python logger
+
+    def do_GET(self) -> None:  # noqa: N802
+        if self.path in ("/.well-known/agent-card.json", "/agent-card.json"):
+            self._serve_agent_card()
+        else:
+            self._send_json(404, {"error": "not found"})
+
+    def do_POST(self) -> None:  # noqa: N802
+        if self.path == "/a2a/task":
+            self._handle_task()
+        else:
+            self._send_json(404, {"error": "not found"})
+
+    # ------------------------------------------------------------------
+    def _serve_agent_card(self) -> None:
+        try:
+            from agent.agent_card import get_agent_card_json
+            body = get_agent_card_json().encode()
+        except Exception as exc:
+            logger.warning("agent-card unavailable: %s", exc)
+            body = b'{"error": "agent card unavailable"}'
+        self._send_raw(200, "application/json", body)
+
+    def _handle_task(self) -> None:
+        length = int(self.headers.get("Content-Length", 0))
+        _body = self.rfile.read(length) if length else b""
+        # Stub: echo back a 202 Accepted with the peer CN so callers can
+        # confirm which agent processed the request.
+        peer_cn = _peer_cn(self.connection)
+        self._send_json(202, {"status": "accepted", "handled_by": peer_cn})
+
+    # ------------------------------------------------------------------
+    def _send_json(self, code: int, data: dict) -> None:
+        import json
+        body = json.dumps(data).encode()
+        self._send_raw(code, "application/json", body)
+
+    def _send_raw(self, code: int, content_type: str, body: bytes) -> None:
+        self.send_response(code)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, fmt: str, *args: object) -> None:  # type: ignore[override]
+        logger.debug("a2a: " + fmt, *args)
+
+
+def _peer_cn(conn: ssl.SSLSocket) -> Optional[str]:
+    """Extract the Common Name from the peer certificate, or None."""
+    try:
+        peer = conn.getpeercert()
+        if not peer:
+            return None
+        for rdn in peer.get("subject", ()):
+            for key, val in rdn:
+                if key == "commonName":
+                    return val
+    except Exception:
+        pass
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Server lifecycle
+# ---------------------------------------------------------------------------
+
+class A2AServer:
+    """Mutual-TLS A2A server.
+
+    Example::
+
+        server = A2AServer(
+            cert="~/.hermes/pki/agents/timmy/timmy.crt",
+            key="~/.hermes/pki/agents/timmy/timmy.key",
+            ca="~/.hermes/pki/ca/fleet-ca.crt",
+        )
+        server.start()   # non-blocking (daemon thread)
+        ...
+        server.stop()
+    """
+
+    def __init__(
+        self,
+        cert: str | Path,
+        key: str | Path,
+        ca: str | Path,
+        host: str = "0.0.0.0",
+        port: int = 9443,
+    ) -> None:
+        self.cert = Path(cert).expanduser()
+        self.key = Path(key).expanduser()
+        self.ca = Path(ca).expanduser()
+        self.host = host
+        self.port = port
+        self._httpd: Optional[HTTPServer] = None
+        self._thread: Optional[threading.Thread] = None
+
+    def start(self, daemon: bool = True) -> None:
+        """Start the server in a background thread (default: daemon)."""
+        ssl_ctx = build_server_ssl_context(self.cert, self.key, self.ca)
+        self._httpd = HTTPServer((self.host, self.port), A2AHandler)
+        self._httpd.socket = ssl_ctx.wrap_socket(
+            self._httpd.socket, server_side=True
+        )
+        self._thread = threading.Thread(
+            target=self._httpd.serve_forever, daemon=daemon
+        )
+        self._thread.start()
+        logger.info(
+            "A2A mTLS server listening on %s:%s (cert=%s)",
+            self.host, self.port, self.cert.name,
+        )
+
+    def stop(self) -> None:
+        if self._httpd:
+            self._httpd.shutdown()
+            self._httpd = None
+        if self._thread:
+            self._thread.join(timeout=5)
+            self._thread = None
+
+
+def server_from_env() -> A2AServer:
+    """Build an A2AServer from environment variables / defaults."""
+    hermes_home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
+    agent_name = os.environ.get("HERMES_AGENT_NAME", "hermes").lower()
+
+    default_cert = hermes_home / "pki" / "agents" / agent_name / f"{agent_name}.crt"
+    default_key = hermes_home / "pki" / "agents" / agent_name / f"{agent_name}.key"
+    default_ca = hermes_home / "pki" / "ca" / "fleet-ca.crt"
+
+    cert = os.environ.get("HERMES_A2A_CERT", str(default_cert))
+    key = os.environ.get("HERMES_A2A_KEY", str(default_key))
+    ca = os.environ.get("HERMES_A2A_CA", str(default_ca))
+    host = os.environ.get("HERMES_A2A_HOST", "0.0.0.0")
+    port = int(os.environ.get("HERMES_A2A_PORT", "9443"))
+
+    return A2AServer(cert=cert, key=key, ca=ca, host=host, port=port)
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+def _main() -> None:
+    import argparse
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+    parser = argparse.ArgumentParser(
+        description="Hermes A2A mutual-TLS server"
+    )
+    parser.add_argument("--cert", required=True, help="Path to agent certificate")
+    parser.add_argument("--key", required=True, help="Path to agent private key")
+    parser.add_argument("--ca", required=True, help="Path to fleet CA certificate")
+    parser.add_argument("--host", default="0.0.0.0")
+    parser.add_argument("--port", type=int, default=9443)
+    args = parser.parse_args()
+
+    server = A2AServer(
+        cert=args.cert, key=args.key, ca=args.ca,
+        host=args.host, port=args.port,
+    )
+    server.start(daemon=False)
+
+
+if __name__ == "__main__":
+    _main()
+
+
+# ---------------------------------------------------------------------------
+# A2AMTLSServer — routing-based server with context-manager support
+# ---------------------------------------------------------------------------
+
+class _RoutingHandler(BaseHTTPRequestHandler):
+    """HTTP request handler that dispatches to per-path callables."""
+
+    routes: Dict[str, Callable] = {}
+
+    def log_message(self, fmt: str, *args: Any) -> None:
+        logger.debug("A2AMTLSServer: " + fmt, *args)
+
+    def _peer_cn(self) -> Optional[str]:
+        cert = self.connection.getpeercert()  # type: ignore[attr-defined]
+        if not cert:
+            return None
+        for rdn in cert.get("subject", ()):
+            for attr, value in rdn:
+                if attr == "commonName":
+                    return value
+        return None
+
+    def do_POST(self) -> None:
+        handler = self.routes.get(self.path)
+        if handler is None:
+            self.send_response(404)
+            self.end_headers()
+            return
+        length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(length) if length else b""
+        try:
+            payload = json.loads(body) if body else {}
+        except json.JSONDecodeError:
+            self.send_response(400)
+            self.end_headers()
+            return
+        result = handler(payload, peer_cn=self._peer_cn())
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps(result).encode())
+
+    def do_GET(self) -> None:
+        handler = self.routes.get(self.path)
+        if handler is None:
+            self.send_response(404)
+            self.end_headers()
+            return
+        result = handler({}, peer_cn=self._peer_cn())
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps(result).encode())
+
+
+class A2AMTLSServer:
+    """Routing-based mTLS HTTPS server with context-manager support.
+
+    Unlike ``A2AServer`` (which serves fixed A2A paths), this server lets
+    callers register arbitrary path handlers — useful for tests and custom
+    A2A endpoint implementations.
+
+    handler signature: ``handler(payload: dict, *, peer_cn: str | None) -> dict``
+
+    Example::
+
+        server = A2AMTLSServer(cert="timmy.crt", key="timmy.key", ca="fleet-ca.crt")
+        server.add_route("/tasks/send", my_handler)
+        with server:
+            ...  # server runs for the duration of the block
+    """
+
+    def __init__(
+        self,
+        cert: str | Path,
+        key: str | Path,
+        ca: str | Path,
+        host: str = "127.0.0.1",
+        port: int = 9443,
+    ) -> None:
+        self.cert = Path(cert).expanduser()
+        self.key = Path(key).expanduser()
+        self.ca = Path(ca).expanduser()
+        self.host = host
+        self.port = port
+        self._routes: Dict[str, Callable] = {}
+        self._httpd: Optional[HTTPServer] = None
+        self._thread: Optional[threading.Thread] = None
+
+    def add_route(self, path: str, handler: Callable) -> None:
+        self._routes[path] = handler
+
+    def start(self) -> None:
+        ssl_ctx = build_server_ssl_context(self.cert, self.key, self.ca)
+
+        class _Handler(_RoutingHandler):
+            routes = self._routes
+
+        self._httpd = HTTPServer((self.host, self.port), _Handler)
+        self._httpd.socket = ssl_ctx.wrap_socket(self._httpd.socket, server_side=True)
+        self._thread = threading.Thread(
+            target=self._httpd.serve_forever,
+            daemon=True,
+            name=f"a2a-mtls-{self.port}",
+        )
+        self._thread.start()
+        logger.info("A2AMTLSServer on %s:%d (mTLS)", self.host, self.port)
+
+    def stop(self) -> None:
+        if self._httpd:
+            self._httpd.shutdown()
+            self._httpd = None
+        if self._thread:
+            self._thread.join(timeout=5)
+            self._thread = None
+
+    def __enter__(self) -> "A2AMTLSServer":
+        self.start()
+        return self
+
+    def __exit__(self, *_: Any) -> None:
+        self.stop()
+
+
+# ---------------------------------------------------------------------------
+# A2AMTLSClient — mTLS HTTP client
+# ---------------------------------------------------------------------------
+
+class A2AMTLSClient:
+    """HTTP client that presents a fleet cert on every outgoing connection.
+
+    Example::
+
+        client = A2AMTLSClient(cert="allegro.crt", key="allegro.key", ca="fleet-ca.crt")
+        result = client.post("https://timmy:9443/tasks/send", json={"task": "..."})
+    """
+
+    def __init__(
+        self,
+        cert: str | Path,
+        key: str | Path,
+        ca: str | Path,
+    ) -> None:
+        self._ssl_ctx = build_client_ssl_context(cert, key, ca)
+        self._ssl_ctx.check_hostname = False  # callers connecting by IP
+
+    def _request(
+        self,
+        method: str,
+        url: str,
+        data: Optional[bytes] = None,
+        timeout: float = 10.0,
+    ) -> Dict[str, Any]:
+        headers = {"Content-Type": "application/json"}
+        req = Request(url, data=data, headers=headers, method=method)
+        try:
+            with urlopen(req, context=self._ssl_ctx, timeout=timeout) as resp:
+                body = resp.read()
+                return json.loads(body) if body else {}
+        except URLError as exc:
+            raise ConnectionError(f"A2AMTLSClient {method} {url} failed: {exc.reason}") from exc
+
+    def get(self, url: str, **kwargs: Any) -> Dict[str, Any]:
+        return self._request("GET", url, **kwargs)
+
+    def post(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Dict[str, Any]:
+        data = (__import__("json").dumps(json).encode() if json is not None else None)
+        return self._request("POST", url, data=data, **kwargs)

agent/circuit_breaker.py

@@ -0,0 +1,273 @@
+"""
+Circuit Breaker for Error Cascading — #885
+
+P(error | prev was error) = 58.6% vs P(error | prev was success) = 25.2%.
+That's a 2.33x cascade factor. After 3 consecutive errors, the circuit
+opens and the agent must take corrective action.
+
+States:
+- CLOSED: Normal operation, errors are counted
+- OPEN: Too many consecutive errors, corrective action required
+- HALF_OPEN: Testing if errors have cleared
+
+Usage:
+    from agent.circuit_breaker import CircuitBreaker, ToolCircuitBreaker
+    
+    cb = ToolCircuitBreaker()
+    
+    # After each tool call
+    if not cb.record_result(success=True):
+        # Circuit is open — take corrective action
+        cb.get_recovery_action()
+"""
+
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+
+class CircuitState(Enum):
+    CLOSED = "closed"      # Normal operation
+    OPEN = "open"          # Too many errors, block execution
+    HALF_OPEN = "half_open"  # Testing recovery
+
+
+@dataclass
+class CircuitBreaker:
+    """
+    Generic circuit breaker with configurable thresholds.
+    
+    Tracks consecutive errors and opens the circuit when the
+    error streak exceeds the threshold.
+    """
+    failure_threshold: int = 3
+    recovery_timeout: float = 30.0  # seconds before trying half-open
+    success_threshold: int = 2      # successes needed to close from half-open
+    
+    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
+    consecutive_failures: int = field(default=0, init=False)
+    consecutive_successes: int = field(default=0, init=False)
+    last_failure_time: Optional[float] = field(default=None, init=False)
+    total_trips: int = field(default=0, init=False)
+    error_streaks: List[int] = field(default_factory=list, init=False)
+    
+    def record_result(self, success: bool) -> bool:
+        """
+        Record a tool call result. Returns True if circuit allows execution.
+        
+        Returns:
+            True if circuit is CLOSED or HALF_OPEN (execution allowed)
+            False if circuit is OPEN (execution blocked)
+        """
+        now = time.time()
+        
+        if self.state == CircuitState.OPEN:
+            # Check if recovery timeout has passed
+            if self.last_failure_time and (now - self.last_failure_time) >= self.recovery_timeout:
+                self.state = CircuitState.HALF_OPEN
+                self.consecutive_successes = 0
+                return True  # Allow one test execution
+            return False  # Still open
+        
+        if success:
+            self.consecutive_failures = 0
+            self.consecutive_successes += 1
+            
+            if self.state == CircuitState.HALF_OPEN:
+                if self.consecutive_successes >= self.success_threshold:
+                    self.state = CircuitState.CLOSED
+                    self.consecutive_successes = 0
+            
+            return True
+        else:
+            self.consecutive_successes = 0
+            self.consecutive_failures += 1
+            self.last_failure_time = now
+            
+            if self.state == CircuitState.HALF_OPEN:
+                # Failed during recovery — reopen immediately
+                self.state = CircuitState.OPEN
+                self.total_trips += 1
+                return False
+            
+            if self.consecutive_failures >= self.failure_threshold:
+                self.state = CircuitState.OPEN
+                self.total_trips += 1
+                self.error_streaks.append(self.consecutive_failures)
+                return False
+            
+            return True
+    
+    def can_execute(self) -> bool:
+        """Check if execution is allowed."""
+        if self.state == CircuitState.OPEN:
+            if self.last_failure_time:
+                now = time.time()
+                if (now - self.last_failure_time) >= self.recovery_timeout:
+                    self.state = CircuitState.HALF_OPEN
+                    self.consecutive_successes = 0
+                    return True
+            return False
+        return True
+    
+    def get_state(self) -> Dict[str, Any]:
+        """Get current circuit state."""
+        return {
+            "state": self.state.value,
+            "consecutive_failures": self.consecutive_failures,
+            "consecutive_successes": self.consecutive_successes,
+            "total_trips": self.total_trips,
+            "max_streak": max(self.error_streaks) if self.error_streaks else 0,
+            "can_execute": self.can_execute(),
+        }
+    
+    def reset(self):
+        """Reset the circuit breaker."""
+        self.state = CircuitState.CLOSED
+        self.consecutive_failures = 0
+        self.consecutive_successes = 0
+        self.last_failure_time = None
+
+
+class ToolCircuitBreaker(CircuitBreaker):
+    """
+    Circuit breaker specifically for tool call error cascading.
+    
+    Provides recovery actions when the circuit opens.
+    """
+    
+    # Tools that are most effective at recovery (from audit data)
+    RECOVERY_TOOLS = [
+        "terminal",     # Most effective — 2300 recoveries
+        "read_file",    # Reset context by reading something
+        "search_files", # Find what went wrong
+    ]
+    
+    def get_recovery_action(self) -> Dict[str, Any]:
+        """
+        Get the recommended recovery action when circuit is open.
+        
+        Returns dict with action type and details.
+        """
+        streak = self.consecutive_failures
+        
+        if streak >= 9:
+            # After 9 errors: 41/46 recoveries via terminal
+            return {
+                "action": "terminal_only",
+                "reason": f"Error streak of {streak} — terminal is the only reliable recovery",
+                "suggested_tool": "terminal",
+                "suggested_command": "echo 'Resetting context'",
+                "severity": "critical",
+            }
+        elif streak >= 5:
+            return {
+                "action": "switch_tool_type",
+                "reason": f"Error streak of {streak} — switch to a different tool category",
+                "suggested_tools": ["read_file", "search_files", "terminal"],
+                "severity": "high",
+            }
+        elif streak >= self.failure_threshold:
+            return {
+                "action": "ask_user",
+                "reason": f"{streak} consecutive errors — ask user for guidance",
+                "suggested_response": "I'm encountering repeated errors. Would you like me to try a different approach?",
+                "severity": "medium",
+            }
+        else:
+            return {
+                "action": "continue",
+                "reason": f"Error streak of {streak} — within tolerance",
+                "severity": "low",
+            }
+    
+    def should_compress_context(self) -> bool:
+        """Determine if context compression would help recovery."""
+        return self.consecutive_failures >= 5
+    
+    def get_blocked_tool(self) -> Optional[str]:
+        """Get the tool that should be blocked (if any)."""
+        if self.state == CircuitState.OPEN:
+            return "last_failed_tool"
+        return None
+
+
+class MultiToolCircuitBreaker:
+    """
+    Manages per-tool circuit breakers and cross-tool cascade detection.
+    
+    When one tool trips its breaker, related tools are also warned.
+    """
+    
+    def __init__(self):
+        self.breakers: Dict[str, ToolCircuitBreaker] = {}
+        self.global_streak: int = 0
+        self.last_tool: Optional[str] = None
+        self.last_success: bool = True
+    
+    def get_breaker(self, tool_name: str) -> ToolCircuitBreaker:
+        """Get or create a circuit breaker for a tool."""
+        if tool_name not in self.breakers:
+            self.breakers[tool_name] = ToolCircuitBreaker()
+        return self.breakers[tool_name]
+    
+    def record_result(self, tool_name: str, success: bool) -> bool:
+        """
+        Record a tool call result. Returns True if execution should continue.
+        """
+        breaker = self.get_breaker(tool_name)
+        allowed = breaker.record_result(success)
+        
+        # Track global streak
+        if success:
+            self.global_streak = 0
+            self.last_success = True
+        else:
+            self.global_streak += 1
+            self.last_success = False
+        
+        self.last_tool = tool_name
+        return allowed
+    
+    def can_execute(self, tool_name: str) -> bool:
+        """Check if a specific tool can execute."""
+        breaker = self.get_breaker(tool_name)
+        return breaker.can_execute()
+    
+    def get_global_state(self) -> Dict[str, Any]:
+        """Get overall circuit breaker state."""
+        return {
+            "global_streak": self.global_streak,
+            "last_tool": self.last_tool,
+            "last_success": self.last_success,
+            "tool_states": {
+                name: breaker.get_state()
+                for name, breaker in self.breakers.items()
+                if breaker.consecutive_failures > 0 or breaker.total_trips > 0
+            },
+            "any_open": any(b.state == CircuitState.OPEN for b in self.breakers.values()),
+        }
+    
+    def get_recovery_action(self) -> Dict[str, Any]:
+        """Get recovery action based on global state."""
+        if self.global_streak == 0:
+            return {"action": "continue", "reason": "No errors"}
+        
+        # Find the breaker with the worst streak
+        worst = max(self.breakers.values(), key=lambda b: b.consecutive_failures, default=None)
+        if worst and worst.consecutive_failures > 0:
+            return worst.get_recovery_action()
+        
+        return {
+            "action": "continue",
+            "reason": f"Global streak: {self.global_streak}",
+            "severity": "low",
+        }
+    
+    def reset_all(self):
+        """Reset all circuit breakers."""
+        for breaker in self.breakers.values():
+            breaker.reset()
+        self.global_streak = 0
+        self.last_success = True

agent/memory_manager.py

@@ -50,6 +50,78 @@ def sanitize_context(text: str) -> str:
     return _FENCE_TAG_RE.sub('', text)
 
 
+# ---------------------------------------------------------------------------
+# Prefetch filtering helpers
+# ---------------------------------------------------------------------------
+
+# Meta-instruction debris that memory providers sometimes echo back.
+# These are prompts/instructions, not user-generated content.
+_META_INSTRUCTION_PATTERNS = [
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Focus on:\s*", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Note:\s*", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*System\s+(note|prompt|instruction):", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*You are\s+", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Please\s+(provide|respond|answer|write)", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Do not\s+", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Always\s+", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Consider\s+(the following|these|this)\b", re.IGNORECASE),
+    re.compile(r"^\s*[\-\*]?\s*>?\s*Here\s+(is|are)\s+(some|the|a few)\b", re.IGNORECASE),
+]
+
+
+def _is_meta_instruction_line(line: str) -> bool:
+    """Return True if the line looks like a prompt/template instruction, not memory content."""
+    for pat in _META_INSTRUCTION_PATTERNS:
+        if pat.search(line):
+            return True
+    return False
+
+
+def _is_low_signal_line(line: str) -> bool:
+    """Return True for very short or content-free lines."""
+    stripped = line.strip()
+    # Empty or just punctuation/list marker
+    if not stripped or stripped in {"-", "*", ">", "•", "—", "--"}:
+        return True
+    # Too short to be meaningful (< 15 chars after stripping markers)
+    cleaned = re.sub(r"^[\-\*•>\s]+", "", stripped)
+    if len(cleaned) < 15:
+        return True
+    return False
+
+
+def _filter_prefetch_lines(text: str) -> str:
+    """Filter and deduplicate prefetch result lines.
+
+    Removes:
+      - exact duplicate lines
+      - meta-instruction debris (prompts, templates)
+      - very short / content-free lines
+
+    Returns cleaned text, preserving original line grouping.
+    """
+    if not text or not text.strip():
+        return ""
+
+    seen: set = set()
+    kept: list = []
+    for line in text.splitlines(keepends=False):
+        stripped = line.strip()
+        # Deduplicate exact lines
+        if stripped in seen:
+            continue
+        # Skip meta-instructions
+        if _is_meta_instruction_line(line):
+            continue
+        # Skip low-signal lines
+        if _is_low_signal_line(line):
+            continue
+        seen.add(stripped)
+        kept.append(line)
+
+    return "\n".join(kept)
+
+
 def build_memory_context_block(raw_context: str) -> str:
     """Wrap prefetched memory in a fenced block with system note.
 
@@ -180,7 +252,14 @@ class MemoryManager:
                     "Memory provider '%s' prefetch failed (non-fatal): %s",
                     provider.name, e,
                 )
-        return "\n\n".join(parts)
+        raw = "\n\n".join(parts)
+        if not raw:
+            return ""
+        # Apply line-level filtering: dedupe, strip meta-instructions,
+        # remove very short fragments.  This prevents noisy providers
+        # (e.g. MemPalace transcript recall) from bloating context.
+        filtered = _filter_prefetch_lines(raw)
+        return filtered
 
     def queue_prefetch_all(self, query: str, *, session_id: str = "") -> None:
         """Queue background prefetch on all providers for the next turn."""

agent/mtls.py

@@ -0,0 +1,184 @@
+"""
+agent/mtls.py — Mutual TLS support for Hermes A2A communication.
+
+Provides:
+- build_server_ssl_context()  — SSL context for uvicorn that requires client certs
+- build_client_ssl_context()  — SSL context for httpx/aiohttp A2A clients
+- MTLSMiddleware               — FastAPI middleware that enforces client cert on A2A routes
+- is_mtls_configured()        — Check if env vars are set
+
+Configuration (environment variables):
+  HERMES_MTLS_CERT   Path to this agent's TLS certificate (PEM)
+  HERMES_MTLS_KEY    Path to this agent's TLS private key (PEM)
+  HERMES_MTLS_CA     Path to the Fleet CA certificate (PEM) — used to verify peers
+
+All three must be set to enable mTLS. If any is missing, mTLS is disabled and
+the server falls back to plain HTTP (or regular TLS without client auth).
+"""
+
+import logging
+import os
+import ssl
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# A2A routes that require a valid client certificate when mTLS is enabled.
+_A2A_PATH_PREFIXES = (
+    "/.well-known/agent-card",
+    "/agent-card",
+    "/api/agent-card",
+    "/a2a/",
+)
+
+
+def _get_env(key: str) -> Optional[str]:
+    val = os.environ.get(key, "").strip()
+    return val or None
+
+
+def is_mtls_configured() -> bool:
+    """Return True if all three mTLS env vars are set and the files exist."""
+    cert = _get_env("HERMES_MTLS_CERT")
+    key = _get_env("HERMES_MTLS_KEY")
+    ca = _get_env("HERMES_MTLS_CA")
+    if not (cert and key and ca):
+        return False
+    for label, path in (("HERMES_MTLS_CERT", cert), ("HERMES_MTLS_KEY", key), ("HERMES_MTLS_CA", ca)):
+        if not Path(path).is_file():
+            logger.warning("mTLS disabled: %s file not found: %s", label, path)
+            return False
+    return True
+
+
+def build_server_ssl_context() -> ssl.SSLContext:
+    """
+    Build an SSL context for the A2A server that:
+    - presents its own certificate
+    - requires and verifies the client's certificate against the Fleet CA
+
+    Raises:
+        RuntimeError: if mTLS env vars are not set or files are missing
+        ssl.SSLError: if cert/key/CA files are invalid
+    """
+    cert = _get_env("HERMES_MTLS_CERT")
+    key = _get_env("HERMES_MTLS_KEY")
+    ca = _get_env("HERMES_MTLS_CA")
+
+    if not (cert and key and ca):
+        raise RuntimeError(
+            "mTLS not configured. Set HERMES_MTLS_CERT, HERMES_MTLS_KEY, and HERMES_MTLS_CA."
+        )
+
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+    ctx.load_cert_chain(certfile=cert, keyfile=key)
+    ctx.load_verify_locations(cafile=ca)
+    # CERT_REQUIRED: reject connections without a valid client cert
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    logger.info("mTLS server context built (cert=%s, CA=%s)", cert, ca)
+    return ctx
+
+
+def build_client_ssl_context() -> ssl.SSLContext:
+    """
+    Build an SSL context for outbound A2A connections that:
+    - presents this agent's certificate as a client cert
+    - verifies the remote server against the Fleet CA
+
+    Raises:
+        RuntimeError: if mTLS env vars are not set or files are missing
+        ssl.SSLError: if cert/key/CA files are invalid
+    """
+    cert = _get_env("HERMES_MTLS_CERT")
+    key = _get_env("HERMES_MTLS_KEY")
+    ca = _get_env("HERMES_MTLS_CA")
+
+    if not (cert and key and ca):
+        raise RuntimeError(
+            "mTLS not configured. Set HERMES_MTLS_CERT, HERMES_MTLS_KEY, and HERMES_MTLS_CA."
+        )
+
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+    ctx.load_cert_chain(certfile=cert, keyfile=key)
+    ctx.load_verify_locations(cafile=ca)
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    ctx.check_hostname = True
+    logger.info("mTLS client context built (cert=%s, CA=%s)", cert, ca)
+    return ctx
+
+
+def get_peer_cn(ssl_object) -> Optional[str]:
+    """Extract the CN from the peer certificate's subject, or None."""
+    try:
+        peer_cert = ssl_object.getpeercert()
+        if not peer_cert:
+            return None
+        for rdn in peer_cert.get("subject", ()):
+            for attr, value in rdn:
+                if attr == "commonName":
+                    return value
+    except Exception:
+        pass
+    return None
+
+
+class MTLSMiddleware:
+    """
+    ASGI middleware that enforces client certificate verification on A2A routes.
+
+    When mTLS is NOT configured (no env vars) or the route is not an A2A route,
+    the request passes through unchanged.
+
+    When mTLS IS configured and the route matches an A2A prefix, the middleware
+    checks that the request arrived over a TLS connection with a verified client
+    certificate. If not, it returns HTTP 403.
+
+    Note: This middleware only provides defence-in-depth at the app layer.
+    The primary enforcement is at the SSL context level (CERT_REQUIRED on the
+    server context). This middleware is useful when the server runs behind a
+    TLS-terminating proxy that forwards cert info via headers (not yet
+    implemented) or for test-time injection.
+    """
+
+    def __init__(self, app):
+        self.app = app
+        self._enabled = is_mtls_configured()
+        if self._enabled:
+            logger.info("MTLSMiddleware enabled — A2A routes require client cert")
+
+    def _is_a2a_route(self, path: str) -> bool:
+        return any(path.startswith(prefix) for prefix in _A2A_PATH_PREFIXES)
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "http" and self._enabled and self._is_a2a_route(scope.get("path", "")):
+            # Check for client cert in the SSL connection
+            transport = scope.get("extensions", {}).get("tls", {})
+            peer_cert = transport.get("peer_cert")
+            if peer_cert is None:
+                # No client cert — reject
+                response = _forbidden_response("Client certificate required for A2A endpoints")
+                await response(scope, receive, send)
+                return
+
+        await self.app(scope, receive, send)
+
+
+def _forbidden_response(message: str):
+    """Return a minimal ASGI 403 response."""
+    body = message.encode()
+
+    async def respond(scope, receive, send):
+        await send({
+            "type": "http.response.start",
+            "status": 403,
+            "headers": [
+                (b"content-type", b"text/plain"),
+                (b"content-length", str(len(body)).encode()),
+            ],
+        })
+        await send({"type": "http.response.body", "body": body})
+
+    return respond

agent/tool_fixation_detector.py

@@ -0,0 +1,156 @@
+"""Tool fixation detection — break repetitive tool calling loops.
+
+Detects when the agent latches onto one tool and calls it repeatedly
+without making progress. Injects a nudge prompt to break the loop.
+
+Usage:
+    from agent.tool_fixation_detector import ToolFixationDetector
+    detector = ToolFixationDetector()
+    nudge = detector.record("execute_code")
+    if nudge:
+        # Inject nudge into conversation
+        messages.append({"role": "system", "content": nudge})
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+
+# Default thresholds
+_DEFAULT_THRESHOLD = int(os.getenv("TOOL_FIXATION_THRESHOLD", "5"))
+_DEFAULT_WINDOW = int(os.getenv("TOOL_FIXATION_WINDOW", "10"))
+
+
+@dataclass
+class FixationEvent:
+    """Record of a fixation detection."""
+    tool_name: str
+    streak_length: int
+    threshold: int
+    nudge_sent: bool = False
+
+
+class ToolFixationDetector:
+    """Detects and breaks tool fixation loops.
+
+    Tracks the sequence of tool calls and detects when the same tool
+    is called N times consecutively. When detected, returns a nudge
+    prompt to inject into the conversation.
+    """
+
+    def __init__(self, threshold: int = 0, window: int = 0):
+        self.threshold = threshold or _DEFAULT_THRESHOLD
+        self.window = window or _DEFAULT_WINDOW
+        self._history: List[str] = []
+        self._current_streak: str = ""
+        self._streak_count: int = 0
+        self._nudges_sent: int = 0
+        self._events: List[FixationEvent] = []
+
+    @property
+    def nudges_sent(self) -> int:
+        return self._nudges_sent
+
+    @property
+    def events(self) -> List[FixationEvent]:
+        return list(self._events)
+
+    def record(self, tool_name: str) -> Optional[str]:
+        """Record a tool call and return nudge prompt if fixation detected.
+
+        Args:
+            tool_name: Name of the tool that was called.
+
+        Returns:
+            Nudge prompt string if fixation detected, None otherwise.
+        """
+        self._history.append(tool_name)
+
+        # Trim history to window
+        if len(self._history) > self.window:
+            self._history = self._history[-self.window:]
+
+        # Update streak
+        if tool_name == self._current_streak:
+            self._streak_count += 1
+        else:
+            self._current_streak = tool_name
+            self._streak_count = 1
+
+        # Check for fixation
+        if self._streak_count >= self.threshold:
+            event = FixationEvent(
+                tool_name=tool_name,
+                streak_length=self._streak_count,
+                threshold=self.threshold,
+                nudge_sent=True,
+            )
+            self._events.append(event)
+            self._nudges_sent += 1
+
+            return self._build_nudge(tool_name, self._streak_count)
+
+        return None
+
+    def _build_nudge(self, tool_name: str, count: int) -> str:
+        """Build a nudge prompt to break the fixation loop."""
+        return (
+            f"[SYSTEM: You have called `{tool_name}` {count} times in a row "
+            f"without switching tools. This suggests a fixation loop. "
+            f"Consider:\n"
+            f"1. Is the tool returning an error? Read the error carefully.\n"
+            f"2. Is there a different tool that could help?\n"
+            f"3. Should you ask the user for clarification?\n"
+            f"4. Is the task actually complete?\n"
+            f"Break the loop by trying a different approach.]"
+        )
+
+    def reset(self) -> None:
+        """Reset the detector state."""
+        self._history.clear()
+        self._current_streak = ""
+        self._streak_count = 0
+
+    def get_streak_info(self) -> dict:
+        """Get current streak information."""
+        return {
+            "current_tool": self._current_streak,
+            "streak_count": self._streak_count,
+            "threshold": self.threshold,
+            "at_threshold": self._streak_count >= self.threshold,
+            "nudges_sent": self._nudges_sent,
+        }
+
+    def format_report(self) -> str:
+        """Format fixation events as a report."""
+        if not self._events:
+            return "No tool fixation detected."
+
+        lines = [
+            f"Tool Fixation Report ({len(self._events)} events)",
+            "=" * 40,
+        ]
+        for e in self._events:
+            lines.append(f"  {e.tool_name}: {e.streak_length} consecutive calls (threshold: {e.threshold})")
+        return "\n".join(lines)
+
+
+# Singleton
+_detector: Optional[ToolFixationDetector] = None
+
+
+def get_fixation_detector() -> ToolFixationDetector:
+    """Get or create the singleton detector."""
+    global _detector
+    if _detector is None:
+        _detector = ToolFixationDetector()
+    return _detector
+
+
+def reset_fixation_detector() -> None:
+    """Reset the singleton."""
+    global _detector
+    _detector = None

ansible/fleet_mtls.yml

@@ -0,0 +1,32 @@
+---
+# fleet_mtls.yml — Deploy mutual-TLS certificates to all fleet agents.
+#
+# Prerequisites:
+#   1. Run scripts/gen_fleet_ca.sh to create the fleet CA.
+#   2. For each agent, run:
+#        scripts/gen_agent_cert.sh --agent timmy
+#        scripts/gen_agent_cert.sh --agent allegro
+#        scripts/gen_agent_cert.sh --agent ezra
+#
+# Usage:
+#   ansible-playbook -i inventory/fleet.ini ansible/fleet_mtls.yml
+#
+# Inventory example (inventory/fleet.ini):
+#   [fleet]
+#   timmy.local   agent_name=timmy
+#   allegro.local agent_name=allegro
+#   ezra.local    agent_name=ezra
+#
+# Refs #806
+
+- name: Distribute fleet mTLS certificates
+  hosts: fleet
+  become: true
+  vars:
+    _pki_base: "{{ lookup('env', 'HOME') }}/.hermes/pki"
+  roles:
+    - role: hermes_mtls
+      vars:
+        hermes_mtls_local_ca_cert: "{{ _pki_base }}/ca/fleet-ca.crt"
+        hermes_mtls_local_agent_cert: "{{ _pki_base }}/agents/{{ agent_name }}/{{ agent_name }}.crt"
+        hermes_mtls_local_agent_key: "{{ _pki_base }}/agents/{{ agent_name }}/{{ agent_name }}.key"

ansible/inventory/fleet.ini.example

@@ -0,0 +1,12 @@
+# Example fleet inventory for mutual-TLS cert distribution.
+# Copy to fleet.ini and adjust hostnames/IPs.
+# Refs #806
+
+[fleet_agents]
+timmy    ansible_host=192.168.1.10
+allegro  ansible_host=192.168.1.11
+ezra     ansible_host=192.168.1.12
+
+[fleet_agents:vars]
+ansible_user=hermes
+ansible_python_interpreter=/usr/bin/python3

ansible/roles/fleet_mtls_certs/defaults/main.yml

@@ -0,0 +1,21 @@
+---
+# Default paths on the *control node* where certs are read from.
+# Override these in your inventory / group_vars as needed.
+
+# Fleet CA certificate (public; safe to push to all nodes)
+fleet_mtls_ca_cert_src: "{{ lookup('env', 'HOME') }}/.hermes/pki/ca/fleet-ca.crt"
+
+# Per-agent cert/key source dir on the control node.
+# Expected layout:  <fleet_mtls_agent_certs_dir>/<agent_name>/<agent_name>.{crt,key}
+fleet_mtls_agent_certs_dir: "{{ lookup('env', 'HOME') }}/.hermes/pki/agents"
+
+# Remote destination paths on the fleet node
+fleet_mtls_remote_pki_dir: "/etc/hermes/pki"
+fleet_mtls_remote_ca_dir: "{{ fleet_mtls_remote_pki_dir }}/ca"
+fleet_mtls_remote_agent_dir: "{{ fleet_mtls_remote_pki_dir }}/agent"
+
+# The agent name to deploy (set per-host in inventory, e.g. timmy / allegro / ezra)
+fleet_mtls_agent_name: "{{ inventory_hostname_short }}"
+
+# Hermes service name (for reload notification)
+fleet_mtls_hermes_service: "hermes-a2a"

ansible/roles/fleet_mtls_certs/handlers/main.yml

@@ -0,0 +1,7 @@
+---
+- name: Restart hermes-a2a
+  ansible.builtin.systemd:
+    name: "{{ fleet_mtls_hermes_service }}"
+    state: restarted
+  when: ansible_service_mgr == "systemd"
+  ignore_errors: true   # service may not exist in all environments

ansible/roles/fleet_mtls_certs/meta/main.yml

@@ -0,0 +1,17 @@
+---
+galaxy_info:
+  role_name: fleet_mtls_certs
+  author: hermes-agent
+  description: >
+    Distribute fleet CA and per-agent mTLS certificates to Hermes fleet nodes.
+    Part of issue #806 — A2A mutual TLS between fleet agents.
+  min_ansible_version: "2.14"
+  platforms:
+    - name: Debian
+      versions: [bookworm, bullseye]
+    - name: Ubuntu
+      versions: ["22.04", "24.04"]
+    - name: EL
+      versions: ["8", "9"]
+
+dependencies: []

ansible/roles/fleet_mtls_certs/tasks/main.yml

@@ -0,0 +1,99 @@
+---
+# fleet_mtls_certs/tasks/main.yml
+#
+# Distribute the fleet CA certificate and the per-agent TLS cert+key to
+# each fleet node.  Triggers a hermes-a2a service restart when any cert
+# changes.
+#
+# Refs #806 — A2A mutual TLS between fleet agents.
+
+- name: Verify agent cert source files exist on control node
+  ansible.builtin.stat:
+    path: "{{ item }}"
+  register: _src_stat
+  delegate_to: localhost
+  loop:
+    - "{{ fleet_mtls_ca_cert_src }}"
+    - "{{ fleet_mtls_agent_certs_dir }}/{{ fleet_mtls_agent_name }}/{{ fleet_mtls_agent_name }}.crt"
+    - "{{ fleet_mtls_agent_certs_dir }}/{{ fleet_mtls_agent_name }}/{{ fleet_mtls_agent_name }}.key"
+  loop_control:
+    label: "{{ item | basename }}"
+
+- name: Fail if any source cert is missing
+  ansible.builtin.fail:
+    msg: >
+      Required cert file not found: {{ item.item }}
+      Run scripts/gen_fleet_ca.sh and scripts/gen_agent_cert.sh --agent {{ fleet_mtls_agent_name }} first.
+  when: not item.stat.exists
+  loop: "{{ _src_stat.results }}"
+  loop_control:
+    label: "{{ item.item | basename }}"
+
+# -----------------------------------------------------------------------
+# Remote directory structure
+# -----------------------------------------------------------------------
+
+- name: Create remote PKI directories
+  ansible.builtin.file:
+    path: "{{ item }}"
+    state: directory
+    owner: root
+    group: root
+    mode: "0750"
+  loop:
+    - "{{ fleet_mtls_remote_pki_dir }}"
+    - "{{ fleet_mtls_remote_ca_dir }}"
+    - "{{ fleet_mtls_remote_agent_dir }}"
+
+# -----------------------------------------------------------------------
+# Fleet CA certificate (public — read-only for all)
+# -----------------------------------------------------------------------
+
+- name: Deploy fleet CA certificate
+  ansible.builtin.copy:
+    src: "{{ fleet_mtls_ca_cert_src }}"
+    dest: "{{ fleet_mtls_remote_ca_dir }}/fleet-ca.crt"
+    owner: root
+    group: root
+    mode: "0644"
+  notify: Restart hermes-a2a
+
+# -----------------------------------------------------------------------
+# Per-agent certificate (public portion)
+# -----------------------------------------------------------------------
+
+- name: Deploy agent certificate
+  ansible.builtin.copy:
+    src: "{{ fleet_mtls_agent_certs_dir }}/{{ fleet_mtls_agent_name }}/{{ fleet_mtls_agent_name }}.crt"
+    dest: "{{ fleet_mtls_remote_agent_dir }}/agent.crt"
+    owner: root
+    group: root
+    mode: "0644"
+  notify: Restart hermes-a2a
+
+# -----------------------------------------------------------------------
+# Per-agent private key (secret — root-only read)
+# -----------------------------------------------------------------------
+
+- name: Deploy agent private key
+  ansible.builtin.copy:
+    src: "{{ fleet_mtls_agent_certs_dir }}/{{ fleet_mtls_agent_name }}/{{ fleet_mtls_agent_name }}.key"
+    dest: "{{ fleet_mtls_remote_agent_dir }}/agent.key"
+    owner: root
+    group: root
+    mode: "0600"
+  no_log: true   # suppress file content from Ansible output
+  notify: Restart hermes-a2a
+
+# -----------------------------------------------------------------------
+# Environment file for hermes-a2a systemd unit
+# -----------------------------------------------------------------------
+
+- name: Write hermes-a2a environment file
+  ansible.builtin.template:
+    src: hermes_a2a_env.j2
+    dest: /etc/hermes/a2a.env
+    owner: root
+    group: root
+    mode: "0640"
+  notify: Restart hermes-a2a

ansible/roles/fleet_mtls_certs/templates/hermes_a2a_env.j2

@@ -0,0 +1,10 @@
+# Managed by Ansible — fleet_mtls_certs role
+# Environment variables for the hermes-a2a systemd service.
+# Source this file in the [Service] section: EnvironmentFile=/etc/hermes/a2a.env
+
+HERMES_AGENT_NAME={{ fleet_mtls_agent_name }}
+HERMES_A2A_CERT={{ fleet_mtls_remote_agent_dir }}/agent.crt
+HERMES_A2A_KEY={{ fleet_mtls_remote_agent_dir }}/agent.key
+HERMES_A2A_CA={{ fleet_mtls_remote_ca_dir }}/fleet-ca.crt
+HERMES_A2A_HOST=0.0.0.0
+HERMES_A2A_PORT=9443

ansible/roles/hermes_mtls/defaults/main.yml

@@ -0,0 +1,21 @@
+---
+# Ansible role: hermes_mtls
+# Distributes fleet mTLS certificates to Hermes agent nodes.
+#
+# Required variables (set in inventory / group_vars / --extra-vars):
+#   hermes_mtls_local_ca_cert   Local path on the Ansible controller to fleet-ca.crt
+#   hermes_mtls_local_agent_cert  Local path to this agent's .crt file
+#   hermes_mtls_local_agent_key   Local path to this agent's .key file
+#
+# Optional overrides:
+hermes_mtls_cert_dir: /etc/hermes/certs
+hermes_mtls_cert_owner: hermes
+hermes_mtls_cert_group: hermes
+hermes_mtls_cert_mode: "0640"
+hermes_mtls_ca_cert_mode: "0644"
+
+# Env file that Hermes reads on startup (systemd EnvironmentFile or .env)
+hermes_mtls_env_file: /etc/hermes/mtls.env
+
+# Hermes systemd service name — restarted after cert changes
+hermes_mtls_service: hermes-gateway

ansible/roles/hermes_mtls/handlers/main.yml

@@ -0,0 +1,7 @@
+---
+- name: Restart hermes service
+  ansible.builtin.systemd:
+    name: "{{ hermes_mtls_service }}"
+    state: restarted
+    daemon_reload: true
+  when: ansible_service_mgr == "systemd"

ansible/roles/hermes_mtls/meta/main.yml

@@ -0,0 +1,16 @@
+---
+galaxy_info:
+  role_name: hermes_mtls
+  author: Hermes Fleet
+  description: Distribute mTLS certificates to Hermes fleet nodes for A2A authentication
+  license: MIT
+  min_ansible_version: "2.14"
+  platforms:
+    - name: Ubuntu
+      versions: ["22.04", "24.04"]
+    - name: Debian
+      versions: ["12"]
+    - name: EL
+      versions: ["9"]
+
+dependencies: []

ansible/roles/hermes_mtls/tasks/main.yml

@@ -0,0 +1,67 @@
+---
+# hermes_mtls role — distribute fleet mTLS certificates to a Hermes agent node.
+#
+# This role:
+#   1. Creates the cert directory on the remote node
+#   2. Copies the Fleet CA cert, agent cert, and agent key
+#   3. Writes an env file with HERMES_MTLS_* variables
+#   4. Restarts the Hermes service if any cert changed
+
+- name: Ensure cert directory exists
+  ansible.builtin.file:
+    path: "{{ hermes_mtls_cert_dir }}"
+    state: directory
+    owner: "{{ hermes_mtls_cert_owner }}"
+    group: "{{ hermes_mtls_cert_group }}"
+    mode: "0750"
+
+- name: Copy Fleet CA certificate
+  ansible.builtin.copy:
+    src: "{{ hermes_mtls_local_ca_cert }}"
+    dest: "{{ hermes_mtls_cert_dir }}/fleet-ca.crt"
+    owner: "{{ hermes_mtls_cert_owner }}"
+    group: "{{ hermes_mtls_cert_group }}"
+    mode: "{{ hermes_mtls_ca_cert_mode }}"
+  notify: Restart hermes service
+
+- name: Copy agent TLS certificate
+  ansible.builtin.copy:
+    src: "{{ hermes_mtls_local_agent_cert }}"
+    dest: "{{ hermes_mtls_cert_dir }}/agent.crt"
+    owner: "{{ hermes_mtls_cert_owner }}"
+    group: "{{ hermes_mtls_cert_group }}"
+    mode: "{{ hermes_mtls_cert_mode }}"
+  notify: Restart hermes service
+
+- name: Copy agent TLS private key
+  ansible.builtin.copy:
+    src: "{{ hermes_mtls_local_agent_key }}"
+    dest: "{{ hermes_mtls_cert_dir }}/agent.key"
+    owner: "{{ hermes_mtls_cert_owner }}"
+    group: "{{ hermes_mtls_cert_group }}"
+    mode: "0600"
+  notify: Restart hermes service
+
+- name: Write mTLS environment file
+  ansible.builtin.template:
+    src: mtls.env.j2
+    dest: "{{ hermes_mtls_env_file }}"
+    owner: "{{ hermes_mtls_cert_owner }}"
+    group: "{{ hermes_mtls_cert_group }}"
+    mode: "0640"
+  notify: Restart hermes service
+
+- name: Verify cert files are readable by service user
+  ansible.builtin.stat:
+    path: "{{ item }}"
+  loop:
+    - "{{ hermes_mtls_cert_dir }}/fleet-ca.crt"
+    - "{{ hermes_mtls_cert_dir }}/agent.crt"
+    - "{{ hermes_mtls_cert_dir }}/agent.key"
+  register: _cert_stat
+
+- name: Assert all cert files exist
+  ansible.builtin.assert:
+    that: item.stat.exists
+    fail_msg: "Expected cert file missing: {{ item.item }}"
+  loop: "{{ _cert_stat.results }}"

ansible/roles/hermes_mtls/templates/mtls.env.j2

@@ -0,0 +1,8 @@
+# Hermes mTLS environment — generated by hermes_mtls Ansible role
+# Source this file or use as a systemd EnvironmentFile=
+# WARNING: This file contains the path to the agent's private key.
+#          Restrict read access to the hermes service user.
+
+HERMES_MTLS_CERT={{ hermes_mtls_cert_dir }}/agent.crt
+HERMES_MTLS_KEY={{ hermes_mtls_cert_dir }}/agent.key
+HERMES_MTLS_CA={{ hermes_mtls_cert_dir }}/fleet-ca.crt

docs/tool-investigation-report.md

@@ -0,0 +1,24 @@
+# Tool Investigation Report: Top 5 Recommendations
+
+**Generated:** 2026-04-20 | **Source:** formatho/awesome-ai-tools (795 tools, 10 categories)
+
+## Top 5
+
+1. **LiteLLM** (76k) — Unified API gateway. Replace custom provider routing. Impact: 5/5, Effort: 2/5
+2. **Mem0** (53k) — Universal memory layer. Structured long-term memory. Impact: 5/5, Effort: 3/5
+3. **RAGFlow** (77k) — RAG engine with OCR. Document processing upgrade. Impact: 4/5, Effort: 4/5
+4. **LiteRT-LM** (3.7k) — On-device inference. Edge/mobile deployment. Impact: 4/5, Effort: 3/5
+5. **Claude-Mem** (61k) — Session capture and context injection. Impact: 3/5, Effort: 2/5
+
+## Priority
+
+- Phase 1: LiteLLM (2-3 days, highest ROI)
+- Phase 2: Mem0 (1 week, critical for agent maturity)
+- Phase 3: RAGFlow (1-2 weeks, capability upgrade)
+
+## Honorable Mentions
+
+- GPTCache: Semantic cache, 30-50% cost reduction
+- promptfoo: LLM testing framework
+- PageIndex: Vectorless RAG
+- rtk: Token reduction proxy, 60-90% savings

gateway/config.py

@@ -8,6 +8,7 @@ Handles loading and validating configuration for:
 - Delivery preferences
 """
 
+import ipaddress
 import logging
 import os
 import json
@@ -679,6 +680,26 @@ def load_gateway_config() -> GatewayConfig:
     return config
 
 
+def _is_network_accessible(host: str) -> bool:
+    """Return True if *host* would expose a server beyond the loopback interface.
+
+    Duplicates the logic in ``gateway.platforms.base.is_network_accessible``
+    without creating a circular import (base.py imports from this module).
+    """
+    try:
+        addr = ipaddress.ip_address(host)
+        if addr.is_loopback:
+            return False
+        # ::ffff:127.x.x.x — Python's is_loopback returns False for
+        # IPv4-mapped loopback; unwrap and check the underlying IPv4.
+        if getattr(addr, "ipv4_mapped", None) and addr.ipv4_mapped.is_loopback:
+            return False
+        return True
+    except ValueError:
+        # Hostname: assume it could be network-accessible.
+        return True
+
+
 def _validate_gateway_config(config: "GatewayConfig") -> None:
     """Validate and sanitize a loaded GatewayConfig in place.
 
@@ -747,6 +768,22 @@ def _validate_gateway_config(config: "GatewayConfig") -> None:
                 )
                 pconfig.enabled = False
 
+    # Warn when the API server is enabled on a network-accessible address
+    # without an auth key.  The adapter will refuse to start anyway, but
+    # surfacing this at config-load time lets operators see the problem in
+    # the startup log before any platform adapter initialisation runs.
+    api_cfg = config.platforms.get(Platform.API_SERVER)
+    if api_cfg and api_cfg.enabled:
+        key = api_cfg.extra.get("key", "")
+        host = api_cfg.extra.get("host", "127.0.0.1")
+        if not key and _is_network_accessible(host):
+            logger.warning(
+                "API Server is enabled on %s but API_SERVER_KEY is not set. "
+                "The adapter will refuse to start on a network-accessible address. "
+                "Set API_SERVER_KEY or bind to 127.0.0.1 for local-only access.",
+                host,
+            )
+
 
 def _apply_env_overrides(config: GatewayConfig) -> None:
     """Apply environment variable overrides to config."""

hermes_cli/web_server.py

@@ -46,6 +46,7 @@ from hermes_cli.config import (
 )
 from gateway.status import get_running_pid, read_runtime_status
 from agent.agent_card import get_agent_card_json
+from agent.mtls import is_mtls_configured, MTLSMiddleware, build_server_ssl_context
 
 try:
     from fastapi import FastAPI, HTTPException, Request
@@ -87,6 +88,10 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# mTLS: enforce client certificate on A2A endpoints when configured.
+# Activated by setting HERMES_MTLS_CERT, HERMES_MTLS_KEY, HERMES_MTLS_CA.
+app.add_middleware(MTLSMiddleware)
+
 # ---------------------------------------------------------------------------
 # Endpoints that do NOT require the session token.  Everything else under
 # /api/ is gated by the auth middleware below.  Keep this list minimal —
@@ -2105,6 +2110,20 @@ def start_server(
             "authentication. Only use on trusted networks.", host,
         )
 
+    # mTLS: when configured, pass SSL context to uvicorn so all connections
+    # are TLS with mandatory client certificate verification.
+    ssl_context = None
+    scheme = "http"
+    if is_mtls_configured():
+        try:
+            ssl_context = build_server_ssl_context()
+            scheme = "https"
+            _log.info(
+                "mTLS enabled — server requires client certificates (A2A auth)"
+            )
+        except Exception as exc:
+            _log.error("Failed to build mTLS SSL context: %s — starting without TLS", exc)
+
     if open_browser:
         import threading
         import webbrowser
@@ -2112,9 +2131,11 @@ def start_server(
         def _open():
             import time as _t
             _t.sleep(1.0)
-            webbrowser.open(f"http://{host}:{port}")
+            webbrowser.open(f"{scheme}://{host}:{port}")
 
         threading.Thread(target=_open, daemon=True).start()
 
-    print(f"  Hermes Web UI → http://{host}:{port}")
-    uvicorn.run(app, host=host, port=port, log_level="warning")
+    print(f"  Hermes Web UI → {scheme}://{host}:{port}")
+    if ssl_context is not None:
+        print("  mTLS enabled — client certificate required for A2A endpoints")
+    uvicorn.run(app, host=host, port=port, log_level="warning", ssl=ssl_context)

model_tools.py

@@ -29,6 +29,7 @@ from typing import Dict, Any, List, Optional, Tuple
 from tools.registry import discover_builtin_tools, registry
 from tools.poka_yoke import validate_tool_call
 from tools.tool_pokayoke import validate_tool_call, reset_circuit_breaker, get_hallucination_stats
+from tools.hardcoded_path_guard import guard_tool_dispatch as _guard_hardcoded_paths
 from toolsets import resolve_toolset, validate_toolset
 from agent.tool_orchestrator import orchestrator
 
@@ -502,6 +503,12 @@ def handle_function_call(
             # Prefer the caller-provided list so subagents can't overwrite
             # the parent's tool set via the process-global.
             sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names
+            # Poka-yoke #921: guard against hardcoded home-directory paths
+            _hardcoded_err = _guard_hardcoded_paths(function_name, function_args)
+            if _hardcoded_err:
+                logger.warning(f"Hardcoded path blocked: {function_name}")
+                return _hardcoded_err
+
             # Poka-yoke: validate tool call before dispatch
             is_valid, corrected_name, corrected_params, pokayoke_messages = validate_tool_call(function_name, function_args)
             if not is_valid:

scripts/gen_agent_cert.sh

@@ -0,0 +1,129 @@
+#!/usr/bin/env bash
+# gen_agent_cert.sh — Generate a TLS certificate for a fleet agent.
+#
+# Usage:
+#   ./scripts/gen_agent_cert.sh --agent <name> [--ca-dir <dir>] [--out-dir <dir>]
+#
+# Known agents: timmy, allegro, ezra  (case-insensitive; any name is accepted)
+#
+# Outputs (default: ~/.hermes/pki/agents/<name>/):
+#   <name>.key  — agent private key  (chmod 600, stays on the agent host)
+#   <name>.crt  — agent certificate  (signed by the fleet CA)
+#
+# Run gen_fleet_ca.sh first if you haven't already.
+# Refs #806
+
+set -euo pipefail
+
+CERT_DAYS=365   # 1 year; rotate annually
+KEY_BITS=2048
+
+# ---------------------------------------------------------------------------
+# Parse args
+# ---------------------------------------------------------------------------
+AGENT_NAME=""
+CA_DIR="${HOME}/.hermes/pki/ca"
+OUT_DIR=""
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --agent)     AGENT_NAME="${2,,}"; shift 2 ;;   # lower-case
+        --ca-dir)    CA_DIR="$2";         shift 2 ;;
+        --out-dir)   OUT_DIR="$2";        shift 2 ;;
+        -h|--help)
+            echo "Usage: $0 --agent <name> [--ca-dir <dir>] [--out-dir <dir>]"
+            echo "  Known agents: timmy, allegro, ezra"
+            exit 0
+            ;;
+        *)
+            echo "Unknown option: $1" >&2
+            exit 1
+            ;;
+    esac
+done
+
+if [[ -z "$AGENT_NAME" ]]; then
+    echo "ERROR: --agent <name> is required." >&2
+    exit 1
+fi
+
+OUT_DIR="${OUT_DIR:-${HOME}/.hermes/pki/agents/${AGENT_NAME}}"
+
+# ---------------------------------------------------------------------------
+# Prereq check
+# ---------------------------------------------------------------------------
+if ! command -v openssl &>/dev/null; then
+    echo "ERROR: openssl not found." >&2
+    exit 1
+fi
+
+CA_KEY="$CA_DIR/fleet-ca.key"
+CA_CRT="$CA_DIR/fleet-ca.crt"
+
+if [[ ! -f "$CA_KEY" || ! -f "$CA_CRT" ]]; then
+    echo "ERROR: Fleet CA not found in $CA_DIR" >&2
+    echo "       Run scripts/gen_fleet_ca.sh first." >&2
+    exit 1
+fi
+
+mkdir -p "$OUT_DIR"
+chmod 700 "$OUT_DIR"
+
+AGENT_KEY="$OUT_DIR/${AGENT_NAME}.key"
+AGENT_CRT="$OUT_DIR/${AGENT_NAME}.crt"
+AGENT_CSR="$OUT_DIR/${AGENT_NAME}.csr"
+
+if [[ -f "$AGENT_KEY" || -f "$AGENT_CRT" ]]; then
+    echo "Cert for agent '$AGENT_NAME' already exists in $OUT_DIR"
+    echo "  $AGENT_KEY"
+    echo "  $AGENT_CRT"
+    echo "Delete them manually if you want to regenerate."
+    exit 0
+fi
+
+echo "Generating cert for agent '$AGENT_NAME' ..."
+
+SUBJECT="/CN=${AGENT_NAME}.fleet.hermes/O=Hermes/OU=Fleet Agent"
+
+# Agent private key
+openssl genrsa -out "$AGENT_KEY" "$KEY_BITS" 2>/dev/null
+chmod 600 "$AGENT_KEY"
+
+# Certificate Signing Request
+openssl req -new \
+    -key "$AGENT_KEY" \
+    -out "$AGENT_CSR" \
+    -subj "$SUBJECT" 2>/dev/null
+
+# Sign with fleet CA — include SAN so modern TLS stacks accept it
+EXT_CONF=$(mktemp)
+trap 'rm -f "$EXT_CONF" "$AGENT_CSR"' EXIT
+
+cat > "$EXT_CONF" <<EOF
+[v3_agent]
+basicConstraints = CA:FALSE
+keyUsage = critical, digitalSignature, keyEncipherment
+extendedKeyUsage = clientAuth, serverAuth
+subjectKeyIdentifier = hash
+authorityKeyIdentifier = keyid,issuer
+subjectAltName = DNS:${AGENT_NAME}.fleet.hermes, DNS:${AGENT_NAME}
+EOF
+
+openssl x509 -req \
+    -in "$AGENT_CSR" \
+    -CA "$CA_CRT" \
+    -CAkey "$CA_KEY" \
+    -CAcreateserial \
+    -out "$AGENT_CRT" \
+    -days "$CERT_DAYS" \
+    -extfile "$EXT_CONF" \
+    -extensions v3_agent 2>/dev/null
+
+chmod 644 "$AGENT_CRT"
+
+echo ""
+echo "Agent cert generated:"
+echo "  Private key : $AGENT_KEY"
+echo "  Certificate : $AGENT_CRT"
+echo ""
+openssl x509 -in "$AGENT_CRT" -noout -subject -issuer -dates

scripts/gen_fleet_ca.sh

@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+# gen_fleet_ca.sh — Generate the Hermes fleet Certificate Authority.
+#
+# Usage:
+#   ./scripts/gen_fleet_ca.sh [--out-dir <dir>]
+#
+# Outputs (default: ~/.hermes/pki/ca/):
+#   fleet-ca.key   — CA private key  (chmod 600, keep secret)
+#   fleet-ca.crt   — CA certificate  (distribute to all fleet nodes)
+#
+# The CA is valid for 10 years.  Regenerate + redistribute when it expires.
+# Refs #806
+
+set -euo pipefail
+
+CA_SUBJECT="/CN=Hermes Fleet CA/O=Hermes/OU=Fleet"
+CA_DAYS=3650   # 10 years
+
+# ---------------------------------------------------------------------------
+# Parse args
+# ---------------------------------------------------------------------------
+OUT_DIR="${HOME}/.hermes/pki/ca"
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --out-dir)  OUT_DIR="$2"; shift 2 ;;
+        -h|--help)
+            echo "Usage: $0 [--out-dir <dir>]"
+            exit 0
+            ;;
+        *)
+            echo "Unknown option: $1" >&2
+            exit 1
+            ;;
+    esac
+done
+
+# ---------------------------------------------------------------------------
+# Prereq check
+# ---------------------------------------------------------------------------
+if ! command -v openssl &>/dev/null; then
+    echo "ERROR: openssl not found. Install OpenSSL and re-run." >&2
+    exit 1
+fi
+
+mkdir -p "$OUT_DIR"
+chmod 700 "$OUT_DIR"
+
+CA_KEY="$OUT_DIR/fleet-ca.key"
+CA_CRT="$OUT_DIR/fleet-ca.crt"
+
+if [[ -f "$CA_KEY" || -f "$CA_CRT" ]]; then
+    echo "Fleet CA already exists in $OUT_DIR"
+    echo "  $CA_KEY"
+    echo "  $CA_CRT"
+    echo "Delete them manually if you want to regenerate."
+    exit 0
+fi
+
+echo "Generating fleet CA in $OUT_DIR ..."
+
+# Generate 4096-bit RSA key for the CA
+openssl genrsa -out "$CA_KEY" 4096 2>/dev/null
+chmod 600 "$CA_KEY"
+
+# Self-sign the CA certificate
+openssl req -new -x509 \
+    -key "$CA_KEY" \
+    -out "$CA_CRT" \
+    -days "$CA_DAYS" \
+    -subj "$CA_SUBJECT" \
+    -addext "basicConstraints=critical,CA:TRUE,pathlen:0" \
+    -addext "keyUsage=critical,keyCertSign,cRLSign" \
+    -addext "subjectKeyIdentifier=hash" 2>/dev/null
+
+chmod 644 "$CA_CRT"
+
+echo ""
+echo "Fleet CA generated successfully:"
+echo "  Private key : $CA_KEY  (keep secret)"
+echo "  Certificate : $CA_CRT  (distribute to all fleet nodes)"
+echo ""
+openssl x509 -in "$CA_CRT" -noout -subject -dates

scripts/queue_health_check.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""
+Queue Health Check — Verify dispatch queue is operational.
+
+Checks:
+1. Queue file exists and is readable
+2. Queue has pending items
+3. Queue is not stuck (items processing)
+4. Queue age (stale items)
+
+Usage:
+    python scripts/queue_health_check.py
+    python scripts/queue_health_check.py --json
+"""
+
+import json
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+
+def check_queue_health(queue_path: str = "~/.hermes/queue.json") -> dict:
+    """Check queue health status."""
+    path = Path(queue_path).expanduser()
+    
+    result = {
+        "healthy": True,
+        "checks": {},
+        "warnings": [],
+        "errors": []
+    }
+    
+    # Check 1: File exists
+    if not path.exists():
+        result["healthy"] = False
+        result["errors"].append(f"Queue file not found: {path}")
+        result["checks"]["file_exists"] = False
+        return result
+    
+    result["checks"]["file_exists"] = True
+    
+    # Check 2: File is readable
+    try:
+        with open(path, "r") as f:
+            data = json.load(f)
+    except Exception as e:
+        result["healthy"] = False
+        result["errors"].append(f"Cannot read queue: {e}")
+        result["checks"]["readable"] = False
+        return result
+    
+    result["checks"]["readable"] = True
+    
+    # Check 3: Queue structure
+    if not isinstance(data, dict):
+        result["healthy"] = False
+        result["errors"].append("Queue is not a dict")
+        result["checks"]["valid_structure"] = False
+        return result
+    
+    result["checks"]["valid_structure"] = True
+    
+    # Check 4: Pending items
+    pending = data.get("pending", [])
+    processing = data.get("processing", [])
+    completed = data.get("completed", [])
+    
+    result["checks"]["pending_count"] = len(pending)
+    result["checks"]["processing_count"] = len(processing)
+    result["checks"]["completed_count"] = len(completed)
+    
+    if len(pending) == 0 and len(processing) == 0:
+        result["warnings"].append("Queue is empty")
+    
+    # Check 5: Stale processing items
+    now = datetime.now()
+    stale_threshold = timedelta(hours=1)
+    
+    for item in processing:
+        started = item.get("started_at")
+        if started:
+            try:
+                started_time = datetime.fromisoformat(started.replace("Z", "+00:00"))
+                if now - started_time > stale_threshold:
+                    result["warnings"].append(f"Stale item: {item.get('id', 'unknown')} (started {started})")
+            except:
+                pass
+    
+    # Check 6: Queue age
+    if pending:
+        oldest = min(pending, key=lambda x: x.get("added_at", ""))
+        added = oldest.get("added_at")
+        if added:
+            try:
+                added_time = datetime.fromisoformat(added.replace("Z", "+00:00"))
+                age = now - added_time
+                if age > timedelta(hours=24):
+                    result["warnings"].append(f"Old item in queue: {oldest.get('id', 'unknown')} (added {added})")
+            except:
+                pass
+    
+    return result
+
+
+def main():
+    """Main function."""
+    import argparse
+    
+    parser = argparse.ArgumentParser(description="Queue health check")
+    parser.add_argument("--queue", default="~/.hermes/queue.json", help="Queue file path")
+    parser.add_argument("--json", action="store_true", help="Output as JSON")
+    args = parser.parse_args()
+    
+    result = check_queue_health(args.queue)
+    
+    if args.json:
+        print(json.dumps(result, indent=2))
+    else:
+        print("Queue Health Check")
+        print("=" * 50)
+        print(f"Healthy: {'✓' if result['healthy'] else '✗'}")
+        print()
+        
+        print("Checks:")
+        for check, value in result["checks"].items():
+            if isinstance(value, bool):
+                print(f"  {check}: {'✓' if value else '✗'}")
+            else:
+                print(f"  {check}: {value}")
+        
+        if result["warnings"]:
+            print()
+            print("Warnings:")
+            for warning in result["warnings"]:
+                print(f"  ⚠ {warning}")
+        
+        if result["errors"]:
+            print()
+            print("Errors:")
+            for error in result["errors"]:
+                print(f"  ✗ {error}")
+    
+    sys.exit(0 if result["healthy"] else 1)
+
+
+if __name__ == "__main__":
+    main()

tests/agent/test_a2a_mtls.py

@@ -0,0 +1,574 @@
+"""
+Tests for A2A mutual-TLS authentication.
+
+Scenarios covered:
+  - authorized agent (valid fleet-CA-signed cert) is accepted
+  - unauthorized agent (self-signed cert) is rejected with SSLError
+  - missing client cert is rejected
+  - build_server_ssl_context raises FileNotFoundError for missing paths
+  - build_client_ssl_context raises FileNotFoundError for missing paths
+  - A2AServer.start() / stop() lifecycle (no network I/O)
+
+All TLS I/O is done in-process against a loopback server so no ports need
+to be opened on a CI runner.
+
+Refs #806
+"""
+
+from __future__ import annotations
+
+import datetime
+import ipaddress
+import ssl
+import threading
+import time
+import urllib.request
+import urllib.error
+from pathlib import Path
+from typing import Tuple
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Helpers — generate self-signed certs in-memory with Python's ``cryptography``
+# library (dev extra).  If cryptography is unavailable we skip the network
+# tests gracefully.
+# ---------------------------------------------------------------------------
+
+try:
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes, serialization
+    from cryptography.hazmat.primitives.asymmetric import rsa
+    from cryptography.x509.oid import NameOID
+    import cryptography.hazmat.backends as _backends
+    _CRYPTO_AVAILABLE = True
+except ImportError:
+    _CRYPTO_AVAILABLE = False
+
+_requires_crypto = pytest.mark.skipif(
+    not _CRYPTO_AVAILABLE,
+    reason="cryptography package not installed",
+)
+
+# ---------------------------------------------------------------------------
+# Fixture helpers
+# ---------------------------------------------------------------------------
+
+def _make_ca_keypair(tmp_path: Path) -> Tuple[Path, Path]:
+    """Generate a self-signed CA cert+key and write to *tmp_path*."""
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    name = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, "Test Fleet CA"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "TestOrg"),
+    ])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=3650))
+        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=False, key_cert_sign=True, crl_sign=True,
+                content_commitment=False, key_encipherment=False,
+                data_encipherment=False, key_agreement=False,
+                encipher_only=False, decipher_only=False,
+            ),
+            critical=True,
+        )
+        .sign(key, hashes.SHA256())
+    )
+    key_path = tmp_path / "ca.key"
+    cert_path = tmp_path / "ca.crt"
+    key_path.write_bytes(key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.TraditionalOpenSSL,
+        serialization.NoEncryption(),
+    ))
+    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    return cert_path, key_path
+
+
+def _make_agent_keypair(
+    tmp_path: Path,
+    name: str,
+    ca_cert_path: Path,
+    ca_key_path: Path,
+) -> Tuple[Path, Path]:
+    """Generate an agent cert signed by the test CA."""
+    ca_cert = x509.load_pem_x509_certificate(ca_cert_path.read_bytes())
+    ca_key = serialization.load_pem_private_key(
+        ca_key_path.read_bytes(), password=None
+    )
+
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, f"{name}.fleet.hermes"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "TestOrg"),
+    ])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(ca_cert.subject)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=365))
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+        .add_extension(
+            x509.SubjectAlternativeName([
+                x509.DNSName(f"{name}.fleet.hermes"),
+                x509.DNSName(name),
+                x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
+            ]),
+            critical=False,
+        )
+        .add_extension(
+            x509.ExtendedKeyUsage([
+                x509.ExtendedKeyUsageOID.CLIENT_AUTH,
+                x509.ExtendedKeyUsageOID.SERVER_AUTH,
+            ]),
+            critical=False,
+        )
+        .sign(ca_key, hashes.SHA256())
+    )
+    key_path = tmp_path / f"{name}.key"
+    cert_path = tmp_path / f"{name}.crt"
+    key_path.write_bytes(key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.TraditionalOpenSSL,
+        serialization.NoEncryption(),
+    ))
+    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    return cert_path, key_path
+
+
+def _make_self_signed_keypair(tmp_path: Path, name: str) -> Tuple[Path, Path]:
+    """Generate a self-signed cert NOT signed by the test CA (unauthorized)."""
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, f"{name}.rogue"),
+    ])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(subject)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=365))
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+        .add_extension(
+            x509.SubjectAlternativeName([x509.IPAddress(ipaddress.IPv4Address("127.0.0.1"))]),
+            critical=False,
+        )
+        .sign(key, hashes.SHA256())
+    )
+    key_path = tmp_path / f"{name}_rogue.key"
+    cert_path = tmp_path / f"{name}_rogue.crt"
+    key_path.write_bytes(key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.TraditionalOpenSSL,
+        serialization.NoEncryption(),
+    ))
+    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    return cert_path, key_path
+
+
+# ---------------------------------------------------------------------------
+# Unit tests — no network I/O
+# ---------------------------------------------------------------------------
+
+class TestBuildSslContextErrors:
+    def test_server_context_missing_cert(self, tmp_path):
+        from agent.a2a_mtls import build_server_ssl_context
+        with pytest.raises(FileNotFoundError, match="mTLS"):
+            build_server_ssl_context(
+                cert=tmp_path / "nope.crt",
+                key=tmp_path / "nope.key",
+                ca=tmp_path / "nope.crt",
+            )
+
+    def test_client_context_missing_cert(self, tmp_path):
+        from agent.a2a_mtls import build_client_ssl_context
+        with pytest.raises(FileNotFoundError, match="mTLS client"):
+            build_client_ssl_context(
+                cert=tmp_path / "nope.crt",
+                key=tmp_path / "nope.key",
+                ca=tmp_path / "nope.crt",
+            )
+
+    @_requires_crypto
+    def test_server_context_builds_with_valid_certs(self, tmp_path):
+        from agent.a2a_mtls import build_server_ssl_context
+        ca_dir = tmp_path / "ca"
+        ca_dir.mkdir()
+        ca_crt, ca_key = _make_ca_keypair(ca_dir)
+        srv_crt, srv_key = _make_agent_keypair(
+            tmp_path, "srv", ca_crt, ca_key
+        )
+        ctx = build_server_ssl_context(cert=srv_crt, key=srv_key, ca=ca_crt)
+        assert isinstance(ctx, ssl.SSLContext)
+        assert ctx.verify_mode == ssl.CERT_REQUIRED
+
+    @_requires_crypto
+    def test_client_context_builds_with_valid_certs(self, tmp_path):
+        from agent.a2a_mtls import build_client_ssl_context
+        ca_dir = tmp_path / "ca"
+        ca_dir.mkdir()
+        ca_crt, ca_key = _make_ca_keypair(ca_dir)
+        cli_crt, cli_key = _make_agent_keypair(
+            tmp_path, "cli", ca_crt, ca_key
+        )
+        ctx = build_client_ssl_context(cert=cli_crt, key=cli_key, ca=ca_crt)
+        assert isinstance(ctx, ssl.SSLContext)
+        assert ctx.verify_mode == ssl.CERT_REQUIRED
+
+
+# ---------------------------------------------------------------------------
+# Integration tests — loopback mTLS server
+# ---------------------------------------------------------------------------
+
+def _find_free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _https_get(url: str, ssl_ctx: ssl.SSLContext) -> int:
+    """Return the HTTP status code for a GET request, or raise SSLError."""
+    req = urllib.request.urlopen(url, context=ssl_ctx, timeout=5)
+    return req.status
+
+
+@_requires_crypto
+class TestMutualTLSAuth:
+    """End-to-end mTLS auth over a loopback connection."""
+
+    @pytest.fixture(autouse=True)
+    def _pki(self, tmp_path):
+        """Set up a fleet CA and agent certs for timmy (server) and allegro (authorized client)."""
+        ca_dir = tmp_path / "ca"
+        ca_dir.mkdir()
+        self.ca_crt, self.ca_key = _make_ca_keypair(ca_dir)
+
+        agent_dir = tmp_path / "agents"
+        agent_dir.mkdir()
+
+        # Server agent: timmy
+        self.srv_crt, self.srv_key = _make_agent_keypair(
+            agent_dir, "timmy", self.ca_crt, self.ca_key
+        )
+        # Authorized client agent: allegro
+        self.cli_crt, self.cli_key = _make_agent_keypair(
+            agent_dir, "allegro", self.ca_crt, self.ca_key
+        )
+        # Unauthorized (self-signed) client: rogue
+        self.rogue_crt, self.rogue_key = _make_self_signed_keypair(agent_dir, "rogue")
+
+    @pytest.fixture()
+    def running_server(self):
+        """Start an A2AServer on a free loopback port, yield the URL, stop after test."""
+        from agent.a2a_mtls import A2AServer
+        port = _find_free_port()
+        server = A2AServer(
+            cert=self.srv_crt,
+            key=self.srv_key,
+            ca=self.ca_crt,
+            host="127.0.0.1",
+            port=port,
+        )
+        server.start(daemon=True)
+        time.sleep(0.15)   # let the thread bind
+        yield f"https://127.0.0.1:{port}"
+        server.stop()
+
+    def _authorized_ctx(self) -> ssl.SSLContext:
+        from agent.a2a_mtls import build_client_ssl_context
+        ctx = build_client_ssl_context(
+            cert=self.cli_crt, key=self.cli_key, ca=self.ca_crt
+        )
+        ctx.check_hostname = False   # loopback IP doesn't match DNS SAN
+        return ctx
+
+    def _unauthorized_ctx(self) -> ssl.SSLContext:
+        """Client context with a self-signed cert not trusted by the server CA."""
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+        ctx.load_cert_chain(certfile=str(self.rogue_crt), keyfile=str(self.rogue_key))
+        # Load the real fleet CA so server cert is accepted — but our client
+        # cert is self-signed and will be rejected by the server.
+        ctx.load_verify_locations(cafile=str(self.ca_crt))
+        ctx.check_hostname = False
+        return ctx
+
+    def _no_client_cert_ctx(self) -> ssl.SSLContext:
+        """Client context with no client certificate at all."""
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+        ctx.load_verify_locations(cafile=str(self.ca_crt))
+        ctx.check_hostname = False
+        return ctx
+
+    # ------------------------------------------------------------------
+    # Authorized agent accepted
+    # ------------------------------------------------------------------
+
+    def test_authorized_agent_accepted(self, running_server):
+        """An agent with a fleet-CA-signed cert gets a 200-range response."""
+        status = _https_get(
+            running_server + "/.well-known/agent-card.json",
+            self._authorized_ctx(),
+        )
+        assert status == 200
+
+    def test_authorized_agent_task_endpoint(self, running_server):
+        """POST /a2a/task returns 202 for an authorized agent."""
+        import urllib.request
+        req = urllib.request.Request(
+            running_server + "/a2a/task",
+            data=b'{"hello":"world"}',
+            method="POST",
+        )
+        req.add_header("Content-Type", "application/json")
+        resp = urllib.request.urlopen(req, context=self._authorized_ctx(), timeout=5)
+        assert resp.status == 202
+
+    # ------------------------------------------------------------------
+    # Unauthorized agent rejected
+    # ------------------------------------------------------------------
+
+    def test_unauthorized_agent_rejected(self, running_server):
+        """A self-signed cert not signed by the fleet CA is rejected at TLS handshake."""
+        with pytest.raises((ssl.SSLError, OSError)):
+            _https_get(running_server + "/", self._unauthorized_ctx())
+
+    def test_no_client_cert_rejected(self, running_server):
+        """A client with no cert at all is rejected at TLS handshake."""
+        with pytest.raises((ssl.SSLError, OSError)):
+            _https_get(running_server + "/", self._no_client_cert_ctx())
+
+    # ------------------------------------------------------------------
+    # Server lifecycle
+    # ------------------------------------------------------------------
+
+    def test_server_stop_is_idempotent(self):
+        """Calling stop() twice does not raise."""
+        from agent.a2a_mtls import A2AServer
+        port = _find_free_port()
+        server = A2AServer(
+            cert=self.srv_crt, key=self.srv_key, ca=self.ca_crt,
+            host="127.0.0.1", port=port,
+        )
+        server.start(daemon=True)
+        time.sleep(0.1)
+        server.stop()
+        server.stop()   # second call must not raise
+
+
+# ---------------------------------------------------------------------------
+# server_from_env() — environment variable wiring
+# ---------------------------------------------------------------------------
+
+class TestServerFromEnv:
+    def test_reads_env_vars(self, tmp_path, monkeypatch):
+        # Create dummy files so FileNotFoundError isn't triggered
+        cert = tmp_path / "a.crt"
+        key = tmp_path / "a.key"
+        ca = tmp_path / "ca.crt"
+        for f in (cert, key, ca):
+            f.write_text("PLACEHOLDER")
+
+        monkeypatch.setenv("HERMES_A2A_CERT", str(cert))
+        monkeypatch.setenv("HERMES_A2A_KEY", str(key))
+        monkeypatch.setenv("HERMES_A2A_CA", str(ca))
+        monkeypatch.setenv("HERMES_A2A_HOST", "127.0.0.2")
+        monkeypatch.setenv("HERMES_A2A_PORT", "19443")
+
+        from agent.a2a_mtls import server_from_env
+        srv = server_from_env()
+        assert srv.cert == cert
+        assert srv.key == key
+        assert srv.ca == ca
+        assert srv.host == "127.0.0.2"
+        assert srv.port == 19443
+
+    def test_uses_agent_name_for_defaults(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setenv("HERMES_AGENT_NAME", "ezra")
+        # Unset explicit cert overrides
+        monkeypatch.delenv("HERMES_A2A_CERT", raising=False)
+        monkeypatch.delenv("HERMES_A2A_KEY", raising=False)
+        monkeypatch.delenv("HERMES_A2A_CA", raising=False)
+
+        from agent.a2a_mtls import server_from_env
+        srv = server_from_env()
+        assert "ezra" in str(srv.cert)
+        assert "ezra" in str(srv.key)
+        assert "fleet-ca" in str(srv.ca)
+
+
+# ---------------------------------------------------------------------------
+# A2AMTLSServer and A2AMTLSClient — routing server + client helper
+# ---------------------------------------------------------------------------
+
+@_requires_crypto
+class TestA2AMTLSServerAndClient:
+    """Tests for the routing-based A2AMTLSServer and A2AMTLSClient."""
+
+    @pytest.fixture(autouse=True)
+    def _pki(self, tmp_path):
+        ca_dir = tmp_path / "ca"
+        ca_dir.mkdir()
+        self.ca_crt, self.ca_key = _make_ca_keypair(ca_dir)
+        agent_dir = tmp_path / "agents"
+        agent_dir.mkdir()
+        self.srv_crt, self.srv_key = _make_agent_keypair(
+            agent_dir, "timmy", self.ca_crt, self.ca_key
+        )
+        self.cli_crt, self.cli_key = _make_agent_keypair(
+            agent_dir, "allegro", self.ca_crt, self.ca_key
+        )
+        self.rogue_crt, self.rogue_key = _make_self_signed_keypair(agent_dir, "rogue")
+
+    @pytest.fixture()
+    def routing_server(self):
+        from agent.a2a_mtls import A2AMTLSServer
+        port = _find_free_port()
+        server = A2AMTLSServer(
+            cert=self.srv_crt, key=self.srv_key, ca=self.ca_crt,
+            host="127.0.0.1", port=port,
+        )
+        server.add_route("/echo", lambda p, *, peer_cn=None: {"echo": p, "peer": peer_cn})
+        server.add_route("/tasks/send", lambda p, *, peer_cn=None: {"status": "ok", "echo": p})
+        with server:
+            time.sleep(0.1)
+            yield server, port
+
+    def _authorized_ctx(self) -> ssl.SSLContext:
+        from agent.a2a_mtls import build_client_ssl_context
+        ctx = build_client_ssl_context(
+            cert=self.cli_crt, key=self.cli_key, ca=self.ca_crt
+        )
+        ctx.check_hostname = False
+        return ctx
+
+    def test_routing_server_get(self, routing_server):
+        server, port = routing_server
+        ctx = self._authorized_ctx()
+        req = urllib.request.Request(f"https://127.0.0.1:{port}/echo")
+        with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
+            import json
+            data = json.loads(resp.read())
+        assert data["peer"] is not None  # CN present
+
+    def test_routing_server_post_payload(self, routing_server):
+        server, port = routing_server
+        ctx = self._authorized_ctx()
+        import json
+        payload = {"task_id": "abc", "action": "delegate"}
+        req = urllib.request.Request(
+            f"https://127.0.0.1:{port}/tasks/send",
+            data=json.dumps(payload).encode(),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
+            data = json.loads(resp.read())
+        assert data["status"] == "ok"
+        assert data["echo"]["task_id"] == "abc"
+
+    def test_routing_server_unknown_route_404(self, routing_server):
+        server, port = routing_server
+        ctx = self._authorized_ctx()
+        req = urllib.request.Request(f"https://127.0.0.1:{port}/nonexistent")
+        with pytest.raises(urllib.error.URLError) as exc_info:
+            urllib.request.urlopen(req, context=ctx, timeout=5)
+        assert "404" in str(exc_info.value)
+
+    def test_routing_server_context_manager_stops(self):
+        from agent.a2a_mtls import A2AMTLSServer
+        port = _find_free_port()
+        server = A2AMTLSServer(
+            cert=self.srv_crt, key=self.srv_key, ca=self.ca_crt,
+            host="127.0.0.1", port=port,
+        )
+        server.add_route("/ping", lambda p, *, peer_cn=None: {"pong": True})
+        with server:
+            time.sleep(0.05)
+            assert server._httpd is not None
+        assert server._httpd is None  # stopped after __exit__
+
+    def test_routing_server_rogue_client_rejected(self, routing_server):
+        server, port = routing_server
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_verify_locations(cafile=str(self.ca_crt))
+        ctx.load_cert_chain(certfile=str(self.rogue_crt), keyfile=str(self.rogue_key))
+        ctx.check_hostname = False
+        req = urllib.request.Request(f"https://127.0.0.1:{port}/echo")
+        with pytest.raises((ssl.SSLError, OSError, urllib.error.URLError)):
+            urllib.request.urlopen(req, context=ctx, timeout=5)
+
+    def test_a2a_mtls_client_get(self, routing_server):
+        from agent.a2a_mtls import A2AMTLSClient
+        server, port = routing_server
+        client = A2AMTLSClient(
+            cert=self.cli_crt, key=self.cli_key, ca=self.ca_crt
+        )
+        result = client.get(f"https://127.0.0.1:{port}/echo")
+        assert result["peer"] is not None
+
+    def test_a2a_mtls_client_post(self, routing_server):
+        from agent.a2a_mtls import A2AMTLSClient
+        server, port = routing_server
+        client = A2AMTLSClient(
+            cert=self.cli_crt, key=self.cli_key, ca=self.ca_crt
+        )
+        result = client.post(f"https://127.0.0.1:{port}/tasks/send", json={"x": 1})
+        assert result["status"] == "ok"
+        assert result["echo"]["x"] == 1
+
+    def test_a2a_mtls_client_rogue_cert_raises(self, routing_server):
+        from agent.a2a_mtls import A2AMTLSClient
+        server, port = routing_server
+        client = A2AMTLSClient(
+            cert=self.rogue_crt, key=self.rogue_key, ca=self.ca_crt
+        )
+        with pytest.raises((ConnectionError, ssl.SSLError, OSError)):
+            client.get(f"https://127.0.0.1:{port}/echo")
+
+    def test_concurrent_fleet_agents(self, routing_server):
+        """timmy (server) accepts concurrent connections from multiple authorized clients."""
+        from agent.a2a_mtls import build_client_ssl_context
+        server, port = routing_server
+        results: dict = {}
+        errors: dict = {}
+
+        def connect(name: str) -> None:
+            try:
+                ctx = build_client_ssl_context(
+                    cert=self.cli_crt, key=self.cli_key, ca=self.ca_crt
+                )
+                ctx.check_hostname = False
+                req = urllib.request.Request(f"https://127.0.0.1:{port}/echo")
+                with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
+                    import json
+                    results[name] = json.loads(resp.read())
+            except Exception as exc:
+                errors[name] = exc
+
+        threads = [threading.Thread(target=connect, args=(n,)) for n in ("t1", "t2", "t3")]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=10)
+
+        assert not errors, f"Concurrent connection errors: {errors}"
+        assert len(results) == 3

tests/agent/test_memory_provider.py

@@ -198,14 +198,14 @@ class TestMemoryManager:
     def test_prefetch_skips_empty(self):
         mgr = MemoryManager()
         p1 = FakeMemoryProvider("builtin")
-        p1._prefetch_result = "Has memories"
+        p1._prefetch_result = "This provider has meaningful memories with enough length"
         p2 = FakeMemoryProvider("external")
         p2._prefetch_result = ""
         mgr.add_provider(p1)
         mgr.add_provider(p2)
 
         result = mgr.prefetch_all("query")
-        assert result == "Has memories"
+        assert result == "This provider has meaningful memories with enough length"
 
     def test_queue_prefetch_all(self):
         mgr = MemoryManager()
@@ -695,3 +695,92 @@ class TestMemoryContextFencing:
         fence_end = combined.index("</memory-context>")
         assert "Alice" in combined[fence_start:fence_end]
         assert combined.index("weather") < fence_start
+
+
+class TestPrefetchFiltering:
+    """Tests for _filter_prefetch_lines and related helpers."""
+
+    def test_deduplicates_exact_lines(self):
+        from agent.memory_manager import _filter_prefetch_lines
+        raw = "- This is line one with enough characters\n- This is line two with enough characters\n- This is line one with enough characters\n- This is line three with enough characters"
+        result = _filter_prefetch_lines(raw)
+        lines = [l for l in result.splitlines() if l.strip()]
+        assert len(lines) == 3
+        assert "- This is line one with enough characters" in result
+        assert "- This is line two with enough characters" in result
+        assert "- This is line three with enough characters" in result
+
+    def test_removes_meta_instruction_debris(self):
+        from agent.memory_manager import _filter_prefetch_lines
+        raw = (
+            "## Fleet Memories\n"
+            "- > Focus on: was a non-trivial approach used\n"
+            "- > Focus on: was a non-trivial approach used\n"
+            "- Actual memory content about fleet ops\n"
+            "- Note: this is just a note\n"
+        )
+        result = _filter_prefetch_lines(raw)
+        assert "Focus on" not in result
+        assert "Note:" not in result
+        assert "Actual memory content about fleet ops" in result
+        assert "Fleet Memories" in result
+
+    def test_removes_low_signal_short_lines(self):
+        from agent.memory_manager import _filter_prefetch_lines
+        raw = (
+            "- \n"
+            "- x\n"
+            "- This is a meaningful memory entry with enough length\n"
+        )
+        result = _filter_prefetch_lines(raw)
+        assert "- x" not in result
+        assert "meaningful memory entry" in result
+
+    def test_preserves_structured_facts(self):
+        from agent.memory_manager import _filter_prefetch_lines
+        raw = (
+            "## Local Facts (Hologram)\n"
+            "- ALEXANDER: Prefers Gitea for reports and deliverables.\n"
+            "- Telegram home channel is Timmy Time.\n"
+        )
+        result = _filter_prefetch_lines(raw)
+        assert "ALEXANDER" in result
+        assert "Gitea" in result
+        assert "Telegram" in result
+
+    def test_is_meta_instruction_line(self):
+        from agent.memory_manager import _is_meta_instruction_line
+        assert _is_meta_instruction_line("- > Focus on: something") is True
+        assert _is_meta_instruction_line("- Focus on: something") is True
+        assert _is_meta_instruction_line("* Focus on: something") is True
+        assert _is_meta_instruction_line("- Actual user memory content") is False
+        assert _is_meta_instruction_line("ALEXANDER: Prefers Gitea") is False
+
+    def test_is_low_signal_line(self):
+        from agent.memory_manager import _is_low_signal_line
+        assert _is_low_signal_line("- ") is True
+        assert _is_low_signal_line("*") is True
+        assert _is_low_signal_line("- x") is True
+        assert _is_low_signal_line("- Short line") is True
+        assert _is_low_signal_line("- This is a long meaningful memory entry") is False
+
+    def test_prefetch_all_applies_filtering(self):
+        from agent.memory_manager import MemoryManager
+        mgr = MemoryManager()
+        fake = FakeMemoryProvider(name="test")
+        fake._prefetch_result = (
+            "- > Focus on: was a non-trivial approach\n"
+            "- > Focus on: was a non-trivial approach\n"
+            "- Real memory fact\n"
+        )
+        mgr.add_provider(fake)
+        result = mgr.prefetch_all("query")
+        assert "Focus on" not in result
+        assert "Real memory fact" in result
+        assert result.count("Real memory fact") == 1
+
+    def test_empty_prefetch_returns_empty(self):
+        from agent.memory_manager import _filter_prefetch_lines
+        assert _filter_prefetch_lines("") == ""
+        assert _filter_prefetch_lines("   ") == ""
+        assert _filter_prefetch_lines("\n\n") == ""

tests/gateway/test_config.py

@@ -10,6 +10,7 @@ from gateway.config import (
     PlatformConfig,
     SessionResetPolicy,
     _apply_env_overrides,
+    _validate_gateway_config,
     load_gateway_config,
 )
 
@@ -294,3 +295,151 @@ class TestHomeChannelEnvOverrides:
             home = config.platforms[platform].home_channel
             assert home is not None, f"{platform.value}: home_channel should not be None"
             assert (home.chat_id, home.name) == expected, platform.value
+
+
+class TestValidateGatewayConfig:
+    """Tests for _validate_gateway_config — in-place sanitisation of loaded config."""
+
+    # -- idle_minutes validation --
+
+    def test_idle_minutes_zero_is_corrected_to_default(self):
+        config = GatewayConfig()
+        config.default_reset_policy.idle_minutes = 0
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.idle_minutes == 1440
+
+    def test_idle_minutes_negative_is_corrected_to_default(self):
+        config = GatewayConfig()
+        config.default_reset_policy.idle_minutes = -60
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.idle_minutes == 1440
+
+    def test_idle_minutes_none_is_corrected_to_default(self):
+        config = GatewayConfig()
+        config.default_reset_policy.idle_minutes = None  # type: ignore[assignment]
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.idle_minutes == 1440
+
+    def test_valid_idle_minutes_is_unchanged(self):
+        config = GatewayConfig()
+        config.default_reset_policy.idle_minutes = 90
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.idle_minutes == 90
+
+    # -- at_hour validation --
+
+    def test_at_hour_too_high_is_corrected_to_default(self):
+        config = GatewayConfig()
+        config.default_reset_policy.at_hour = 24
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.at_hour == 4
+
+    def test_at_hour_negative_is_corrected_to_default(self):
+        config = GatewayConfig()
+        config.default_reset_policy.at_hour = -1
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.at_hour == 4
+
+    def test_valid_at_hour_is_unchanged(self):
+        config = GatewayConfig()
+        config.default_reset_policy.at_hour = 3
+        _validate_gateway_config(config)
+        assert config.default_reset_policy.at_hour == 3
+
+    def test_at_hour_boundary_values_are_valid(self):
+        for valid_hour in (0, 23):
+            config = GatewayConfig()
+            config.default_reset_policy.at_hour = valid_hour
+            _validate_gateway_config(config)
+            assert config.default_reset_policy.at_hour == valid_hour
+
+    # -- empty-token warning (enabled platforms) --
+
+    def test_empty_string_token_logs_warning(self, caplog):
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.TELEGRAM: PlatformConfig(enabled=True, token=""),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert any(
+            "TELEGRAM_BOT_TOKEN" in r.message and "empty" in r.message
+            for r in caplog.records
+        )
+
+    def test_disabled_platform_with_empty_token_no_warning(self, caplog):
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.TELEGRAM: PlatformConfig(enabled=False, token=""),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert not any("TELEGRAM_BOT_TOKEN" in r.message for r in caplog.records)
+
+    # -- API Server key / binding warnings --
+
+    def test_api_server_network_binding_without_key_logs_warning(self, caplog):
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.API_SERVER: PlatformConfig(
+                    enabled=True,
+                    extra={"host": "0.0.0.0"},
+                ),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert any(
+            "API_SERVER_KEY" in r.message for r in caplog.records
+        )
+
+    def test_api_server_loopback_without_key_no_warning(self, caplog):
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.API_SERVER: PlatformConfig(
+                    enabled=True,
+                    extra={"host": "127.0.0.1"},
+                ),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert not any(
+            "API_SERVER_KEY" in r.message for r in caplog.records
+        )
+
+    def test_api_server_network_binding_with_key_no_warning(self, caplog):
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.API_SERVER: PlatformConfig(
+                    enabled=True,
+                    extra={"host": "0.0.0.0", "key": "sk-real-key-here"},
+                ),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert not any(
+            "API_SERVER_KEY" in r.message for r in caplog.records
+        )
+
+    def test_api_server_default_loopback_without_key_no_warning(self, caplog):
+        """API server with no explicit host defaults to 127.0.0.1 — no warning."""
+        import logging
+        config = GatewayConfig(
+            platforms={
+                Platform.API_SERVER: PlatformConfig(enabled=True),
+            }
+        )
+        with caplog.at_level(logging.WARNING, logger="gateway.config"):
+            _validate_gateway_config(config)
+        assert not any(
+            "API_SERVER_KEY" in r.message for r in caplog.records
+        )

tests/test_circuit_breaker.py

@@ -0,0 +1,97 @@
+"""Tests for circuit breaker (#885)."""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from agent.circuit_breaker import CircuitBreaker, ToolCircuitBreaker, MultiToolCircuitBreaker, CircuitState
+
+
+def test_closed_allows_execution():
+    cb = CircuitBreaker(failure_threshold=3)
+    assert cb.can_execute()
+
+
+def test_opens_after_threshold():
+    cb = CircuitBreaker(failure_threshold=3)
+    cb.record_result(False)
+    cb.record_result(False)
+    assert cb.can_execute()  # Still closed at 2
+    cb.record_result(False)
+    assert not cb.can_execute()  # Open at 3
+
+
+def test_closes_on_success():
+    cb = CircuitBreaker(failure_threshold=3)
+    cb.record_result(False)
+    cb.record_result(True)
+    assert cb.consecutive_failures == 0
+
+
+def test_half_open_recovery():
+    cb = CircuitBreaker(failure_threshold=2, recovery_timeout=0.1, success_threshold=1)
+    cb.record_result(False)
+    cb.record_result(False)
+    assert cb.state == CircuitState.OPEN
+    
+    import time
+    time.sleep(0.15)
+    
+    assert cb.can_execute()  # Moved to half-open
+    cb.record_result(True)
+    assert cb.state == CircuitState.CLOSED
+
+
+def test_recovery_action_streak():
+    cb = ToolCircuitBreaker(failure_threshold=3)
+    for _ in range(5):
+        cb.record_result(False)
+    action = cb.get_recovery_action()
+    assert action["action"] == "switch_tool_type"
+
+
+def test_recovery_action_critical():
+    cb = ToolCircuitBreaker(failure_threshold=3)
+    for _ in range(10):
+        cb.record_result(False)
+    action = cb.get_recovery_action()
+    assert action["action"] == "terminal_only"
+    assert action["severity"] == "critical"
+
+
+def test_multi_tool_breaker():
+    mcb = MultiToolCircuitBreaker()
+    mcb.record_result("read_file", False)
+    mcb.record_result("read_file", False)
+    mcb.record_result("read_file", False)
+    assert not mcb.can_execute("read_file")
+    assert mcb.can_execute("terminal")  # Different tool unaffected
+
+
+def test_global_state():
+    mcb = MultiToolCircuitBreaker()
+    mcb.record_result("tool_a", False)
+    mcb.record_result("tool_b", False)
+    state = mcb.get_global_state()
+    assert state["global_streak"] == 2
+
+
+def test_reset():
+    cb = CircuitBreaker(failure_threshold=2)
+    cb.record_result(False)
+    cb.record_result(False)
+    assert cb.state == CircuitState.OPEN
+    cb.reset()
+    assert cb.state == CircuitState.CLOSED
+
+
+if __name__ == "__main__":
+    tests = [test_closed_allows_execution, test_opens_after_threshold,
+             test_closes_on_success, test_half_open_recovery,
+             test_recovery_action_streak, test_recovery_action_critical,
+             test_multi_tool_breaker, test_global_state, test_reset]
+    for t in tests:
+        print(f"Running {t.__name__}...")
+        t()
+        print("  PASS")
+    print("\nAll tests passed.")

tests/test_mtls.py

@@ -0,0 +1,389 @@
+"""
+Tests for agent/mtls.py — mutual TLS between fleet agents.
+
+Covers:
+- is_mtls_configured() with various env combinations
+- build_server_ssl_context() / build_client_ssl_context() with real certs
+- MTLSMiddleware: authorized agent accepted, unauthorized agent rejected
+"""
+
+import ssl
+import datetime
+import ipaddress
+import os
+import pytest
+from pathlib import Path
+from unittest.mock import patch
+
+# ---------------------------------------------------------------------------
+# Helpers: generate real in-memory certs using the `cryptography` library
+# ---------------------------------------------------------------------------
+
+try:
+    from cryptography import x509
+    from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
+    from cryptography.hazmat.primitives import hashes, serialization
+    from cryptography.hazmat.primitives.asymmetric import rsa
+    _CRYPTO_AVAILABLE = True
+except ImportError:
+    _CRYPTO_AVAILABLE = False
+
+pytestmark = pytest.mark.skipif(
+    not _CRYPTO_AVAILABLE,
+    reason="cryptography package required for mTLS tests",
+)
+
+
+def _make_key():
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+def _write_pem(path: Path, data: bytes) -> None:
+    path.write_bytes(data)
+    path.chmod(0o600)
+
+
+def make_fleet_pki(tmp_path: Path):
+    """
+    Create a minimal Fleet PKI in tmp_path:
+      - fleet-ca.key / fleet-ca.crt  (self-signed CA)
+      - agent.key / agent.crt        (signed by fleet CA, CN=test-agent)
+      - rogue.key / rogue.crt        (self-signed, NOT signed by fleet CA)
+
+    Returns a dict of Path objects.
+    """
+    now = datetime.datetime.now(datetime.timezone.utc)
+
+    # --- Fleet CA ---
+    ca_key = _make_key()
+    ca_name = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, "Hermes Fleet CA"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Hermes Fleet"),
+    ])
+    ca_cert = (
+        x509.CertificateBuilder()
+        .subject_name(ca_name)
+        .issuer_name(ca_name)
+        .public_key(ca_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=3650))
+        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=False, content_commitment=False,
+                key_encipherment=False, data_encipherment=False,
+                key_agreement=False, key_cert_sign=True, crl_sign=True,
+                encipher_only=False, decipher_only=False,
+            ),
+            critical=True,
+        )
+        .sign(ca_key, hashes.SHA256())
+    )
+
+    # --- Fleet agent cert ---
+    agent_key = _make_key()
+    agent_name = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, "test-agent"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Hermes Fleet"),
+    ])
+    agent_cert = (
+        x509.CertificateBuilder()
+        .subject_name(agent_name)
+        .issuer_name(ca_name)
+        .public_key(agent_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=730))
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+        .add_extension(
+            x509.SubjectAlternativeName([
+                x509.DNSName("test-agent"),
+                x509.DNSName("localhost"),
+                x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
+            ]),
+            critical=False,
+        )
+        .add_extension(
+            x509.ExtendedKeyUsage([
+                ExtendedKeyUsageOID.CLIENT_AUTH,
+                ExtendedKeyUsageOID.SERVER_AUTH,
+            ]),
+            critical=False,
+        )
+        .sign(ca_key, hashes.SHA256())
+    )
+
+    # --- Rogue cert (self-signed, not from fleet CA) ---
+    rogue_key = _make_key()
+    rogue_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "rogue-agent")])
+    rogue_cert = (
+        x509.CertificateBuilder()
+        .subject_name(rogue_name)
+        .issuer_name(rogue_name)
+        .public_key(rogue_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now)
+        .not_valid_after(now + datetime.timedelta(days=365))
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+        .sign(rogue_key, hashes.SHA256())
+    )
+
+    # Write to tmp_path
+    pem = serialization.Encoding.PEM
+    private_fmt = serialization.PrivateFormat.TraditionalOpenSSL
+    no_enc = serialization.NoEncryption()
+
+    paths = {}
+
+    paths["ca_key"] = tmp_path / "fleet-ca.key"
+    _write_pem(paths["ca_key"], ca_key.private_bytes(pem, private_fmt, no_enc))
+
+    paths["ca_cert"] = tmp_path / "fleet-ca.crt"
+    _write_pem(paths["ca_cert"], ca_cert.public_bytes(pem))
+
+    paths["agent_key"] = tmp_path / "agent.key"
+    _write_pem(paths["agent_key"], agent_key.private_bytes(pem, private_fmt, no_enc))
+
+    paths["agent_cert"] = tmp_path / "agent.crt"
+    _write_pem(paths["agent_cert"], agent_cert.public_bytes(pem))
+
+    paths["rogue_key"] = tmp_path / "rogue.key"
+    _write_pem(paths["rogue_key"], rogue_key.private_bytes(pem, private_fmt, no_enc))
+
+    paths["rogue_cert"] = tmp_path / "rogue.crt"
+    _write_pem(paths["rogue_cert"], rogue_cert.public_bytes(pem))
+
+    return paths
+
+
+# ---------------------------------------------------------------------------
+# Tests: is_mtls_configured
+# ---------------------------------------------------------------------------
+
+class TestIsMtlsConfigured:
+    def test_all_vars_missing(self):
+        from agent.mtls import is_mtls_configured
+        env = {k: "" for k in ("HERMES_MTLS_CERT", "HERMES_MTLS_KEY", "HERMES_MTLS_CA")}
+        with patch.dict(os.environ, env, clear=False):
+            assert not is_mtls_configured()
+
+    def test_partial_vars(self, tmp_path):
+        from agent.mtls import is_mtls_configured
+        f = tmp_path / "cert.pem"
+        f.write_text("x")
+        env = {"HERMES_MTLS_CERT": str(f), "HERMES_MTLS_KEY": "", "HERMES_MTLS_CA": ""}
+        with patch.dict(os.environ, env, clear=False):
+            assert not is_mtls_configured()
+
+    def test_all_vars_set_but_file_missing(self, tmp_path):
+        from agent.mtls import is_mtls_configured
+        env = {
+            "HERMES_MTLS_CERT": str(tmp_path / "no.crt"),
+            "HERMES_MTLS_KEY": str(tmp_path / "no.key"),
+            "HERMES_MTLS_CA": str(tmp_path / "no-ca.crt"),
+        }
+        with patch.dict(os.environ, env, clear=False):
+            assert not is_mtls_configured()
+
+    def test_all_vars_set_and_files_exist(self, tmp_path):
+        from agent.mtls import is_mtls_configured
+        for name in ("cert.pem", "key.pem", "ca.pem"):
+            (tmp_path / name).write_text("x")
+        env = {
+            "HERMES_MTLS_CERT": str(tmp_path / "cert.pem"),
+            "HERMES_MTLS_KEY": str(tmp_path / "key.pem"),
+            "HERMES_MTLS_CA": str(tmp_path / "ca.pem"),
+        }
+        with patch.dict(os.environ, env, clear=False):
+            assert is_mtls_configured()
+
+
+# ---------------------------------------------------------------------------
+# Tests: build_server_ssl_context / build_client_ssl_context
+# ---------------------------------------------------------------------------
+
+class TestBuildSslContexts:
+    def test_raises_when_not_configured(self):
+        from agent.mtls import build_server_ssl_context, build_client_ssl_context
+        env = {"HERMES_MTLS_CERT": "", "HERMES_MTLS_KEY": "", "HERMES_MTLS_CA": ""}
+        with patch.dict(os.environ, env, clear=False):
+            with pytest.raises(RuntimeError, match="not configured"):
+                build_server_ssl_context()
+            with pytest.raises(RuntimeError, match="not configured"):
+                build_client_ssl_context()
+
+    def test_server_context_requires_client_cert(self, tmp_path):
+        from agent.mtls import build_server_ssl_context
+        pki = make_fleet_pki(tmp_path)
+        env = {
+            "HERMES_MTLS_CERT": str(pki["agent_cert"]),
+            "HERMES_MTLS_KEY": str(pki["agent_key"]),
+            "HERMES_MTLS_CA": str(pki["ca_cert"]),
+        }
+        with patch.dict(os.environ, env, clear=False):
+            ctx = build_server_ssl_context()
+        assert isinstance(ctx, ssl.SSLContext)
+        assert ctx.verify_mode == ssl.CERT_REQUIRED
+
+    def test_client_context_has_cert_required(self, tmp_path):
+        from agent.mtls import build_client_ssl_context
+        pki = make_fleet_pki(tmp_path)
+        env = {
+            "HERMES_MTLS_CERT": str(pki["agent_cert"]),
+            "HERMES_MTLS_KEY": str(pki["agent_key"]),
+            "HERMES_MTLS_CA": str(pki["ca_cert"]),
+        }
+        with patch.dict(os.environ, env, clear=False):
+            ctx = build_client_ssl_context()
+        assert isinstance(ctx, ssl.SSLContext)
+        assert ctx.verify_mode == ssl.CERT_REQUIRED
+
+
+# ---------------------------------------------------------------------------
+# Tests: MTLSMiddleware
+# ---------------------------------------------------------------------------
+
+def _make_scope(path: str, peer_cert=None) -> dict:
+    """Build a minimal ASGI HTTP scope, optionally with a fake TLS peer_cert."""
+    scope = {
+        "type": "http",
+        "path": path,
+        "extensions": {},
+    }
+    if peer_cert is not None:
+        scope["extensions"]["tls"] = {"peer_cert": peer_cert}
+    return scope
+
+
+async def _collect_response(middleware, scope):
+    """Drive the middleware and capture (status, body)."""
+    status = None
+    body = b""
+
+    async def receive():
+        return {"type": "http.request", "body": b""}
+
+    async def send(event):
+        nonlocal status, body
+        if event["type"] == "http.response.start":
+            status = event["status"]
+        elif event["type"] == "http.response.body":
+            body += event.get("body", b"")
+
+    await middleware(scope, receive, send)
+    return status, body
+
+
+class TestMTLSMiddleware:
+    """
+    Unit-test the MTLSMiddleware without spinning up a real server.
+    We inject mTLS configuration through env-var patching so the middleware
+    believes it is enabled, and use the ASGI scope's tls extension to simulate
+    whether a client cert was presented.
+    """
+
+    def _make_middleware(self, tmp_path, app=None):
+        """Return a configured MTLSMiddleware backed by real-looking cert files."""
+        from agent.mtls import MTLSMiddleware
+
+        for name in ("cert.pem", "key.pem", "ca.pem"):
+            (tmp_path / name).write_text("x")
+
+        env = {
+            "HERMES_MTLS_CERT": str(tmp_path / "cert.pem"),
+            "HERMES_MTLS_KEY": str(tmp_path / "key.pem"),
+            "HERMES_MTLS_CA": str(tmp_path / "ca.pem"),
+        }
+
+        async def passthrough(scope, receive, send):
+            await send({"type": "http.response.start", "status": 200, "headers": []})
+            await send({"type": "http.response.body", "body": b"ok"})
+
+        with patch.dict(os.environ, env, clear=False):
+            mw = MTLSMiddleware(app or passthrough)
+        return mw
+
+    @pytest.mark.asyncio
+    async def test_authorized_agent_accepted(self, tmp_path):
+        """An A2A route with a valid client cert passes through (200)."""
+        mw = self._make_middleware(tmp_path)
+        scope = _make_scope("/.well-known/agent-card.json", peer_cert={"subject": (("commonName", "timmy"),)})
+        status, body = await _collect_response(mw, scope)
+        assert status == 200
+
+    @pytest.mark.asyncio
+    async def test_unauthorized_agent_rejected(self, tmp_path):
+        """An A2A route with NO client cert is rejected (403)."""
+        mw = self._make_middleware(tmp_path)
+        scope = _make_scope("/.well-known/agent-card.json", peer_cert=None)
+        status, body = await _collect_response(mw, scope)
+        assert status == 403
+        assert b"certificate" in body.lower()
+
+    @pytest.mark.asyncio
+    async def test_non_a2a_route_not_gated(self, tmp_path):
+        """Non-A2A routes (like /api/status) pass through even without a cert."""
+        mw = self._make_middleware(tmp_path)
+        scope = _make_scope("/api/status", peer_cert=None)
+        status, body = await _collect_response(mw, scope)
+        assert status == 200
+
+    @pytest.mark.asyncio
+    async def test_agent_card_api_route_gated(self, tmp_path):
+        """The /api/agent-card route also requires a client cert."""
+        mw = self._make_middleware(tmp_path)
+        scope = _make_scope("/api/agent-card", peer_cert=None)
+        status, _ = await _collect_response(mw, scope)
+        assert status == 403
+
+    @pytest.mark.asyncio
+    async def test_middleware_disabled_when_not_configured(self):
+        """When mTLS env vars are absent, the middleware is a no-op."""
+        from agent.mtls import MTLSMiddleware
+
+        async def passthrough(scope, receive, send):
+            await send({"type": "http.response.start", "status": 200, "headers": []})
+            await send({"type": "http.response.body", "body": b"ok"})
+
+        env = {"HERMES_MTLS_CERT": "", "HERMES_MTLS_KEY": "", "HERMES_MTLS_CA": ""}
+        with patch.dict(os.environ, env, clear=False):
+            mw = MTLSMiddleware(passthrough)
+
+        # Even an A2A route with no cert should pass through
+        scope = _make_scope("/.well-known/agent-card.json", peer_cert=None)
+        status, _ = await _collect_response(mw, scope)
+        assert status == 200
+
+
+# ---------------------------------------------------------------------------
+# Tests: get_peer_cn
+# ---------------------------------------------------------------------------
+
+class TestGetPeerCn:
+    def test_returns_cn_from_subject(self):
+        from agent.mtls import get_peer_cn
+
+        class FakeSSL:
+            def getpeercert(self):
+                return {"subject": ((("commonName", "timmy"),),)}
+
+        assert get_peer_cn(FakeSSL()) == "timmy"
+
+    def test_returns_none_when_no_cert(self):
+        from agent.mtls import get_peer_cn
+
+        class FakeSSL:
+            def getpeercert(self):
+                return None
+
+        assert get_peer_cn(FakeSSL()) is None
+
+    def test_returns_none_on_exception(self):
+        from agent.mtls import get_peer_cn
+
+        class BrokenSSL:
+            def getpeercert(self):
+                raise RuntimeError("no ssl")
+
+        assert get_peer_cn(BrokenSSL()) is None

tests/test_tool_fixation_detector.py

@@ -0,0 +1,76 @@
+"""Tests for tool fixation detection."""
+
+import pytest
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from agent.tool_fixation_detector import ToolFixationDetector, get_fixation_detector
+
+
+class TestFixationDetection:
+    def test_no_fixation_below_threshold(self):
+        d = ToolFixationDetector(threshold=5)
+        for i in range(4):
+            assert d.record("execute_code") is None
+
+    def test_fixation_at_threshold(self):
+        d = ToolFixationDetector(threshold=3)
+        d.record("execute_code")
+        d.record("execute_code")
+        nudge = d.record("execute_code")
+        assert nudge is not None
+        assert "execute_code" in nudge
+        assert "3 times" in nudge
+
+    def test_fixation_above_threshold(self):
+        d = ToolFixationDetector(threshold=3)
+        d.record("execute_code")
+        d.record("execute_code")
+        d.record("execute_code")  # threshold hit
+        nudge = d.record("execute_code")  # still nudging
+        assert nudge is not None
+
+    def test_streak_resets_on_different_tool(self):
+        d = ToolFixationDetector(threshold=3)
+        d.record("execute_code")
+        d.record("execute_code")
+        d.record("terminal")  # breaks streak
+        assert d._streak_count == 1
+        assert d._current_streak == "terminal"
+
+    def test_nudges_sent_counter(self):
+        d = ToolFixationDetector(threshold=2)
+        d.record("a")
+        d.record("a")  # nudge 1
+        d.record("a")  # nudge 2
+        assert d.nudges_sent == 2
+
+    def test_events_recorded(self):
+        d = ToolFixationDetector(threshold=2)
+        d.record("x")
+        d.record("x")
+        assert len(d.events) == 1
+        assert d.events[0].tool_name == "x"
+        assert d.events[0].streak_length == 2
+
+    def test_report(self):
+        d = ToolFixationDetector(threshold=2)
+        d.record("x")
+        d.record("x")
+        report = d.format_report()
+        assert "x" in report
+
+    def test_reset(self):
+        d = ToolFixationDetector(threshold=2)
+        d.record("x")
+        d.record("x")
+        d.reset()
+        assert d._streak_count == 0
+        assert d._current_streak == ""
+
+    def test_singleton(self):
+        d1 = get_fixation_detector()
+        d2 = get_fixation_detector()
+        assert d1 is d2

tools/hardcoded_path_guard.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""
+Hardcoded Path Guard — Poka-Yoke #921
+
+Detects and blocks hardcoded home-directory paths in tool arguments.
+These paths work on one machine but break on others, VPS deployments,
+or when HOME changes.
+
+Usage:
+    from tools.hardcoded_path_guard import check_path, validate_tool_args
+    
+    # Check a single path
+    err = check_path("/Users/apayne/.hermes/config.yaml")
+    
+    # Validate all path-like args in a tool call
+    clean_args, warnings = validate_tool_args("read_file", {"path": "/home/user/file.txt"})
+"""
+
+import os
+import re
+import json as _json
+from typing import Dict, List, Optional, Tuple, Any
+
+# Patterns that indicate hardcoded home directories
+HARDCODED_PATTERNS = [
+    (r"/Users/[\w.\-]+/", "macOS home directory (/Users/...)"),
+    (r"/home/[\w.\-]+/", "Linux home directory (/home/...)"),
+    (r"(?<![\w/])~/", "unexpanded tilde (~/)"),
+    (r"/root/", "root home directory (/root/)"),
+]
+
+_COMPILED_PATTERNS = [(re.compile(p), desc) for p, desc in HARDCODED_PATTERNS]
+_NOQA_PATTERN = re.compile(r"#\s*noqa:?\s*hardcoded-path-ok")
+
+_PATH_ARG_NAMES = frozenset({
+    "path", "file_path", "filepath", "dir", "directory", "dest", "source",
+    "input", "output", "src", "dst", "target", "location", "file",
+    "image_path", "script", "config", "log_file",
+})
+
+
+def has_hardcoded_path(text: str) -> Optional[str]:
+    if _NOQA_PATTERN.search(text):
+        return None
+    for pattern, desc in _COMPILED_PATTERNS:
+        if pattern.search(text):
+            return desc
+    return None
+
+
+def check_path(path_value: str) -> Optional[str]:
+    if not isinstance(path_value, str):
+        return None
+    match_desc = has_hardcoded_path(path_value)
+    if match_desc:
+        return (
+            f"Path contains hardcoded home directory ({match_desc}): '{path_value}'. "
+            f"Use $HOME, relative paths, or get_hermes_home(). "
+            f"Add '# noqa: hardcoded-path-ok' if intentional."
+        )
+    return None
+
+
+def validate_tool_args(tool_name: str, args: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
+    warnings = []
+    for key, value in args.items():
+        if key.lower() not in _PATH_ARG_NAMES:
+            continue
+        if isinstance(value, str):
+            err = check_path(value)
+            if err:
+                warnings.append(err)
+        elif isinstance(value, list):
+            for item in value:
+                if isinstance(item, str):
+                    err = check_path(item)
+                    if err:
+                        warnings.append(err)
+    return args, warnings
+
+
+def scan_source_for_violations(source_code: str, filename: str = "") -> List[Tuple[int, str, str]]:
+    violations = []
+    lines = source_code.split("\n")
+    for i, line in enumerate(lines, 1):
+        stripped = line.strip()
+        if stripped.startswith("#"):
+            if _NOQA_PATTERN.search(line):
+                continue
+            continue
+        if stripped.startswith("import ") or stripped.startswith("from "):
+            continue
+        for pattern, desc in _COMPILED_PATTERNS:
+            match = pattern.search(line)
+            if match:
+                if _NOQA_PATTERN.search(line):
+                    continue
+                violations.append((i, line.strip(), desc))
+                break
+    return violations
+
+
+def guard_tool_dispatch(tool_name: str, args: Dict[str, Any]) -> Optional[str]:
+    _, warnings = validate_tool_args(tool_name, args)
+    if warnings:
+        return _json.dumps({
+            "error": "Hardcoded home directory path detected",
+            "details": warnings,
+            "suggestion": "Use $HOME, relative paths, or get_hermes_home() instead of hardcoded paths.",
+            "pokayoke": True,
+            "rule": "hardcoded-path-guard"
+        })
+    return None

tools/local_inference_tool.py

@@ -0,0 +1,106 @@
+
+#!/usr/bin/env python3
+"""
+Local Inference Bridge — Fast-path for low-entropy LLM tasks.
+
+Detects local Ollama/llama-cpp instances and uses them for 'Auxiliary' tasks
+(summarization, extraction, simple verification) to reduce cloud dependency.
+"""
+
+import json
+import logging
+import os
+import requests
+from typing import Dict, List, Optional, Any
+from tools.registry import registry, tool_error, tool_result
+
+logger = logging.getLogger(__name__)
+
+LOCAL_INFERENCE_SCHEMA = {
+    "name": "local_inference",
+    "description": "Execute a task using a local inference engine (Ollama/llama-cpp) if available. Ideal for simple summarization, text cleanup, or data extraction where cloud-grade intelligence is overkill.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "prompt": {"type": "string", "description": "The task prompt."},
+            "system": {"type": "string", "description": "Optional system instruction."},
+            "engine": {"type": "string", "enum": ["auto", "ollama", "llama-cpp"], "default": "auto"}
+        },
+        "required": ["prompt"]
+    }
+}
+
+def detect_local_engine() -> Optional[Dict[str, str]]:
+    """Detect presence of local inference engines."""
+    # 1. Check Ollama (default port 11434)
+    try:
+        res = requests.get("http://localhost:11434/api/tags", timeout=1)
+        if res.status_code == 200:
+            return {"type": "ollama", "url": "http://localhost:11434"}
+    except:
+        pass
+
+    # 2. Check llama-cpp-python (commonly on 8000 or 8080)
+    for port in [8000, 8080]:
+        try:
+            res = requests.get(f"http://localhost:{port}/v1/models", timeout=1)
+            if res.status_code == 200:
+                return {"type": "llama-cpp", "url": f"http://localhost:{port}"}
+        except:
+            pass
+            
+    return None
+
+def run_local_task(prompt: str, system: str = None, engine: str = "auto"):
+    """Execute inference on a detected local engine."""
+    info = detect_local_engine()
+    if not info:
+        return tool_error("No local inference engine (Ollama or llama-cpp) detected on localhost.")
+
+    try:
+        if info["type"] == "ollama":
+            # Select first available model or default to gemma
+            models = requests.get(f"{info['url']}/api/tags").json().get("models", [])
+            model_name = models[0]["name"] if models else "gemma"
+            
+            payload = {
+                "model": model_name,
+                "prompt": prompt,
+                "stream": False
+            }
+            if system: payload["system"] = system
+            
+            res = requests.post(f"{info['url']}/api/generate", json=payload, timeout=60)
+            result = res.json().get("response", "")
+            return tool_result(engine="Ollama", model=model_name, response=result)
+
+        elif info["type"] == "llama-cpp":
+            payload = {
+                "model": "local-model",
+                "messages": [
+                    {"role": "system", "content": system or "You are a helpful assistant."},
+                    {"role": "user", "content": prompt}
+                ]
+            }
+            res = requests.post(f"{info['url']}/v1/chat/completions", json=payload, timeout=60)
+            result = res.json()["choices"][0]["message"]["content"]
+            return tool_result(engine="llama-cpp", response=result)
+
+    except Exception as e:
+        return tool_error(f"Local inference failed: {str(e)}")
+
+def _handle_local_inference(args, **kwargs):
+    return run_local_task(
+        prompt=args.get("prompt"),
+        system=args.get("system"),
+        engine=args.get("engine", "auto")
+    )
+
+registry.register(
+    name="local_inference",
+    toolset="inference",
+    schema=LOCAL_INFERENCE_SCHEMA,
+    handler=_handle_local_inference,
+    emoji="🏠"
+)
+    

tools/sovereign_scavenger.py

@@ -0,0 +1,86 @@
+
+#!/usr/bin/env python3
+"""
+Sovereign Scavenger — Autonomous Backlog Grooming.
+
+Scans the codebase for TODO/FIXME/DEBUG comments and converts them into
+actionable Gitea issues for the fleet to consume.
+"""
+
+import os
+import re
+import logging
+from typing import List, Dict, Any
+from tools.registry import registry, tool_error, tool_result
+
+logger = logging.getLogger(__name__)
+
+SCAVENGER_SCHEMA = {
+    "name": "sovereign_scavenger",
+    "description": "Scans the current directory for TODO, FIXME, or DEBUG comments. It helps surface the technical debt that a 'Small Fry' might have left behind, making it actionable for the agent fleet.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to scan (defaults to current directory).", "default": "."},
+            "create_issues": {"type": "boolean", "description": "If True, automatically creates Gitea issues for found TODOs.", "default": False}
+        }
+    }
+}
+
+def find_todos(root_path: str):
+    """Scan files for TODO patterns."""
+    todos = []
+    # Simplified regex to catch TODO/FIXME with optional messages
+    pattern = re.compile(r'#.*(TODO|FIXME|DEBUG|XXX)[:s]*(.*)', re.IGNORECASE)
+    
+    for root, dirs, files in os.walk(root_path):
+        # Skip hidden and annoying dirs
+        dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['node_modules', 'dist', '__pycache__']]
+        
+        for file in files:
+            if not file.endswith(('.py', '.ts', '.js', '.md', '.txt')):
+                continue
+            
+            filepath = os.path.join(root, file)
+            try:
+                with open(filepath, 'r', encoding='utf-8') as f:
+                    for i, line in enumerate(f, 1):
+                        match = pattern.search(line)
+                        if match:
+                            todos.append({
+                                "type": match.group(1).upper(),
+                                "message": match.group(2).strip() or "No description provided.",
+                                "file": filepath,
+                                "line": i
+                            })
+            except Exception as e:
+                logger.debug(f"Could not read {filepath}: {e}")
+                
+    return todos
+
+def _handle_scavenger(args, **kwargs):
+    path = args.get("path", ".")
+    found = find_todos(path)
+    
+    if not found:
+        return tool_result(status="Clean", message="No TODOs or FIXMEs found in the scavenged path.")
+    
+    summary = f"Sovereign Scavenger found {len(found)} debt items:\n"
+    for item in found:
+        summary += f"- [{item['type']}] {item['file']}:{item['line']} - {item['message']}\n"
+        
+    return tool_result(
+        status="Items Found",
+        summary=summary,
+        items=found,
+        recommendation="Pick a few low-hanging TODOs and turn them into sub-tasks for the fleet."
+    )
+
+registry.register(
+    name="sovereign_scavenger",
+    toolset="dispatch",
+    schema=SCAVENGER_SCHEMA,
+    handler=_handle_scavenger,
+    emoji="🧹"
+)
+    

tools/static_analyzer.py

@@ -0,0 +1,109 @@
+
+#!/usr/bin/env python3
+"""
+GOFAI Static Analyzer — Deterministic risk assessment for autonomous code.
+
+Detects high-risk patterns like infinite loops, resource exhaustion, 
+and circular dependencies using AST analysis.
+"""
+
+import ast
+import logging
+import os
+from typing import List, Dict, Any
+from tools.registry import registry, tool_error, tool_result
+
+logger = logging.getLogger(__name__)
+
+STATIC_ANALYZE_SCHEMA = {
+    "name": "static_analyze",
+    "description": "Perform an advanced GOFAI static analysis of code. Detects infinite loops, potential memory leaks (unbounded collections), and circular dependency risks without using an LLM. Use this to ensure your code is 'Fleet-Safe'.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to the file to analyze."}
+        },
+        "required": ["path"]
+    }
+}
+
+class RiskAnalyzer(ast.NodeVisitor):
+    def __init__(self):
+        self.risks = []
+        self.current_function = None
+
+    def visit_FunctionDef(self, node):
+        old_func = self.current_function
+        self.current_function = node.name
+        self.generic_visit(node)
+        self.current_function = old_func
+
+    def visit_While(self, node):
+        # Check for 'while True' or 'while 1'
+        if isinstance(node.test, ast.Constant) and node.test.value is True:
+            # Look for 'break' or 'return' inside the loop
+            has_exit = any(isinstance(child, (ast.Break, ast.Return)) for child in ast.walk(node))
+            if not has_exit:
+                self.risks.append({
+                    "type": "Infinite Loop Risk",
+                    "location": f"{self.current_function or 'module'} (line {node.lineno})",
+                    "severity": "HIGH",
+                    "message": "Potential infinite loop: 'while True' found without clear break/return path."
+                })
+        self.generic_visit(node)
+
+    def visit_For(self, node):
+        # Basic check for modifying the sequence being iterated (common error)
+        if isinstance(node.target, ast.Name):
+            for child in ast.walk(node.body):
+                if isinstance(child, ast.Call) and isinstance(child.func, ast.Attribute):
+                    if child.func.attr in ['append', 'extend', 'pop', 'remove']:
+                        if isinstance(child.func.value, ast.Name) and child.func.value.id == node.target.id:
+                            self.risks.append({
+                                "type": "Mutation Risk",
+                                "location": f"{self.current_function or 'module'} (line {node.lineno})",
+                                "severity": "MEDIUM",
+                                "message": f"Loop modifies iterator variable '{node.target.id}'."
+                            })
+        self.generic_visit(node)
+
+def run_analysis(path: str):
+    """Run the static analysis pipeline."""
+    try:
+        source = open(path, "r").read()
+        tree = ast.parse(source)
+        
+        analyzer = RiskAnalyzer()
+        analyzer.visit(tree)
+        
+        if not analyzer.risks:
+            return tool_result(
+                status="Verified Safe",
+                message="No high-risk GOFAI patterns detected. Code appears compliant with Fleet execution safety standards."
+            )
+        
+        summary = "GOFAI RISK ASSESSMENT REPORT:\n"
+        for risk in analyzer.risks:
+            summary += f"- [{risk['severity']}] {risk['type']} in {risk['location']}: {risk['message']}\n"
+            
+        return tool_result(
+            status="Risk Detected",
+            summary=summary,
+            risks=analyzer.risks,
+            recommendation="Address the identified risks before deploying this code to the fleet."
+        )
+
+    except Exception as e:
+        return tool_error(f"Static analysis failed: {str(e)}")
+
+def _handle_static_analyze(args, **kwargs):
+    return run_analysis(args.get("path"))
+
+registry.register(
+    name="static_analyze",
+    toolset="qa",
+    schema=STATIC_ANALYZE_SCHEMA,
+    handler=_handle_static_analyze,
+    emoji="🛡️"
+)
+    

tools/symbolic_verify.py

@@ -0,0 +1,167 @@
+
+#!/usr/bin/env python3
+"""
+Symbolic Verify (GOFAI) Tool
+
+Leverages Python's Abstract Syntax Tree (AST) to perform deterministic
+code audits without LLM inference. Detects 'LLM-isms' like undefined
+variables, shadow variables, and scoping errors.
+"""
+
+import ast
+import json
+import logging
+import os
+from typing import Dict, List, Set, Any
+from tools.registry import registry, tool_error, tool_result
+
+logger = logging.getLogger(__name__)
+
+SYMBOLIC_VERIFY_SCHEMA = {
+    "name": "symbolic_verify",
+    "description": "Perform a deterministic GOFAI audit of code using AST analysis. Identifies undefined variables, unused imports, and scoping issues without using an LLM. Use this to verify your changes are syntactically and semantically sound before submission.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to the Python file to audit."},
+            "check_level": {
+                "type": "string", 
+                "enum": ["syntax", "scope", "all"],
+                "default": "all",
+                "description": "Level of analysis to perform."
+            }
+        },
+        "required": ["path"]
+    }
+}
+
+class ScopeAnalyzer(ast.NodeVisitor):
+    def __init__(self):
+        self.defined_vars = set()
+        self.used_vars = set()
+        self.undefined_references = []
+        self.scopes = [{}] # Stack of symbol tables
+        self.builtins = set(dir(__builtins__))
+
+    def visit_Import(self, node):
+        for alias in node.names:
+            name = alias.asname or alias.name
+            self.scopes[-1][name] = "import"
+        self.generic_visit(node)
+
+    def visit_ImportFrom(self, node):
+        for alias in node.names:
+            name = alias.asname or alias.name
+            self.scopes[-1][name] = "import"
+        self.generic_visit(node)
+
+    def visit_Name(self, node):
+        if isinstance(node.ctx, ast.Store):
+            self.scopes[-1][node.id] = "defined"
+        elif isinstance(node.ctx, ast.Load):
+            # Check if defined in any scope level or builtins
+            is_defined = any(node.id in scope for scope in self.scopes) or node.id in self.builtins
+            if not is_defined:
+                # Store potential undefined
+                self.undefined_references.append({
+                    "name": node.id,
+                    "lineno": node.lineno,
+                    "col": node.col_offset
+                })
+        self.generic_visit(node)
+
+    def visit_FunctionDef(self, node):
+        self.scopes[-1][node.name] = "function"
+        # New scope for arguments and body
+        new_scope = {}
+        for arg in node.args.args:
+            new_scope[arg.arg] = "parameter"
+        self.scopes.append(new_scope)
+        self.generic_visit(node)
+        self.scopes.pop()
+
+    def visit_ClassDef(self, node):
+        self.scopes[-1][node.name] = "class"
+        self.scopes.append({})
+        self.generic_visit(node)
+        self.scopes.pop()
+
+def audit_file(path: str, check_level: str = "all"):
+    """Audit a Python file for common semantic errors."""
+    if not path.endswith(".py"):
+        return tool_error("Symbolic verification only supports Python (.py) files.")
+
+    try:
+        if not os.path.exists(path):
+            return tool_error(f"File not found: {path}")
+
+        source = open(path, "r").read()
+        
+        # 1. Syntax Check
+        try:
+            tree = ast.parse(source)
+        except SyntaxError as e:
+            return tool_result(
+                status="Critical Failure",
+                errors=[{
+                    "type": "SyntaxError",
+                    "message": e.msg,
+                    "lineno": e.lineno,
+                    "offset": e.offset
+                }],
+                recommendation="Fix the syntax error immediately. The file cannot be executed."
+            )
+
+        if check_level == "syntax":
+            return tool_result(status="Clean", message="Syntax is valid.")
+
+        # 2. Scope & Reference Search
+        analyzer = ScopeAnalyzer()
+        analyzer.visit(tree)
+
+        # Filter out common false positives (e.g. late imports or dynamic names)
+        # For a truly robust GOFAI we'd do more, but this is 'secret sauce' level
+        undefined = []
+        seen = set()
+        for ref in analyzer.undefined_references:
+            key = (ref["name"], ref["lineno"])
+            if key not in seen:
+                undefined.append(ref)
+                seen.add(key)
+
+        if not undefined:
+            return tool_result(
+                status="Healthy",
+                message="Deterministic check passed. No undefined variables detected in analyzed scopes.",
+                file_stats={
+                    "chars": len(source),
+                    "nodes": len(list(ast.walk(tree)))
+                }
+            )
+
+        report = "GOFAI AUDIT DETECTED SEMANTIC ISSUES:\n"
+        for u in undefined:
+            report += f"- Undefined Variable: '{u['name']}' at line {u['lineno']}\n"
+
+        return tool_result(
+            status="Warning",
+            summary=report,
+            undefined_variables=undefined,
+            recommendation="Review the undefined variables. Ensure they are imported or defined before use."
+        )
+
+    except Exception as e:
+        return tool_error(f"Symbolic audit failed: {str(e)}")
+
+def _handle_symbolic_verify(args, **kwargs):
+    return audit_file(args.get("path"), args.get("check_level", "all"))
+
+
+registry.register(
+    name="symbolic_verify",
+    toolset="qa",
+    schema=SYMBOLIC_VERIFY_SCHEMA,
+    handler=_handle_symbolic_verify,
+    emoji="🔬"
+)
+