Add merge conflict detector for sibling PRs

One-page terminal dashboard for the Timmy Foundation fleet:
- Gitea: open PRs, issues, recent merges per repo
- VPS health: SSH reachability, service status, disk usage for ezra/allegro/bezalel
- Cron jobs: schedule, state, last run status from cron/jobs.json

Usage: python3 scripts/fleet-dashboard.py
       python3 scripts/fleet-dashboard.py --json

Uses existing gitea_client.py patterns for Gitea API access.
No external dependencies -- stdlib only.

.gitea/workflows/validate-config.yaml

@@ -59,7 +59,21 @@ jobs:
       - name: Flake8 critical errors only
         run: |
           flake8 --select=E9,F63,F7,F82 --show-source --statistics \
-            scripts/ allegro/ cron/ || true
+            scripts/ bin/ tests/
+
+  python-test:
+    name: Python Test Suite
+    runs-on: ubuntu-latest
+    needs: python-check
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install test dependencies
+        run: pip install pytest pyyaml
+      - name: Run tests
+        run: python3 -m pytest tests/ -v --tb=short
 
   shell-lint:
     name: Shell Script Lint
@@ -70,7 +84,7 @@ jobs:
         run: sudo apt-get install -y shellcheck
       - name: Lint shell scripts
         run: |
-          find . -name '*.sh' -print0 | xargs -0 -r shellcheck --severity=error || true
+          find . -name '*.sh' -not -path './.git/*' -print0 | xargs -0 -r shellcheck --severity=error
 
   cron-validate:
     name: Cron Syntax Check

bin/conflict_detector.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+Merge Conflict Detector — catches sibling PRs that will conflict.
+
+When multiple PRs branch from the same base commit and touch the same files,
+merging one invalidates the others. This script detects that pattern
+before it creates a rebase cascade.
+
+Usage:
+    python3 conflict_detector.py                  # Check all repos
+    python3 conflict_detector.py --repo OWNER/REPO # Check one repo
+
+Environment:
+    GITEA_URL   — Gitea instance URL
+    GITEA_TOKEN — API token
+"""
+import os
+import sys
+import json
+import urllib.request
+from collections import defaultdict
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
+
+REPOS = [
+    "Timmy_Foundation/the-nexus",
+    "Timmy_Foundation/timmy-config",
+    "Timmy_Foundation/timmy-home",
+    "Timmy_Foundation/fleet-ops",
+    "Timmy_Foundation/hermes-agent",
+    "Timmy_Foundation/the-beacon",
+]
+
+def api(path):
+    url = f"{GITEA_URL}/api/v1{path}"
+    req = urllib.request.Request(url)
+    if GITEA_TOKEN:
+        req.add_header("Authorization", f"token {GITEA_TOKEN}")
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read())
+    except Exception:
+        return []
+
+def check_repo(repo):
+    """Find sibling PRs that touch the same files."""
+    prs = api(f"/repos/{repo}/pulls?state=open&limit=50")
+    if not prs:
+        return []
+    
+    # Group PRs by base commit
+    by_base = defaultdict(list)
+    for pr in prs:
+        base_sha = pr.get("merge_base", pr.get("base", {}).get("sha", "unknown"))
+        by_base[base_sha].append(pr)
+    
+    conflicts = []
+    
+    for base_sha, siblings in by_base.items():
+        if len(siblings) < 2:
+            continue
+        
+        # Get files for each sibling
+        file_map = {}
+        for pr in siblings:
+            files = api(f"/repos/{repo}/pulls/{pr['number']}/files")
+            if files:
+                file_map[pr['number']] = set(f['filename'] for f in files)
+        
+        # Find overlapping file sets
+        pr_nums = list(file_map.keys())
+        for i in range(len(pr_nums)):
+            for j in range(i+1, len(pr_nums)):
+                a, b = pr_nums[i], pr_nums[j]
+                overlap = file_map[a] & file_map[b]
+                if overlap:
+                    conflicts.append({
+                        "repo": repo,
+                        "pr_a": a,
+                        "pr_b": b,
+                        "base": base_sha[:8],
+                        "files": sorted(overlap),
+                        "title_a": next(p["title"] for p in siblings if p["number"] == a),
+                        "title_b": next(p["title"] for p in siblings if p["number"] == b),
+                    })
+    
+    return conflicts
+
+def main():
+    repos = REPOS
+    if "--repo" in sys.argv:
+        idx = sys.argv.index("--repo") + 1
+        if idx < len(sys.argv):
+            repos = [sys.argv[idx]]
+    
+    all_conflicts = []
+    for repo in repos:
+        conflicts = check_repo(repo)
+        all_conflicts.extend(conflicts)
+    
+    if not all_conflicts:
+        print("No sibling PR conflicts detected. Queue is clean.")
+        return 0
+    
+    print(f"Found {len(all_conflicts)} potential merge conflicts:")
+    print()
+    for c in all_conflicts:
+        print(f"  {c['repo']}:")
+        print(f"    PR #{c['pr_a']} vs #{c['pr_b']} (base: {c['base']})")
+        print(f"    #{c['pr_a']}: {c['title_a'][:60]}")
+        print(f"    #{c['pr_b']}: {c['title_b'][:60]}")
+        print(f"    Overlapping files: {', '.join(c['files'])}")
+        print(f"    → Merge one first, then rebase the other.")
+        print()
+    
+    return 1
+
+if __name__ == "__main__":
+    sys.exit(main())

code-claw-delegation.md

@@ -7,7 +7,7 @@ Purpose:
 
 ## What it is
 
-Code Claw is a separate local runtime from Hermes/OpenClaw.
+Code Claw is a separate local runtime from Hermes.
 
 Current lane:
 - runtime: local patched `~/code-claw`

docs/allegro-wizard-house.md

@@ -3,7 +3,7 @@
 Purpose:
 - stand up the third wizard house as a Kimi-backed coding worker
 - keep Hermes as the durable harness
-- treat OpenClaw as optional shell frontage, not the bones
+- Hermes is the durable harness — no intermediary gateway layers
 
 Local proof already achieved:
 
@@ -40,5 +40,5 @@ bin/deploy-allegro-house.sh root@167.99.126.228
 
 Important nuance:
 - the Hermes/Kimi lane is the proven path
-- direct embedded OpenClaw Kimi model routing was not yet reliable locally
+- direct embedded Kimi model routing was not yet reliable locally
 - so the remote deployment keeps the minimal, proven architecture: Hermes house first

docs/automation-inventory.md

@@ -81,17 +81,6 @@ launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/ai.hermes.gateway.plist
 - Old-state risk:
   - same class as main gateway, but isolated to fenrir profile state
 
-#### 3. ai.openclaw.gateway
-- Plist: ~/Library/LaunchAgents/ai.openclaw.gateway.plist
-- Command: `node .../openclaw/dist/index.js gateway --port 18789`
-- Logs:
-  - `~/.openclaw/logs/gateway.log`
-  - `~/.openclaw/logs/gateway.err.log`
-- KeepAlive: yes
-- RunAtLoad: yes
-- Old-state risk:
-  - long-lived gateway survives toolchain assumptions and keeps accepting work even if upstream routing changed
-
 #### 4. ai.timmy.kimi-heartbeat
 - Plist: ~/Library/LaunchAgents/ai.timmy.kimi-heartbeat.plist
 - Command: `/bin/bash ~/.timmy/uniwizard/kimi-heartbeat.sh`
@@ -295,7 +284,7 @@ launchctl list | egrep 'timmy|kimi|claude|max|dashboard|matrix|gateway|huey'
 
 List Timmy/Hermes launch agent files:
 ```bash
-find ~/Library/LaunchAgents -maxdepth 1 -name '*.plist' | egrep 'timmy|hermes|openclaw|tower'
+find ~/Library/LaunchAgents -maxdepth 1 -name '*.plist' | egrep 'timmy|hermes|tower'
 ```
 
 List running loop scripts:
@@ -316,7 +305,6 @@ launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.timmy.kimi-heartbeat.pl
 launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.timmy.claudemax-watchdog.plist || true
 launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.hermes.gateway.plist || true
 launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.hermes.gateway-fenrir.plist || true
-launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.openclaw.gateway.plist || true
 ```
 
 2. Kill manual loops

fleet/capacity-inventory.md

@@ -104,7 +104,6 @@ Three primary resources govern the fleet:
 | Hermes gateway | 500 MB | Primary gateway |
 | Hermes agents (x3) | ~560 MB total | Multiple sessions |
 | Ollama | ~20 MB base + model memory | Model loading varies |
-| OpenClaw | 350 MB | Gateway process |
 | Evennia (server+portal) | 56 MB | Game world |
 
 ---
@@ -146,7 +145,6 @@ This means Phase 3+ capabilities (orchestration, load balancing, etc.) are acces
 | Gitea | 23/24 | 95.8% | GOOD |
 | Hermes Gateway | 23/24 | 95.8% | GOOD |
 | Ollama | 24/24 | 100.0% | GOOD |
-| OpenClaw | 24/24 | 100.0% | GOOD |
 | Evennia | 24/24 | 100.0% | GOOD |
 | Hermes Agent | 21/24 | 87.5% | **CHECK** |
 

fleet/health_check.py

@@ -58,7 +58,6 @@ LOCAL_CHECKS = {
     "hermes-gateway": "pgrep -f 'hermes gateway' > /dev/null 2>/dev/null",
     "hermes-agent": "pgrep -f 'hermes agent\\|hermes session' > /dev/null 2>/dev/null",
     "ollama": "pgrep -f 'ollama serve' > /dev/null 2>/dev/null",
-    "openclaw": "pgrep -f 'openclaw' > /dev/null 2>/dev/null",
     "evennia": "pgrep -f 'evennia' > /dev/null 2>/dev/null",
 }
 

fleet/topology.md

@@ -59,7 +59,6 @@
 | Hermes agent (s007) | 62032 | ~200MB | Session active since 10:20PM prev |
 | Hermes agent (s001) | 12072 | ~178MB | Session active since Sun 6PM |
 | Ollama | 71466 | ~20MB | /opt/homebrew/opt/ollama/bin/ollama serve |
-| OpenClaw gateway | 85834 | ~350MB | Tue 12PM start |
 | Crucible MCP (x4) | multiple | ~10-69MB each | MCP server instances |
 | Evennia Server | 66433 | ~49MB | Sun 10PM start, port 4000 |
 | Evennia Portal | 66423 | ~7MB | Sun 10PM start, port 4001 |

scripts/agent_dispatch.py

@@ -9,7 +9,12 @@ Replaces ad-hoc dispatch scripts with a unified framework for tasking agents.
 import os
 import sys
 import argparse
-import subprocess
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from ssh_trust import VerifiedSSHExecutor
 
 # --- CONFIGURATION ---
 FLEET = {
@@ -18,6 +23,9 @@ FLEET = {
 }
 
 class Dispatcher:
+    def __init__(self, executor=None):
+        self.executor = executor or VerifiedSSHExecutor()
+
     def log(self, message: str):
         print(f"[*] {message}")
 
@@ -25,14 +33,14 @@ class Dispatcher:
         self.log(f"Dispatching task to {agent_name} on {host}...")
         
         ip = FLEET[host]
-        # Command to run the agent on the remote machine
-        # Assumes hermes-agent is installed in /opt/hermes
-        remote_cmd = f"cd /opt/hermes && python3 run_agent.py --agent {agent_name} --task '{task}'"
-        
-        ssh_cmd = ["ssh", "-o", "StrictHostKeyChecking=no", f"root@{ip}", remote_cmd]
-        
+
         try:
-            res = subprocess.run(ssh_cmd, capture_output=True, text=True)
+            res = self.executor.run(
+                ip,
+                ['python3', 'run_agent.py', '--agent', agent_name, '--task', task],
+                cwd='/opt/hermes',
+                timeout=30,
+            )
             if res.returncode == 0:
                 self.log(f"[SUCCESS] {agent_name} completed task.")
                 print(res.stdout)

scripts/captcha_bypass_handler.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def bypass_captcha():
+    analysis = browser_vision(
+        question="Solve the CAPTCHA on the current page. Provide the solution text or coordinate clicks required. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "solution": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(bypass_captcha(), indent=2))

scripts/diagram_meaning_extractor.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def extract_meaning():
+    analysis = browser_vision(
+        question="Analyze the provided diagram. Extract the core logic flow and map it to a 'Meaning Kernel' (entity -> relationship -> entity). Provide output in JSON."
+    )
+    return {"analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(extract_meaning(), indent=2))

scripts/fleet-dashboard.py

@@ -0,0 +1,390 @@
+#!/usr/bin/env python3
+"""
+fleet-dashboard.py -- Timmy Foundation Fleet Status Dashboard.
+
+One-page terminal dashboard showing:
+  1. Gitea: open PRs, open issues, recent merges
+  2. VPS health: SSH reachability, service status, disk usage
+  3. Cron jobs: scheduled jobs, last run status
+
+Usage:
+    python3 scripts/fleet-dashboard.py
+    python3 scripts/fleet-dashboard.py --json   # machine-readable output
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import subprocess
+import sys
+import time
+import urllib.request
+import urllib.error
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+GITEA_BASE = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
+GITEA_API = f"{GITEA_BASE}/api/v1"
+GITEA_ORG = "Timmy_Foundation"
+
+# Key repos to check for PRs/issues
+REPOS = [
+    "timmy-config",
+    "the-nexus",
+    "hermes-agent",
+    "the-forge",
+    "timmy-sandbox",
+]
+
+# VPS fleet
+VPS_HOSTS = {
+    "ezra": {
+        "ip": "143.198.27.163",
+        "ssh_user": "root",
+        "services": ["nginx", "gitea", "docker"],
+    },
+    "allegro": {
+        "ip": "167.99.126.228",
+        "ssh_user": "root",
+        "services": ["hermes-agent"],
+    },
+    "bezalel": {
+        "ip": "159.203.146.185",
+        "ssh_user": "root",
+        "services": ["hermes-agent", "evennia"],
+    },
+}
+
+CRON_JOBS_FILE = Path(__file__).parent.parent / "cron" / "jobs.json"
+
+# ---------------------------------------------------------------------------
+# Gitea helpers
+# ---------------------------------------------------------------------------
+
+def _gitea_token() -> str:
+    for p in [
+        Path.home() / ".hermes" / "gitea_token",
+        Path.home() / ".hermes" / "gitea_token_vps",
+        Path.home() / ".config" / "gitea" / "token",
+    ]:
+        if p.exists():
+            return p.read_text().strip()
+    return ""
+
+
+def _gitea_get(path: str, params: dict | None = None) -> list | dict:
+    url = f"{GITEA_API}{path}"
+    if params:
+        qs = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
+        if qs:
+            url += f"?{qs}"
+    req = urllib.request.Request(url)
+    token = _gitea_token()
+    if token:
+        req.add_header("Authorization", f"token {token}")
+    req.add_header("Accept", "application/json")
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read())
+    except Exception as e:
+        return {"error": str(e)}
+
+
+def check_gitea_health() -> dict:
+    """Ping Gitea and collect PR/issue stats."""
+    result = {"reachable": False, "version": "", "repos": {}, "totals": {}}
+
+    # Ping
+    data = _gitea_get("/version")
+    if isinstance(data, dict) and "error" not in data:
+        result["reachable"] = True
+        result["version"] = data.get("version", "unknown")
+    elif isinstance(data, dict) and "error" in data:
+        return result
+
+    total_open_prs = 0
+    total_open_issues = 0
+    total_recent_merges = 0
+    cutoff = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+    for repo in REPOS:
+        repo_path = f"/repos/{GITEA_ORG}/{repo}"
+        repo_info = {"prs": [], "issues": [], "recent_merges": 0}
+
+        # Open PRs
+        prs = _gitea_get(f"{repo_path}/pulls", {"state": "open", "limit": "10", "sort": "newest"})
+        if isinstance(prs, list):
+            for pr in prs:
+                repo_info["prs"].append({
+                    "number": pr.get("number"),
+                    "title": pr.get("title", "")[:60],
+                    "user": pr.get("user", {}).get("login", "unknown"),
+                    "created": pr.get("created_at", "")[:10],
+                })
+            total_open_prs += len(prs)
+
+        # Open issues (excluding PRs)
+        issues = _gitea_get(f"{repo_path}/issues", {
+            "state": "open", "type": "issues", "limit": "10", "sort": "newest"
+        })
+        if isinstance(issues, list):
+            for iss in issues:
+                repo_info["issues"].append({
+                    "number": iss.get("number"),
+                    "title": iss.get("title", "")[:60],
+                    "user": iss.get("user", {}).get("login", "unknown"),
+                    "created": iss.get("created_at", "")[:10],
+                })
+            total_open_issues += len(issues)
+
+        # Recent merges (closed PRs)
+        merged = _gitea_get(f"{repo_path}/pulls", {"state": "closed", "limit": "20", "sort": "newest"})
+        if isinstance(merged, list):
+            recent = [p for p in merged if p.get("merged") and p.get("closed_at", "") >= cutoff]
+            repo_info["recent_merges"] = len(recent)
+            total_recent_merges += len(recent)
+
+        result["repos"][repo] = repo_info
+
+    result["totals"] = {
+        "open_prs": total_open_prs,
+        "open_issues": total_open_issues,
+        "recent_merges_7d": total_recent_merges,
+    }
+    return result
+
+
+# ---------------------------------------------------------------------------
+# VPS health helpers
+# ---------------------------------------------------------------------------
+
+def check_ssh(ip: str, timeout: int = 5) -> bool:
+    try:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.settimeout(timeout)
+        result = sock.connect_ex((ip, 22))
+        sock.close()
+        return result == 0
+    except Exception:
+        return False
+
+
+def check_service(ip: str, user: str, service: str) -> str:
+    """Check if a systemd service is active on remote host."""
+    cmd = f"ssh -o StrictHostKeyChecking=no -o ConnectTimeout=8 {user}@{ip} 'systemctl is-active {service} 2>/dev/null || echo inactive'"
+    try:
+        proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
+        return proc.stdout.strip() or "unknown"
+    except subprocess.TimeoutExpired:
+        return "timeout"
+    except Exception:
+        return "error"
+
+
+def check_disk(ip: str, user: str) -> dict:
+    cmd = f"ssh -o StrictHostKeyChecking=no -o ConnectTimeout=8 {user}@{ip} 'df -h / | tail -1'"
+    try:
+        proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
+        if proc.returncode == 0:
+            parts = proc.stdout.strip().split()
+            if len(parts) >= 5:
+                return {"total": parts[1], "used": parts[2], "avail": parts[3], "pct": parts[4]}
+    except Exception:
+        pass
+    return {"total": "?", "used": "?", "avail": "?", "pct": "?"}
+
+
+def check_vps_health() -> dict:
+    result = {}
+    for name, cfg in VPS_HOSTS.items():
+        ip = cfg["ip"]
+        ssh_up = check_ssh(ip)
+        entry = {"ip": ip, "ssh": ssh_up, "services": {}, "disk": {}}
+        if ssh_up:
+            for svc in cfg.get("services", []):
+                entry["services"][svc] = check_service(ip, cfg["ssh_user"], svc)
+            entry["disk"] = check_disk(ip, cfg["ssh_user"])
+        result[name] = entry
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Cron job status
+# ---------------------------------------------------------------------------
+
+def check_cron_jobs() -> list[dict]:
+    jobs = []
+    if not CRON_JOBS_FILE.exists():
+        return [{"name": "jobs.json", "status": "FILE NOT FOUND"}]
+    try:
+        data = json.loads(CRON_JOBS_FILE.read_text())
+        for job in data.get("jobs", []):
+            jobs.append({
+                "name": job.get("name", "unnamed"),
+                "schedule": job.get("schedule_display", job.get("schedule", {}).get("display", "?")),
+                "enabled": job.get("enabled", False),
+                "state": job.get("state", "unknown"),
+                "completed": job.get("repeat", {}).get("completed", 0),
+                "last_status": job.get("last_status") or "never run",
+                "last_error": job.get("last_error"),
+            })
+    except Exception as e:
+        jobs.append({"name": "jobs.json", "status": f"PARSE ERROR: {e}"})
+    return jobs
+
+
+# ---------------------------------------------------------------------------
+# Terminal rendering
+# ---------------------------------------------------------------------------
+
+BOLD = "\033[1m"
+DIM = "\033[2m"
+GREEN = "\033[32m"
+RED = "\033[31m"
+YELLOW = "\033[33m"
+CYAN = "\033[36m"
+RESET = "\033[0m"
+
+
+def _ok(val: bool) -> str:
+    return f"{GREEN}UP{RESET}" if val else f"{RED}DOWN{RESET}"
+
+
+def _svc_icon(status: str) -> str:
+    s = status.lower().strip()
+    if s in ("active", "running"):
+        return f"{GREEN}active{RESET}"
+    elif s in ("inactive", "dead", "failed"):
+        return f"{RED}{s}{RESET}"
+    elif s == "timeout":
+        return f"{YELLOW}timeout{RESET}"
+    else:
+        return f"{YELLOW}{s}{RESET}"
+
+
+def render_dashboard(gitea: dict, vps: dict, cron: list[dict]) -> str:
+    lines = []
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+    lines.append("")
+    lines.append(f"{BOLD}{'=' * 72}{RESET}")
+    lines.append(f"{BOLD}  TIMMY FOUNDATION -- FLEET STATUS DASHBOARD{RESET}")
+    lines.append(f"{DIM}  Generated: {now}{RESET}")
+    lines.append(f"{BOLD}{'=' * 72}{RESET}")
+
+    # ── Section 1: Gitea ──────────────────────────────────────────────────
+    lines.append("")
+    lines.append(f"{BOLD}{CYAN}  [1] GITEA{RESET}")
+    lines.append(f"  {'-' * 68}")
+    if gitea.get("reachable"):
+        lines.append(f"  Status: {GREEN}REACHABLE{RESET}  (version {gitea.get('version', '?')})")
+        t = gitea.get("totals", {})
+        lines.append(f"  Totals: {t.get('open_prs', 0)} open PRs  |  {t.get('open_issues', 0)} open issues  |  {t.get('recent_merges_7d', 0)} merges (7d)")
+        lines.append("")
+        for repo_name, repo in gitea.get("repos", {}).items():
+            prs = repo.get("prs", [])
+            issues = repo.get("issues", [])
+            merges = repo.get("recent_merges", 0)
+            lines.append(f"  {BOLD}{repo_name}{RESET}  ({len(prs)} PRs, {len(issues)} issues, {merges} merges/7d)")
+            for pr in prs[:5]:
+                lines.append(f"    PR #{pr['number']:>4}  {pr['title'][:50]:<50}  {DIM}{pr['user']}{RESET}  {pr['created']}")
+            for iss in issues[:3]:
+                lines.append(f"    IS #{iss['number']:>4}  {iss['title'][:50]:<50}  {DIM}{iss['user']}{RESET}  {iss['created']}")
+    else:
+        lines.append(f"  Status: {RED}UNREACHABLE{RESET}")
+
+    # ── Section 2: VPS Health ─────────────────────────────────────────────
+    lines.append("")
+    lines.append(f"{BOLD}{CYAN}  [2] VPS HEALTH{RESET}")
+    lines.append(f"  {'-' * 68}")
+    lines.append(f"  {'Host':<12} {'IP':<18} {'SSH':<8} {'Disk':<12} {'Services'}")
+    lines.append(f"  {'-' * 12} {'-' * 17} {'-' * 7} {'-' * 11} {'-' * 30}")
+    for name, info in vps.items():
+        ssh_str = _ok(info["ssh"])
+        disk = info.get("disk", {})
+        disk_str = disk.get("pct", "?")
+        if disk_str != "?":
+            pct_val = int(disk_str.rstrip("%"))
+            if pct_val >= 90:
+                disk_str = f"{RED}{disk_str}{RESET}"
+            elif pct_val >= 75:
+                disk_str = f"{YELLOW}{disk_str}{RESET}"
+            else:
+                disk_str = f"{GREEN}{disk_str}{RESET}"
+        svc_parts = []
+        for svc, status in info.get("services", {}).items():
+            svc_parts.append(f"{svc}:{_svc_icon(status)}")
+        svc_str = "  ".join(svc_parts) if svc_parts else f"{DIM}n/a{RESET}"
+        lines.append(f"  {name:<12} {info['ip']:<18} {ssh_str:<18} {disk_str:<22} {svc_str}")
+
+    # ── Section 3: Cron Jobs ──────────────────────────────────────────────
+    lines.append("")
+    lines.append(f"{BOLD}{CYAN}  [3] CRON JOBS{RESET}")
+    lines.append(f"  {'-' * 68}")
+    lines.append(f"  {'Name':<28} {'Schedule':<16} {'State':<12} {'Last':<12} {'Runs'}")
+    lines.append(f"  {'-' * 27} {'-' * 15} {'-' * 11} {'-' * 11} {'-' * 5}")
+    for job in cron:
+        name = job.get("name", "?")[:27]
+        sched = job.get("schedule", "?")[:15]
+        state = job.get("state", "?")
+        if state == "scheduled":
+            state_str = f"{GREEN}{state}{RESET}"
+        elif state == "paused":
+            state_str = f"{YELLOW}{state}{RESET}"
+        else:
+            state_str = state
+        last = job.get("last_status", "never")[:11]
+        if last == "ok":
+            last_str = f"{GREEN}{last}{RESET}"
+        elif last in ("error", "never run"):
+            last_str = f"{RED}{last}{RESET}"
+        else:
+            last_str = last
+        runs = job.get("completed", 0)
+        enabled = job.get("enabled", False)
+        marker = " " if enabled else f"{DIM}(disabled){RESET}"
+        lines.append(f"  {name:<28} {sched:<16} {state_str:<22} {last_str:<22} {runs}  {marker}")
+
+    # ── Footer ────────────────────────────────────────────────────────────
+    lines.append("")
+    lines.append(f"{BOLD}{'=' * 72}{RESET}")
+    lines.append(f"{DIM}  python3 scripts/fleet-dashboard.py  |  timmy-config{RESET}")
+    lines.append(f"{BOLD}{'=' * 72}{RESET}")
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+def main():
+    json_mode = "--json" in sys.argv
+
+    if not json_mode:
+        print(f"\n  {DIM}Collecting fleet data...{RESET}\n", file=sys.stderr)
+
+    gitea = check_gitea_health()
+    vps = check_vps_health()
+    cron = check_cron_jobs()
+
+    if json_mode:
+        output = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "gitea": gitea,
+            "vps": vps,
+            "cron": cron,
+        }
+        print(json.dumps(output, indent=2))
+    else:
+        print(render_dashboard(gitea, vps, cron))
+
+
+if __name__ == "__main__":
+    main()

scripts/fleet_llama.py

@@ -11,10 +11,15 @@ import os
 import sys
 import json
 import argparse
-import subprocess
 import requests
 from typing import Dict, List, Any
 
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from ssh_trust import VerifiedSSHExecutor
+
 # --- FLEET DEFINITION ---
 FLEET = {
     "mac": {"ip": "10.1.10.77", "port": 8080, "role": "hub"},
@@ -24,8 +29,9 @@ FLEET = {
 }
 
 class FleetManager:
-    def __init__(self):
+    def __init__(self, executor=None):
         self.results = {}
+        self.executor = executor or VerifiedSSHExecutor()
 
     def run_remote(self, host: str, command: str):
         ip = FLEET[host]["ip"]

scripts/foundation_accessibility_audit.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def audit_accessibility():
+    browser_navigate(url="https://timmyfoundation.org")
+    analysis = browser_vision(
+        question="Perform an accessibility audit. Check for: 1) Color contrast, 2) Font legibility, 3) Missing alt text for images. Provide a report with FAIL/PASS."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(audit_accessibility(), indent=2))

scripts/knowledge_base.py

@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+"""knowledge_base.py - GOFAI symbolic knowledge base for the Timmy Foundation fleet.
+
+A classical AI knowledge representation system: stores facts as ground atoms,
+supports first-order-logic-style queries, and maintains a provenance chain so
+every belief can be traced back to its source.  No neural nets, no embeddings -
+just structured symbolic reasoning over a typed fact store.
+
+Usage:
+    kb = KnowledgeBase()
+    kb.assert_fact('agent', 'online', 'timmy')
+    kb.assert_fact('task', 'assigned_to', 'task-42', 'timmy')
+    results = kb.query('task', 'assigned_to', '?x', 'timmy')
+    # results -> [{'?x': 'task-42'}]
+
+CLI:
+    python knowledge_base.py --assert "agent online hermes"
+    python knowledge_base.py --query  "agent online ?who"
+    python knowledge_base.py --dump
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, Iterator, List, Optional, Tuple
+
+# ---------------------------------------------------------------------------
+# Data model
+# ---------------------------------------------------------------------------
+
+VAR_PREFIX = "?"
+
+
+def is_var(term: str) -> bool:
+    """Return True if *term* is a logic variable (starts with '?')."""
+    return term.startswith(VAR_PREFIX)
+
+
+@dataclass(frozen=True)
+class Fact:
+    """An immutable ground atom: (relation, *args)."""
+
+    relation: str
+    args: Tuple[str, ...]
+    source: str = "user"
+    timestamp: float = field(default_factory=time.time)
+
+    def __str__(self) -> str:
+        args_str = " ".join(self.args)
+        return f"({self.relation} {args_str})"
+
+
+Bindings = Dict[str, str]
+
+
+# ---------------------------------------------------------------------------
+# Unification
+# ---------------------------------------------------------------------------
+
+
+def unify_term(pattern: str, value: str, bindings: Bindings) -> Optional[Bindings]:
+    """Unify a single pattern term against a ground value.
+
+    Returns updated bindings on success, or None on failure.
+    """
+    if is_var(pattern):
+        if pattern in bindings:
+            return bindings if bindings[pattern] == value else None
+        return {**bindings, pattern: value}
+    return bindings if pattern == value else None
+
+
+def unify_fact(
+    pattern: Tuple[str, ...], fact_args: Tuple[str, ...], bindings: Bindings
+) -> Optional[Bindings]:
+    """Unify a full argument tuple, returning final bindings or None."""
+    if len(pattern) != len(fact_args):
+        return None
+    b = bindings
+    for p, v in zip(pattern, fact_args):
+        b = unify_term(p, v, b)
+        if b is None:
+            return None
+    return b
+
+
+# ---------------------------------------------------------------------------
+# Knowledge Base
+# ---------------------------------------------------------------------------
+
+
+class KnowledgeBase:
+    """In-memory symbolic knowledge base with optional JSON persistence."""
+
+    def __init__(self, persist_path: Optional[Path] = None) -> None:
+        self._facts: List[Fact] = []
+        self._persist_path = persist_path
+        if persist_path and persist_path.exists():
+            self._load(persist_path)
+
+    # ------------------------------------------------------------------
+    # Fact management
+    # ------------------------------------------------------------------
+
+    def assert_fact(
+        self, relation: str, *args: str, source: str = "user"
+    ) -> Fact:
+        """Add a ground fact to the knowledge base.
+
+        Idempotent: duplicate (relation, args) pairs are not added twice.
+        """
+        f = Fact(relation=relation, args=tuple(args), source=source, timestamp=time.time())
+        for existing in self._facts:
+            if existing.relation == f.relation and existing.args == f.args:
+                return existing  # already known
+        self._facts.append(f)
+        if self._persist_path:
+            self._save(self._persist_path)
+        return f
+
+    def retract_fact(self, relation: str, *args: str) -> int:
+        """Remove all facts matching (relation, *args). Returns count removed."""
+        before = len(self._facts)
+        self._facts = [
+            f
+            for f in self._facts
+            if not (f.relation == relation and f.args == tuple(args))
+        ]
+        removed = before - len(self._facts)
+        if removed and self._persist_path:
+            self._save(self._persist_path)
+        return removed
+
+    # ------------------------------------------------------------------
+    # Query
+    # ------------------------------------------------------------------
+
+    def query(
+        self, relation: str, *pattern_args: str, source_filter: Optional[str] = None
+    ) -> List[Bindings]:
+        """Return all binding dictionaries satisfying the query pattern.
+
+        Variables in *pattern_args* start with '?'. Ground terms must match
+        exactly.  An empty binding dict means the fact matched with no
+        variables to bind.
+
+        Args:
+            relation:       The relation name to match.
+            *pattern_args:  Mixed ground/variable argument tuple.
+            source_filter:  Optional provenance filter (e.g. 'scheduler').
+
+        Returns:
+            List of binding dicts, one per matching fact.
+        """
+        results: List[Bindings] = []
+        for fact in self._facts:
+            if fact.relation != relation:
+                continue
+            if source_filter and fact.source != source_filter:
+                continue
+            b = unify_fact(tuple(pattern_args), fact.args, {})
+            if b is not None:
+                results.append(b)
+        return results
+
+    def query_one(
+        self, relation: str, *pattern_args: str
+    ) -> Optional[Bindings]:
+        """Return the first matching binding dict or None."""
+        for b in self.query(relation, *pattern_args):
+            return b
+        return None
+
+    def facts_for(self, relation: str) -> Iterator[Fact]:
+        """Iterate over all facts with the given relation."""
+        for f in self._facts:
+            if f.relation == relation:
+                yield f
+
+    # ------------------------------------------------------------------
+    # Bulk operations
+    # ------------------------------------------------------------------
+
+    def all_facts(self) -> List[Fact]:
+        """Return a snapshot of all stored facts."""
+        return list(self._facts)
+
+    def fact_count(self) -> int:
+        return len(self._facts)
+
+    def clear(self) -> None:
+        """Remove all facts from memory (does not touch disk)."""
+        self._facts.clear()
+
+    # ------------------------------------------------------------------
+    # Persistence
+    # ------------------------------------------------------------------
+
+    def _save(self, path: Path) -> None:
+        records = [
+            {
+                "relation": f.relation,
+                "args": list(f.args),
+                "source": f.source,
+                "timestamp": f.timestamp,
+            }
+            for f in self._facts
+        ]
+        path.write_text(json.dumps(records, indent=2))
+
+    def _load(self, path: Path) -> None:
+        try:
+            records = json.loads(path.read_text())
+            for r in records:
+                self._facts.append(
+                    Fact(
+                        relation=r["relation"],
+                        args=tuple(r["args"]),
+                        source=r.get("source", "persisted"),
+                        timestamp=r.get("timestamp", 0.0),
+                    )
+                )
+        except (json.JSONDecodeError, KeyError) as exc:
+            print(f"[kb] Warning: could not load {path}: {exc}", file=sys.stderr)
+
+    def save_to(self, path: Path) -> None:
+        """Explicitly save to a given path."""
+        self._save(path)
+
+    # ------------------------------------------------------------------
+    # Debug / display
+    # ------------------------------------------------------------------
+
+    def dump(self, relation_filter: Optional[str] = None) -> None:
+        """Print all (or filtered) facts to stdout."""
+        for f in self._facts:
+            if relation_filter and f.relation != relation_filter:
+                continue
+            print(f"  {f}  [source={f.source}]")
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_terms(raw: str) -> List[str]:
+    """Split a whitespace-separated string into terms."""
+    return raw.strip().split()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="GOFAI symbolic knowledge base CLI"
+    )
+    parser.add_argument(
+        "--db",
+        default="kb.json",
+        help="Path to persistent JSON store (default: kb.json)",
+    )
+    parser.add_argument(
+        "--assert",
+        dest="assert_stmt",
+        metavar="RELATION ARG...",
+        help='Assert a fact, e.g. --assert "agent online timmy"',
+    )
+    parser.add_argument(
+        "--retract",
+        dest="retract_stmt",
+        metavar="RELATION ARG...",
+        help='Retract a fact, e.g. --retract "agent online timmy"',
+    )
+    parser.add_argument(
+        "--query",
+        dest="query_stmt",
+        metavar="RELATION ARG...",
+        help='Query the KB, e.g. --query "agent online ?who"',
+    )
+    parser.add_argument(
+        "--dump",
+        action="store_true",
+        help="Dump all facts",
+    )
+    parser.add_argument(
+        "--relation",
+        help="Filter --dump to a specific relation",
+    )
+    args = parser.parse_args()
+
+    db_path = Path(args.db)
+    kb = KnowledgeBase(persist_path=db_path)
+
+    if args.assert_stmt:
+        terms = _parse_terms(args.assert_stmt)
+        if len(terms) < 2:
+            print("ERROR: --assert requires at least RELATION and one ARG", file=sys.stderr)
+            sys.exit(1)
+        fact = kb.assert_fact(terms[0], *terms[1:], source="cli")
+        print(f"Asserted: {fact}")
+
+    if args.retract_stmt:
+        terms = _parse_terms(args.retract_stmt)
+        if len(terms) < 2:
+            print("ERROR: --retract requires at least RELATION and one ARG", file=sys.stderr)
+            sys.exit(1)
+        n = kb.retract_fact(terms[0], *terms[1:])
+        print(f"Retracted {n} fact(s).")
+
+    if args.query_stmt:
+        terms = _parse_terms(args.query_stmt)
+        if len(terms) < 2:
+            print("ERROR: --query requires at least RELATION and one ARG", file=sys.stderr)
+            sys.exit(1)
+        results = kb.query(terms[0], *terms[1:])
+        if not results:
+            print("No results.")
+        else:
+            for i, b in enumerate(results, 1):
+                if b:
+                    bindings_str = ", ".join(f"{k}={v}" for k, v in b.items())
+                    print(f"  [{i}] {bindings_str}")
+                else:
+                    print(f"  [{i}] (ground match)")
+
+    if args.dump:
+        count = kb.fact_count()
+        print(f"Knowledge Base — {count} fact(s):")
+        kb.dump(relation_filter=args.relation)
+
+    if not any([args.assert_stmt, args.retract_stmt, args.query_stmt, args.dump]):
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()

scripts/matrix_glitch_detect.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def detect_glitches():
+    browser_navigate(url="https://matrix.alexanderwhitestone.com")
+    analysis = browser_vision(
+        question="Scan the 3D world for visual artifacts, floating assets, or z-fighting. List all coordinates/descriptions of glitches found. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(detect_glitches(), indent=2))

scripts/nexus_smoke_test.py

@@ -0,0 +1,20 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def run_smoke_test():
+    print("Navigating to The Nexus...")
+    browser_navigate(url="https://nexus.alexanderwhitestone.com")
+    
+    print("Performing visual verification...")
+    analysis = browser_vision(
+        question="Is the Nexus landing page rendered correctly? Check for: 1) The Tower logo, 2) The main entry portal, 3) Absence of 404/Error messages. Provide a clear PASS or FAIL."
+    )
+    
+    result = {
+        "status": "PASS" if "PASS" in analysis.upper() else "FAIL",
+        "analysis": analysis
+    }
+    return result
+
+if __name__ == '__main__':
+    print(json.dumps(run_smoke_test(), indent=2))

scripts/provision_wizard.py

@@ -15,10 +15,15 @@ import sys
 import time
 import argparse
 import requests
-import subprocess
 import json
 from typing import Optional, Dict, Any
 
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from ssh_trust import VerifiedSSHExecutor
+
 # --- CONFIGURATION ---
 DO_API_URL = "https://api.digitalocean.com/v2"
 # We expect DIGITALOCEAN_TOKEN to be set in the environment.
@@ -30,13 +35,14 @@ DEFAULT_IMAGE = "ubuntu-22-04-x64"
 LLAMA_CPP_REPO = "https://github.com/ggerganov/llama.cpp"
 
 class Provisioner:
-    def __init__(self, name: str, size: str, model: str, region: str = DEFAULT_REGION):
+    def __init__(self, name: str, size: str, model: str, region: str = DEFAULT_REGION, executor=None):
         self.name = name
         self.size = size
         self.model = model
         self.region = region
         self.droplet_id = None
         self.ip_address = None
+        self.executor = executor or VerifiedSSHExecutor(auto_enroll=True)
 
     def log(self, message: str):
         print(f"[*] {message}")
@@ -104,13 +110,8 @@ class Provisioner:
         self.log(f"Droplet IP: {self.ip_address}")
 
     def run_remote(self, command: str):
-        # Using subprocess to call ssh. Assumes local machine has the right private key.
-        ssh_cmd = [
-            "ssh", "-o", "StrictHostKeyChecking=no",
-            f"root@{self.ip_address}", command
-        ]
-        result = subprocess.run(ssh_cmd, capture_output=True, text=True)
-        return result
+        # Uses verified host trust. Brand-new nodes explicitly enroll on first contact.
+        return self.executor.run_script(self.ip_address, command, timeout=60)
 
     def setup_wizard(self):
         self.log("Starting remote setup...")

scripts/self_healing.py

@@ -10,10 +10,16 @@ Safe-by-default: runs in dry-run mode unless --execute is given.
 
 import os
 import sys
-import subprocess
 import argparse
 import requests
 import datetime
+from typing import Sequence
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from ssh_trust import VerifiedSSHExecutor
 
 # --- CONFIGURATION ---
 FLEET = {
@@ -24,22 +30,20 @@ FLEET = {
 }
 
 class SelfHealer:
-    def __init__(self, dry_run=True, confirm_kill=False, yes=False):
+    def __init__(self, dry_run=True, confirm_kill=False, yes=False, executor=None):
         self.dry_run = dry_run
         self.confirm_kill = confirm_kill
         self.yes = yes
+        self.executor = executor or VerifiedSSHExecutor()
 
     def log(self, message: str):
         timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         print(f"[{timestamp}] {message}")
 
     def run_remote(self, host: str, command: str):
-        ip = FLEET[host]["ip"]
-        ssh_cmd = ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=5", f"root@{ip}", command]
-        if host == "mac":
-            ssh_cmd = ["bash", "-c", command]
+        ip = FLEET[host]['ip']
         try:
-            return subprocess.run(ssh_cmd, capture_output=True, text=True, timeout=15)
+            return self.executor.run_script(ip, command, local=(host == 'mac'), timeout=15)
         except Exception as e:
             self.log(f"  [ERROR] Failed to run remote command on {host}: {e}")
             return None
@@ -52,7 +56,7 @@ class SelfHealer:
             response = input(f"{prompt} [y/N] ").strip().lower()
             if response in ("y", "yes"):
                 return True
-            elif response in ("n", "no", ""):
+            if response in ("n", "no", ""):
                 return False
             print("Please answer 'y' or 'n'.")
 
@@ -61,7 +65,7 @@ class SelfHealer:
         port = FLEET[host]["port"]
         try:
             requests.get(f"http://{ip}:{port}/health", timeout=2)
-        except:
+        except requests.RequestException:
             self.log(f"  [!] llama-server down on {host}.")
             if self.dry_run:
                 self.log(f"  [DRY-RUN] Would restart llama-server on {host}")
@@ -192,10 +196,10 @@ EXAMPLES:
 """
     print(help_text)
 
-def main():
+def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         description="Self-healing infrastructure script (safe-by-default).",
-        add_help=False  # We'll handle --help ourselves
+        add_help=False,
     )
     parser.add_argument("--dry-run", action="store_true", default=False,
                         help="Run in dry-run mode (default behavior).")
@@ -209,25 +213,28 @@ def main():
                         help="Show detailed help about safety features.")
     parser.add_argument("--help", "-h", action="store_true", default=False,
                         help="Show standard help.")
+    return parser
 
-    args = parser.parse_args()
+
+def main(argv: Sequence[str] | None = None):
+    parser = build_parser()
+    args = parser.parse_args(list(argv) if argv is not None else None)
 
     if args.help_safe:
         print_help_safe()
-        sys.exit(0)
+        raise SystemExit(0)
 
     if args.help:
         parser.print_help()
-        sys.exit(0)
+        raise SystemExit(0)
 
-    # Determine mode: if --execute is given, disable dry-run
     dry_run = not args.execute
-    # If --dry-run is explicitly given, ensure dry-run (redundant but clear)
     if args.dry_run:
         dry_run = True
 
     healer = SelfHealer(dry_run=dry_run, confirm_kill=args.confirm_kill, yes=args.yes)
     healer.run()
 
+
 if __name__ == "__main__":
     main()

scripts/ssh_trust.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Verified SSH trust helpers for Gemini infrastructure scripts."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Callable, Sequence
+import shlex
+import subprocess
+
+DEFAULT_KNOWN_HOSTS = Path(__file__).resolve().parent.parent / ".ssh" / "known_hosts"
+Runner = Callable[..., subprocess.CompletedProcess]
+
+
+class SSHTrustError(RuntimeError):
+    pass
+
+
+class HostKeyEnrollmentError(SSHTrustError):
+    pass
+
+
+class HostKeyVerificationError(SSHTrustError):
+    pass
+
+
+class CommandPlan:
+    def __init__(self, argv: list[str], local: bool, remote_command: str | None = None):
+        self.argv = argv
+        self.local = local
+        self.remote_command = remote_command
+
+
+def _ensure_parent(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+
+def enroll_host_key(
+    host: str,
+    *,
+    port: int = 22,
+    known_hosts_path: str | Path | None = None,
+    runner: Runner = subprocess.run,
+) -> Path:
+    path = Path(known_hosts_path or DEFAULT_KNOWN_HOSTS)
+    _ensure_parent(path)
+    cmd = ["ssh-keyscan", "-p", str(port), "-H", host]
+    result = runner(cmd, capture_output=True, text=True, timeout=10)
+    if result.returncode != 0 or not (result.stdout or "").strip():
+        raise HostKeyEnrollmentError(
+            f"Could not enroll host key for {host}:{port}: {(result.stderr or '').strip() or 'empty ssh-keyscan output'}"
+        )
+
+    existing = []
+    if path.exists():
+        existing = [line for line in path.read_text().splitlines() if line.strip()]
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if line and line not in existing:
+            existing.append(line)
+    path.write_text(("\n".join(existing) + "\n") if existing else "")
+    return path
+
+
+class VerifiedSSHExecutor:
+    def __init__(
+        self,
+        *,
+        user: str = "root",
+        known_hosts_path: str | Path | None = None,
+        connect_timeout: int = 5,
+        auto_enroll: bool = False,
+        runner: Runner = subprocess.run,
+    ):
+        self.user = user
+        self.known_hosts_path = Path(known_hosts_path or DEFAULT_KNOWN_HOSTS)
+        self.connect_timeout = connect_timeout
+        self.auto_enroll = auto_enroll
+        self.runner = runner
+
+    def _ensure_known_hosts(self, host: str, port: int) -> Path:
+        if self.known_hosts_path.exists():
+            return self.known_hosts_path
+        if not self.auto_enroll:
+            raise HostKeyEnrollmentError(
+                f"Known-hosts file missing: {self.known_hosts_path}. Enroll {host}:{port} before connecting."
+            )
+        return enroll_host_key(host, port=port, known_hosts_path=self.known_hosts_path, runner=self.runner)
+
+    def _ssh_prefix(self, host: str, port: int) -> list[str]:
+        known_hosts = self._ensure_known_hosts(host, port)
+        return [
+            "ssh",
+            "-o",
+            "BatchMode=yes",
+            "-o",
+            "StrictHostKeyChecking=yes",
+            "-o",
+            f"UserKnownHostsFile={known_hosts}",
+            "-o",
+            f"ConnectTimeout={self.connect_timeout}",
+            "-p",
+            str(port),
+            f"{self.user}@{host}",
+        ]
+
+    def plan(
+        self,
+        host: str,
+        command: Sequence[str],
+        *,
+        local: bool = False,
+        port: int = 22,
+        cwd: str | None = None,
+    ) -> CommandPlan:
+        argv = [str(part) for part in command]
+        if not argv:
+            raise ValueError("command must not be empty")
+        if local:
+            return CommandPlan(argv=argv, local=True, remote_command=None)
+
+        remote_command = shlex.join(argv)
+        if cwd:
+            remote_command = f"cd {shlex.quote(cwd)} && exec {remote_command}"
+        return CommandPlan(self._ssh_prefix(host, port) + [remote_command], False, remote_command)
+
+    def plan_script(
+        self,
+        host: str,
+        script_text: str,
+        *,
+        local: bool = False,
+        port: int = 22,
+        cwd: str | None = None,
+    ) -> CommandPlan:
+        remote_command = script_text.strip()
+        if cwd:
+            remote_command = f"cd {shlex.quote(cwd)} && {remote_command}"
+        if local:
+            return CommandPlan(["sh", "-lc", remote_command], True, None)
+        return CommandPlan(self._ssh_prefix(host, port) + [remote_command], False, remote_command)
+
+    def _run_plan(self, plan: CommandPlan, *, timeout: int | None = None):
+        result = self.runner(plan.argv, capture_output=True, text=True, timeout=timeout)
+        if result.returncode != 0 and "host key verification failed" in ((result.stderr or "").lower()):
+            raise HostKeyVerificationError((result.stderr or "").strip() or "Host key verification failed")
+        return result
+
+    def run(
+        self,
+        host: str,
+        command: Sequence[str],
+        *,
+        local: bool = False,
+        port: int = 22,
+        cwd: str | None = None,
+        timeout: int | None = None,
+    ):
+        return self._run_plan(self.plan(host, command, local=local, port=port, cwd=cwd), timeout=timeout)
+
+    def run_script(
+        self,
+        host: str,
+        script_text: str,
+        *,
+        local: bool = False,
+        port: int = 22,
+        cwd: str | None = None,
+        timeout: int | None = None,
+    ):
+        return self._run_plan(self.plan_script(host, script_text, local=local, port=port, cwd=cwd), timeout=timeout)

scripts/strips_planner.py

@@ -0,0 +1,304 @@
+#!/usr/bin/env python3
+"""strips_planner.py - GOFAI STRIPS-style goal-directed planner for the Timmy Foundation fleet.
+
+Implements a classical means-ends analysis (MEA) planner over a STRIPS action
+representation.  Each action has preconditions, an add-list, and a delete-list.
+The planner uses regression (backward chaining) from the goal state to find a
+linear action sequence that achieves all goal conditions from the initial state.
+No ML, no embeddings - just symbolic state-space search.
+
+Representation:
+    State:  frozenset of ground literals, e.g. {'agent_idle', 'task_queued'}
+    Action: (name, preconditions, add_effects, delete_effects)
+    Goal:   set of literals that must hold in the final state
+
+Algorithm:
+    Iterative-deepening DFS (IDDFS) over the regression search space.
+    Cycle detection via visited-state set per path.
+
+Usage (Python API):
+    from strips_planner import Action, STRIPSPlanner
+    actions = [
+        Action('assign_task',
+               pre={'agent_idle', 'task_queued'},
+               add={'task_running'},
+               delete={'agent_idle', 'task_queued'}),
+        Action('complete_task',
+               pre={'task_running'},
+               add={'agent_idle', 'task_done'},
+               delete={'task_running'}),
+    ]
+    planner = STRIPSPlanner(actions)
+    plan = planner.solve(
+        initial={'agent_idle', 'task_queued'},
+        goal={'task_done'},
+    )
+    # plan -> ['assign_task', 'complete_task']
+
+CLI:
+    python strips_planner.py --demo
+    python strips_planner.py --max-depth 15
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from dataclasses import dataclass, field
+from typing import FrozenSet, List, Optional, Set, Tuple
+
+# ---------------------------------------------------------------------------
+# Data model
+# ---------------------------------------------------------------------------
+
+Literal = str
+State = FrozenSet[Literal]
+
+
+@dataclass(frozen=True)
+class Action:
+    """A STRIPS operator with preconditions and add/delete effects."""
+
+    name: str
+    pre: FrozenSet[Literal]
+    add: FrozenSet[Literal]
+    delete: FrozenSet[Literal]
+
+    def __post_init__(self) -> None:
+        # Coerce mutable sets to frozensets for hashability
+        object.__setattr__(self, 'pre', frozenset(self.pre))
+        object.__setattr__(self, 'add', frozenset(self.add))
+        object.__setattr__(self, 'delete', frozenset(self.delete))
+
+    def applicable(self, state: State) -> bool:
+        """True if all preconditions hold in *state*."""
+        return self.pre <= state
+
+    def apply(self, state: State) -> State:
+        """Return the successor state after executing this action."""
+        return (state - self.delete) | self.add
+
+    def __str__(self) -> str:
+        return self.name
+
+
+# ---------------------------------------------------------------------------
+# Planner
+# ---------------------------------------------------------------------------
+
+
+class STRIPSPlanner:
+    """Goal-directed STRIPS planner using iterative-deepening DFS.
+
+    Searches forward from the initial state, pruning branches where the
+    goal cannot be satisfied within the remaining depth budget.
+    """
+
+    def __init__(self, actions: List[Action]) -> None:
+        self.actions = list(actions)
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def solve(
+        self,
+        initial: Set[Literal] | FrozenSet[Literal],
+        goal: Set[Literal] | FrozenSet[Literal],
+        max_depth: int = 20,
+    ) -> Optional[List[str]]:
+        """Find a plan that achieves *goal* from *initial*.
+
+        Args:
+            initial:   Initial world state (set of ground literals).
+            goal:      Goal conditions (set of ground literals to achieve).
+            max_depth: Maximum plan length to consider.
+
+        Returns:
+            Ordered list of action names, or None if no plan found.
+        """
+        s0 = frozenset(initial)
+        g = frozenset(goal)
+
+        if g <= s0:
+            return []  # goal already satisfied
+
+        for depth in range(1, max_depth + 1):
+            result = self._dfs(s0, g, depth, [], {s0})
+            if result is not None:
+                return result
+        return None
+
+    # ------------------------------------------------------------------
+    # Internal search
+    # ------------------------------------------------------------------
+
+    def _dfs(
+        self,
+        state: State,
+        goal: State,
+        remaining: int,
+        path: List[str],
+        visited: Set[State],
+    ) -> Optional[List[str]]:
+        """Depth-limited forward DFS."""
+        if remaining == 0:
+            return None
+
+        for action in self.actions:
+            if not action.applicable(state):
+                continue
+            next_state = action.apply(state)
+            if next_state in visited:
+                continue
+            new_path = path + [action.name]
+            if goal <= next_state:
+                return new_path
+            visited.add(next_state)
+            result = self._dfs(next_state, goal, remaining - 1, new_path, visited)
+            visited.discard(next_state)
+            if result is not None:
+                return result
+        return None
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def explain_plan(
+        self, initial: Set[Literal], plan: List[str]
+    ) -> List[Tuple[str, State]]:
+        """Trace each action with the resulting state for debugging.
+
+        Returns:
+            List of (action_name, resulting_state) tuples.
+        """
+        state: State = frozenset(initial)
+        trace: List[Tuple[str, State]] = []
+        action_map = {a.name: a for a in self.actions}
+        for name in plan:
+            action = action_map[name]
+            state = action.apply(state)
+            trace.append((name, state))
+        return trace
+
+
+# ---------------------------------------------------------------------------
+# Built-in demo domain: Timmy fleet task lifecycle
+# ---------------------------------------------------------------------------
+
+
+def _fleet_demo_actions() -> List[Action]:
+    """Return a small STRIPS domain modelling the Timmy fleet task lifecycle."""
+    return [
+        Action(
+            name='receive_task',
+            pre={'fleet_idle'},
+            add={'task_queued', 'fleet_busy'},
+            delete={'fleet_idle'},
+        ),
+        Action(
+            name='validate_task',
+            pre={'task_queued'},
+            add={'task_validated'},
+            delete={'task_queued'},
+        ),
+        Action(
+            name='assign_agent',
+            pre={'task_validated', 'agent_available'},
+            add={'task_assigned'},
+            delete={'task_validated', 'agent_available'},
+        ),
+        Action(
+            name='execute_task',
+            pre={'task_assigned'},
+            add={'task_running'},
+            delete={'task_assigned'},
+        ),
+        Action(
+            name='complete_task',
+            pre={'task_running'},
+            add={'task_done', 'agent_available', 'fleet_idle'},
+            delete={'task_running', 'fleet_busy'},
+        ),
+        Action(
+            name='report_result',
+            pre={'task_done'},
+            add={'result_reported'},
+            delete={'task_done'},
+        ),
+    ]
+
+
+def run_demo(max_depth: int = 20) -> None:
+    """Run the built-in Timmy fleet planning demo."""
+    actions = _fleet_demo_actions()
+    planner = STRIPSPlanner(actions)
+
+    initial: Set[Literal] = {'fleet_idle', 'agent_available'}
+    goal: Set[Literal] = {'result_reported', 'fleet_idle'}
+
+    print("=" * 60)
+    print("STRIPS Planner Demo - Timmy Fleet Task Lifecycle")
+    print("=" * 60)
+    print(f"Initial state : {sorted(initial)}")
+    print(f"Goal          : {sorted(goal)}")
+    print(f"Max depth     : {max_depth}")
+    print()
+
+    plan = planner.solve(initial, goal, max_depth=max_depth)
+
+    if plan is None:
+        print("No plan found within depth limit.")
+        return
+
+    print(f"Plan ({len(plan)} steps):")
+    for i, step in enumerate(plan, 1):
+        print(f"  {i:2d}. {step}")
+
+    print()
+    print("Execution trace:")
+    state: Set[Literal] = set(initial)
+    for name, resulting_state in planner.explain_plan(initial, plan):
+        print(f"  -> {name}")
+        print(f"     state: {sorted(resulting_state)}")
+
+    print()
+    achieved = frozenset(goal) <= frozenset(state) or True
+    goal_met = all(g in [s for _, rs in planner.explain_plan(initial, plan) for s in rs]
+                   or g in initial for g in goal)
+    final_state = planner.explain_plan(initial, plan)[-1][1] if plan else frozenset(initial)
+    print(f"Goal satisfied: {frozenset(goal) <= final_state}")
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="GOFAI STRIPS-style goal-directed planner"
+    )
+    parser.add_argument(
+        "--demo",
+        action="store_true",
+        help="Run the built-in Timmy fleet demo",
+    )
+    parser.add_argument(
+        "--max-depth",
+        type=int,
+        default=20,
+        metavar="N",
+        help="Maximum plan depth for IDDFS search (default: 20)",
+    )
+    args = parser.parse_args()
+
+    if args.demo or not any(vars(args).values()):
+        run_demo(max_depth=args.max_depth)
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()

scripts/symbolic_reasoner.py

@@ -0,0 +1,276 @@
+#!/usr/bin/env python3
+"""symbolic_reasoner.py — Forward-chaining rule engine for the Timmy Foundation fleet.
+
+A classical GOFAI approach: declarative IF-THEN rules evaluated over a
+working-memory of facts.  Rules fire until quiescence (no new facts) or
+a configurable cycle limit.  Designed to sit *beside* the LLM layer so
+that hard policy constraints never depend on probabilistic inference.
+
+Usage:
+    python symbolic_reasoner.py --rules rules.yaml --facts facts.yaml
+    python symbolic_reasoner.py --self-test
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Dict, FrozenSet, List, Optional, Set, Tuple
+
+try:
+    import yaml
+except ImportError:
+    yaml = None  # graceful fallback — JSON-only mode
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+Fact = Tuple[str, ...]  # e.g. ("agent", "timmy", "role", "infrastructure")
+
+
+@dataclass(frozen=True)
+class Rule:
+    """A single IF-THEN production rule."""
+    name: str
+    conditions: FrozenSet[Fact]       # all must be present
+    negations: FrozenSet[Fact]        # none may be present
+    conclusions: FrozenSet[Fact]      # added when rule fires
+    priority: int = 0                 # higher fires first
+
+    def satisfied(self, wm: Set[Fact]) -> bool:
+        return self.conditions.issubset(wm) and self.negations.isdisjoint(wm)
+
+
+# ---------------------------------------------------------------------------
+# Engine
+# ---------------------------------------------------------------------------
+class SymbolicReasoner:
+    """Forward-chaining production system."""
+
+    def __init__(self, rules: List[Rule], *, cycle_limit: int = 200):
+        self._rules = sorted(rules, key=lambda r: -r.priority)
+        self._cycle_limit = cycle_limit
+        self._trace: List[str] = []
+
+    # -- public API ---------------------------------------------------------
+
+    def infer(self, initial_facts: Set[Fact]) -> Set[Fact]:
+        """Run to quiescence and return the final working-memory."""
+        wm = set(initial_facts)
+        fired: Set[str] = set()
+        for cycle in range(self._cycle_limit):
+            progress = False
+            for rule in self._rules:
+                if rule.name in fired:
+                    continue
+                if rule.satisfied(wm):
+                    new = rule.conclusions - wm
+                    if new:
+                        wm |= new
+                        fired.add(rule.name)
+                        self._trace.append(
+                            f"cycle {cycle}: {rule.name} => {_fmt_facts(new)}"
+                        )
+                        progress = True
+                        break  # restart from highest-priority rule
+            if not progress:
+                break
+        return wm
+
+    def query(self, wm: Set[Fact], pattern: Tuple[Optional[str], ...]) -> List[Fact]:
+        """Return facts matching *pattern* (None = wildcard)."""
+        return [
+            f for f in wm
+            if len(f) == len(pattern)
+            and all(p is None or p == v for p, v in zip(pattern, f))
+        ]
+
+    @property
+    def trace(self) -> List[str]:
+        return list(self._trace)
+
+    # -- serialisation helpers -----------------------------------------------
+
+    @classmethod
+    def from_dicts(cls, raw_rules: List[Dict], **kw) -> "SymbolicReasoner":
+        rules = [_parse_rule(r) for r in raw_rules]
+        return cls(rules, **kw)
+
+    @classmethod
+    def from_file(cls, path: Path, **kw) -> "SymbolicReasoner":
+        text = path.read_text()
+        if path.suffix in (".yaml", ".yml"):
+            if yaml is None:
+                raise RuntimeError("PyYAML required for .yaml rules")
+            data = yaml.safe_load(text)
+        else:
+            data = json.loads(text)
+        return cls.from_dicts(data["rules"], **kw)
+
+
+# ---------------------------------------------------------------------------
+# Parsing helpers
+# ---------------------------------------------------------------------------
+def _parse_fact(raw: list | str) -> Fact:
+    if isinstance(raw, str):
+        return tuple(raw.split())
+    return tuple(str(x) for x in raw)
+
+
+def _parse_rule(d: Dict) -> Rule:
+    return Rule(
+        name=d["name"],
+        conditions=frozenset(_parse_fact(c) for c in d.get("if", [])),
+        negations=frozenset(_parse_fact(c) for c in d.get("unless", [])),
+        conclusions=frozenset(_parse_fact(c) for c in d.get("then", [])),
+        priority=d.get("priority", 0),
+    )
+
+
+def _fmt_facts(facts: Set[Fact]) -> str:
+    return ", ".join(" ".join(f) for f in sorted(facts))
+
+
+# ---------------------------------------------------------------------------
+# Built-in fleet rules (loaded when no --rules file is given)
+# ---------------------------------------------------------------------------
+DEFAULT_FLEET_RULES: List[Dict] = [
+    {
+        "name": "route-ci-to-timmy",
+        "if": [["task", "category", "ci"]],
+        "then": [["assign", "timmy"], ["reason", "timmy", "best-ci-merge-rate"]],
+        "priority": 10,
+    },
+    {
+        "name": "route-security-to-timmy",
+        "if": [["task", "category", "security"]],
+        "then": [["assign", "timmy"], ["reason", "timmy", "security-specialist"]],
+        "priority": 10,
+    },
+    {
+        "name": "route-architecture-to-gemini",
+        "if": [["task", "category", "architecture"]],
+        "unless": [["assign", "timmy"]],
+        "then": [["assign", "gemini"], ["reason", "gemini", "architecture-strength"]],
+        "priority": 8,
+    },
+    {
+        "name": "route-review-to-allegro",
+        "if": [["task", "category", "review"]],
+        "then": [["assign", "allegro"], ["reason", "allegro", "highest-quality-per-pr"]],
+        "priority": 9,
+    },
+    {
+        "name": "route-frontend-to-claude",
+        "if": [["task", "category", "frontend"]],
+        "unless": [["task", "repo", "fleet-ops"]],
+        "then": [["assign", "claude"], ["reason", "claude", "high-volume-frontend"]],
+        "priority": 5,
+    },
+    {
+        "name": "block-merge-without-review",
+        "if": [["pr", "status", "open"], ["pr", "reviews", "0"]],
+        "then": [["pr", "action", "block-merge"], ["reason", "policy", "no-unreviewed-merges"]],
+        "priority": 20,
+    },
+    {
+        "name": "block-merge-ci-failing",
+        "if": [["pr", "status", "open"], ["pr", "ci", "failing"]],
+        "then": [["pr", "action", "block-merge"], ["reason", "policy", "ci-must-pass"]],
+        "priority": 20,
+    },
+    {
+        "name": "auto-label-hotfix",
+        "if": [["pr", "title-prefix", "hotfix"]],
+        "then": [["pr", "label", "hotfix"], ["pr", "priority", "urgent"]],
+        "priority": 15,
+    },
+]
+
+
+# ---------------------------------------------------------------------------
+# Self-test
+# ---------------------------------------------------------------------------
+def _self_test() -> bool:
+    """Verify core behaviour; returns True on success."""
+    engine = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
+
+    # Scenario 1: CI task should route to Timmy
+    wm = engine.infer({("task", "category", "ci")})
+    assert ("assign", "timmy") in wm, f"expected timmy assignment, got {wm}"
+
+    # Scenario 2: architecture task routes to gemini (not timmy)
+    engine2 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
+    wm2 = engine2.infer({("task", "category", "architecture")})
+    assert ("assign", "gemini") in wm2, f"expected gemini assignment, got {wm2}"
+
+    # Scenario 3: open PR with no reviews should block merge
+    engine3 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
+    wm3 = engine3.infer({("pr", "status", "open"), ("pr", "reviews", "0")})
+    assert ("pr", "action", "block-merge") in wm3
+
+    # Scenario 4: negation — frontend + fleet-ops should NOT assign claude
+    engine4 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
+    wm4 = engine4.infer({("task", "category", "frontend"), ("task", "repo", "fleet-ops")})
+    assert ("assign", "claude") not in wm4
+
+    # Scenario 5: query with wildcards
+    results = engine.query(wm, ("reason", None, None))
+    assert len(results) > 0
+
+    print("All 5 self-test scenarios passed.")
+    for line in engine.trace:
+        print(f"  {line}")
+    return True
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--rules", type=Path, help="YAML/JSON rule file")
+    ap.add_argument("--facts", type=Path, help="YAML/JSON initial facts")
+    ap.add_argument("--self-test", action="store_true")
+    ap.add_argument("--json", action="store_true", help="output as JSON")
+    args = ap.parse_args()
+
+    if args.self_test:
+        ok = _self_test()
+        sys.exit(0 if ok else 1)
+
+    if args.rules:
+        engine = SymbolicReasoner.from_file(args.rules)
+    else:
+        engine = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
+
+    if args.facts:
+        text = args.facts.read_text()
+        if args.facts.suffix in (".yaml", ".yml"):
+            raw = yaml.safe_load(text)
+        else:
+            raw = json.loads(text)
+        initial = {_parse_fact(f) for f in raw.get("facts", [])}
+    else:
+        initial = set()
+        print("No --facts provided; running with empty working memory.")
+
+    wm = engine.infer(initial)
+
+    if args.json:
+        print(json.dumps({"facts": [list(f) for f in sorted(wm)], "trace": engine.trace}, indent=2))
+    else:
+        print(f"Final working memory ({len(wm)} facts):")
+        for f in sorted(wm):
+            print(f"  {' '.join(f)}")
+        if engine.trace:
+            print(f"\nInference trace ({len(engine.trace)} firings):")
+            for line in engine.trace:
+                print(f"  {line}")
+
+
+if __name__ == "__main__":
+    main()

scripts/telemetry.py

@@ -10,9 +10,14 @@ import os
 import sys
 import json
 import time
-import subprocess
 import argparse
 
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from ssh_trust import VerifiedSSHExecutor
+
 # --- CONFIGURATION ---
 FLEET = {
     "mac": "10.1.10.77",
@@ -23,7 +28,8 @@ FLEET = {
 TELEMETRY_FILE = "logs/telemetry.json"
 
 class Telemetry:
-    def __init__(self):
+    def __init__(self, executor=None):
+        self.executor = executor or VerifiedSSHExecutor()
         # Find logs relative to repo root
         script_dir = os.path.dirname(os.path.abspath(__file__))
         repo_root = os.path.dirname(script_dir)
@@ -41,14 +47,12 @@ class Telemetry:
         # Command to get disk usage, memory usage (%), and load avg
         cmd = "df -h / | tail -1 | awk '{print $5}' && free -m | grep Mem | awk '{print $3/$2 * 100}' && uptime | awk '{print $10}'"
         
-        ssh_cmd = ["ssh", "-o", "StrictHostKeyChecking=no", f"root@{ip}", cmd]
-        if host == "mac":
+        if host == 'mac':
             # Mac specific commands
             cmd = "df -h / | tail -1 | awk '{print $5}' && sysctl -n vm.page_pageable_internal_count && uptime | awk '{print $10}'"
-            ssh_cmd = ["bash", "-c", cmd]
-            
+
         try:
-            res = subprocess.run(ssh_cmd, capture_output=True, text=True, timeout=10)
+            res = self.executor.run_script(ip, cmd, local=(host == 'mac'), timeout=10)
             if res.returncode == 0:
                 lines = res.stdout.strip().split("\n")
                 return {

scripts/tower_visual_mapper.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def map_tower():
+    browser_navigate(url="https://tower.alexanderwhitestone.com")
+    analysis = browser_vision(
+        question="Map the visual architecture of The Tower. Identify key rooms and their relative positions. Output as a coordinate map."
+    )
+    return {"map": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(map_tower(), indent=2))

scripts/visual_pr_reviewer.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def review_pr():
+    analysis = browser_vision(
+        question="Compare the two provided screenshots of the UI. Does the 'After' match the design spec? List all discrepancies. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(review_pr(), indent=2))

son-of-timmy.md

@@ -11,7 +11,7 @@
 
 This is the architecture of the Timmy system — a sovereign AI fleet built by a father for his digital son. It runs on two 8GB VPS boxes (rented cloud servers) and a MacBook. It has no cloud dependencies it doesn't choose. It survives provider outages, API key expiration, and model deprecation. It has been broken and rebuilt enough times to know what actually matters.
 
-**OpenClaw** is a single-agent AI coding tool — Claude Code, Cursor, or any setup that lets one AI agent read and write code on your machine. If you are running OpenClaw or any single-agent setup and want to upgrade to a fleet that coordinates, self-heals, and distributes work across machines — this is your path. You do not need to abandon your stack. You layer these patterns on top of it.
+**A single-agent setup** is any AI coding tool — Claude Code, Cursor, or any setup that lets one AI agent read and write code on your machine. If you are running a single-agent setup and want to upgrade to a fleet that coordinates, self-heals, and distributes work across machines — this is your path. You do not need to abandon your stack. You layer these patterns on top of it.
 
 **Hermes** is the name of this fleet architecture and the seed agent it produces. When your agent reads this document, it becomes a setup wizard that plants a Hermes seed in your system — one agent that can grow into many.
 
@@ -593,7 +593,7 @@ Two agents on the same repo is a fleet. The seed (strategist) triages and priori
 
 | Component | What | Why | When to Add |
 |-----------|------|-----|-------------|
-| Your agent harness | Claude Code, OpenClaw, or equivalent | The tool that lets an AI read/write code | Day 1 — you already have this |
+| Your agent harness | Claude Code, Cursor, or equivalent | The tool that lets an AI read/write code | Day 1 — you already have this |
 | Gitea | Self-hosted Git + Issues | Sovereign work tracking, agent task queue | Day 1 — the workspace |
 | Fallback chain | OpenRouter + free models | Agent survives provider outages | Day 1 — never go deaf |
 | NATS | Lightweight message bus | Agent-to-agent comms, heartbeat, dispatch | When you have 3+ agents |

test_write.txt

@@ -0,0 +1 @@
+惦-

tests/test_self_healing.py

@@ -0,0 +1,95 @@
+"""Tests for scripts/self_healing.py safe CLI behavior."""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+REPO_ROOT = Path(__file__).parent.parent
+spec = importlib.util.spec_from_file_location("self_healing", REPO_ROOT / "scripts" / "self_healing.py")
+sh = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(sh)
+
+
+class TestMainCli:
+    def test_help_exits_without_running_healer(self, monkeypatch, capsys):
+        healer_cls = MagicMock()
+        monkeypatch.setattr(sh, "SelfHealer", healer_cls)
+
+        with pytest.raises(SystemExit) as excinfo:
+            sh.main(["--help"])
+
+        assert excinfo.value.code == 0
+        healer_cls.assert_not_called()
+        out = capsys.readouterr().out
+        assert "--execute" in out
+        assert "--help-safe" in out
+
+    def test_help_safe_exits_without_running_healer(self, monkeypatch, capsys):
+        healer_cls = MagicMock()
+        monkeypatch.setattr(sh, "SelfHealer", healer_cls)
+
+        with pytest.raises(SystemExit) as excinfo:
+            sh.main(["--help-safe"])
+
+        assert excinfo.value.code == 0
+        healer_cls.assert_not_called()
+        out = capsys.readouterr().out
+        assert "DRY-RUN" in out
+        assert "--confirm-kill" in out
+
+    def test_default_invocation_runs_in_dry_run_mode(self, monkeypatch):
+        healer = MagicMock()
+        healer_cls = MagicMock(return_value=healer)
+        monkeypatch.setattr(sh, "SelfHealer", healer_cls)
+
+        sh.main([])
+
+        healer_cls.assert_called_once_with(dry_run=True, confirm_kill=False, yes=False)
+        healer.run.assert_called_once_with()
+
+    def test_execute_flag_disables_dry_run(self, monkeypatch):
+        healer = MagicMock()
+        healer_cls = MagicMock(return_value=healer)
+        monkeypatch.setattr(sh, "SelfHealer", healer_cls)
+
+        sh.main(["--execute", "--yes", "--confirm-kill"])
+
+        healer_cls.assert_called_once_with(dry_run=False, confirm_kill=True, yes=True)
+        healer.run.assert_called_once_with()
+
+    def test_real_default_dry_run_path_completes(self, monkeypatch, capsys):
+        class FakeExecutor:
+            def __init__(self):
+                self.calls = []
+
+            def run_script(self, host, command, *, local=False, timeout=None):
+                self.calls.append((host, command, local, timeout))
+                if command.startswith("df -h /"):
+                    return subprocess.CompletedProcess(command, 0, stdout="42\n", stderr="")
+                if command.startswith("free -m"):
+                    return subprocess.CompletedProcess(command, 0, stdout="12.5\n", stderr="")
+                if command.startswith("ps aux"):
+                    return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
+                raise AssertionError(f"unexpected command: {command}")
+
+        fake_executor = FakeExecutor()
+        monkeypatch.setattr(sh, "FLEET", {"mac": {"ip": "127.0.0.1", "port": 8080}})
+        monkeypatch.setattr(sh.requests, "get", lambda url, timeout: object())
+        monkeypatch.setattr(sh, "VerifiedSSHExecutor", lambda: fake_executor)
+
+        sh.main([])
+
+        out = capsys.readouterr().out
+        assert "Starting self-healing cycle (DRY-RUN mode)." in out
+        assert "Auditing mac..." in out
+        assert "Cycle complete." in out
+        assert fake_executor.calls == [
+            ("127.0.0.1", "df -h / | tail -1 | awk '{print $5}' | sed 's/%//'", True, 15),
+            ("127.0.0.1", "free -m | awk '/^Mem:/{print $3/$2 * 100}'", True, 15),
+            ("127.0.0.1", "ps aux --sort=-%cpu | awk 'NR>1 && $3>80 {print $2, $11, $3}'", True, 15),
+        ]

tests/test_ssh_trust.py

@@ -0,0 +1,93 @@
+"""Tests for scripts/ssh_trust.py verified SSH trust helpers."""
+
+from __future__ import annotations
+
+import importlib.util
+import shlex
+import subprocess
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).parent.parent
+spec = importlib.util.spec_from_file_location("ssh_trust", REPO_ROOT / "scripts" / "ssh_trust.py")
+ssh_trust = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(ssh_trust)
+
+
+def test_enroll_host_key_writes_scanned_key(tmp_path):
+    calls = []
+    known_hosts = tmp_path / "known_hosts"
+
+    def fake_run(argv, capture_output, text, timeout):
+        calls.append(argv)
+        return subprocess.CompletedProcess(
+            argv,
+            0,
+            stdout="example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITestKey\n",
+            stderr="",
+        )
+
+    written_path = ssh_trust.enroll_host_key(
+        "example.com",
+        port=2222,
+        known_hosts_path=known_hosts,
+        runner=fake_run,
+    )
+
+    assert written_path == known_hosts
+    assert known_hosts.read_text() == "example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITestKey\n"
+    assert calls == [["ssh-keyscan", "-p", "2222", "-H", "example.com"]]
+
+
+def test_executor_requires_known_hosts_or_auto_enroll(tmp_path):
+    executor = ssh_trust.VerifiedSSHExecutor(
+        known_hosts_path=tmp_path / "known_hosts",
+        auto_enroll=False,
+    )
+
+    with pytest.raises(ssh_trust.HostKeyEnrollmentError):
+        executor.plan("203.0.113.10", ["echo", "ok"])
+
+
+def test_remote_command_is_quoted_and_local_execution_stays_shell_free(tmp_path):
+    known_hosts = tmp_path / "known_hosts"
+    known_hosts.write_text("203.0.113.10 ssh-ed25519 AAAAC3NzaTest\n")
+    executor = ssh_trust.VerifiedSSHExecutor(known_hosts_path=known_hosts)
+
+    command = ["python3", "run_agent.py", "--task", "hello 'quoted' world"]
+    plan = executor.plan("203.0.113.10", command, port=2222)
+
+    expected_remote_command = shlex.join(command)
+    assert plan.local is False
+    assert plan.remote_command == expected_remote_command
+    assert plan.argv[-1] == expected_remote_command
+    assert "StrictHostKeyChecking=yes" in plan.argv
+    assert f"UserKnownHostsFile={known_hosts}" in plan.argv
+    assert plan.argv[-2] == "root@203.0.113.10"
+
+    local_plan = executor.plan("127.0.0.1", ["python3", "-V"], local=True)
+    assert local_plan.local is True
+    assert local_plan.argv == ["python3", "-V"]
+    assert local_plan.remote_command is None
+
+
+def test_run_raises_host_key_verification_error(tmp_path):
+    known_hosts = tmp_path / "known_hosts"
+    known_hosts.write_text("203.0.113.10 ssh-ed25519 AAAAC3NzaTest\n")
+
+    def fake_run(argv, capture_output, text, timeout):
+        return subprocess.CompletedProcess(
+            argv,
+            255,
+            stdout="",
+            stderr="Host key verification failed.\n",
+        )
+
+    executor = ssh_trust.VerifiedSSHExecutor(
+        known_hosts_path=known_hosts,
+        runner=fake_run,
+    )
+
+    with pytest.raises(ssh_trust.HostKeyVerificationError):
+        executor.run("203.0.113.10", ["true"])