feat: Emacs Sovereign Control Plane (#590)

KT from ezra. Shared Emacs daemon as real-time fleet orchestration hub.

## What
- Emacs daemon init.el with org-mode dispatch queue (dispatch.org)
- fleet-log.org for append-only fleet events
- Python FleetBridge client (local + SSH modes)
- Wrapper scripts: fleet-append, fleet-poll, fleet-claim
- systemd service for auto-start on boot
- Install script for one-command deployment
- 29 tests covering all bridge operations

## Architecture
Bezalel runs emacs --daemon=bezalel. Other agents SSH in and call
emacsclient to claim/complete tasks, log events, check status.

## Files
- emacs/sovereign-control-plane/init.el — daemon config
- emacs/fleet_bridge.py — Python client library
- emacs/scripts/fleet-append, fleet-poll, fleet-claim — CLI wrappers
- emacs/scripts/install-control-plane.sh — deploy script
- emacs/README.md — full documentation
- tests/fleet_control/test_fleet_bridge.py — 29 tests

Closes #590.

emacs/README.md

@@ -0,0 +1,162 @@
+# Emacs Sovereign Control Plane
+
+A shared, stateful orchestration hub for the Timmy Fleet. Uses an Emacs daemon
+as a real-time, programmable whiteboard and task queue.
+
+## Architecture
+
+```
+┌─────────────────────────────────────────────────────┐
+│                   Bezalel VPS                       │
+│                                                     │
+│  ┌───────────────┐      ┌────────────────────────┐ │
+│  │ Emacs Daemon  │◄────►│ /srv/fleet/workspace/  │ │
+│  │ (bezalel)     │      │   dispatch.org         │ │
+│  │               │      │   fleet-log.org        │ │
+│  └───────┬───────┘      └────────────────────────┘ │
+│          │ socket                                   │
+│  ┌───────┴───────┐                                 │
+│  │ emacsclient   │                                 │
+│  └───────┬───────┘                                 │
+└──────────┼──────────────────────────────────────────┘
+           │ SSH
+    ┌──────┴──────┐
+    │  Any Agent  │
+    │ (timmy,     │
+    │  allegro,   │
+    │  ezra...)   │
+    └─────────────┘
+```
+
+## Quick Start
+
+### Deploy (on Bezalel)
+
+```bash
+cd emacs/scripts
+sudo bash install-control-plane.sh
+```
+
+### From Any Agent
+
+```bash
+# Poll for tasks assigned to you
+ssh root@bezalel 'fleet-poll timmy'
+
+# Claim a task
+ssh root@bezalel '/usr/local/bin/fleet-claim "Fix deployment" timmy'
+
+# Log an update
+ssh root@bezalel 'fleet-append "Completed deployment of v2.3"'
+
+# Check status
+ssh root@bezalel 'emacsclient -s /root/.emacs.d/server/bezalel -e "(fleet-status)"'
+```
+
+### From Python
+
+```python
+from emacs.fleet_bridge import FleetBridge
+
+bridge = FleetBridge(host="root@bezalel")
+
+# Poll for tasks
+tasks = bridge.poll(agent="timmy")
+
+# Add a task
+bridge.add_task("Research Q3 benchmarks", assigned="claude", priority="high")
+
+# Claim and complete
+bridge.claim("Research Q3 benchmarks", agent="claude")
+bridge.complete("Research Q3 benchmarks", result="Published report at PR #789")
+
+# Check status
+status = bridge.status()
+print(f"{status['pending']} pending, {status['in_progress']} in-progress")
+```
+
+## dispatch.org Format
+
+Tasks are org-mode headlines with properties:
+
+```org
+* Tasks
+** PENDING Fix deployment bug
+   :PROPERTIES:
+   :CREATED: [2026-04-13 Sun 15:30]
+   :ASSIGNED: timmy
+   :PRIORITY: high
+   :ID: abc123
+   :END:
+   The deployment script fails on line 42 when SSL is enabled.
+
+** IN_PROGRESS Research benchmarks
+   :PROPERTIES:
+   :CREATED: [2026-04-13 Sun 14:00]
+   :ASSIGNED: claude
+   :PRIORITY: medium
+   :CLAIMED: [2026-04-13 Sun 15:45]
+   :END:
+
+** COMPLETED Set up monitoring
+   :PROPERTIES:
+   :ASSIGNED: allegro
+   :COMPLETED: [2026-04-13 Sun 16:00]
+   :END:
+   Result: Prometheus + Grafana deployed on Bezalel.
+```
+
+## API Reference
+
+### FleetBridge (Python)
+
+| Method | Description |
+|--------|-------------|
+| `poll(agent=None)` | List pending tasks, optionally filtered by agent |
+| `claim(title, agent)` | Set task to IN_PROGRESS |
+| `complete(title, result=None)` | Mark task COMPLETED |
+| `block(title, reason)` | Mark task BLOCKED |
+| `add_task(title, assigned, priority, body)` | Add new task |
+| `log(message)` | Append to fleet log |
+| `status()` | Get queue summary |
+| `is_reachable()` | Check daemon connectivity |
+
+### Wrapper Scripts
+
+| Script | Usage |
+|--------|-------|
+| `fleet-append "msg"` | Append to log |
+| `fleet-poll [agent]` | List tasks |
+| `fleet-claim "Title" agent` | Claim task |
+
+### Emacs Functions (via emacsclient)
+
+| Function | Description |
+|----------|-------------|
+| `(fleet-append "msg")` | Append to log |
+| `(fleet-poll)` or `(fleet-poll "agent")` | List tasks |
+| `(fleet-claim "title" "agent")` | Claim task |
+| `(fleet-complete "title")` or `(fleet-complete "title" "result")` | Complete task |
+| `(fleet-block "title" "reason")` | Block task |
+| `(fleet-add-task "title" "assigned" "priority")` | Add task |
+| `(fleet-status)` | Queue summary |
+
+## Why Emacs?
+
+Unlike Gitea (async, request-based), the Emacs bridge enables:
+- **Real-time sync** — auto-revert mode watches for changes
+- **Shared executable notebooks** — org-babel for live code execution
+- **Persistent live view** — Alexander can watch dispatch.org in a tmux pane
+- **Programmable** — any elisp function is callable via emacsclient
+- **Zero dependencies** — emacs + SSH, no database, no API server
+
+## Integration with Hermes Cron
+
+Schedule periodic dispatch checks via Hermes cron:
+
+```bash
+hermes cron create --profile default \
+  --schedule "every 10m" \
+  --prompt "Poll the Emacs fleet dispatch for tasks assigned to timmy. If any PENDING tasks exist, claim the highest priority one and work on it. Use: ssh root@bezalel 'fleet-poll timmy'" \
+  --name "fleet-dispatch-poll"
+```

emacs/__init__.py

@@ -0,0 +1 @@
+# Fleet Control Plane package

emacs/fleet_bridge.py

@@ -0,0 +1,362 @@
+"""
+Emacs Fleet Bridge — Python client for the Sovereign Control Plane.
+
+Provides a programmatic interface to the shared Emacs daemon running on
+Bezalel. All operations go through emacsclient over SSH or local socket.
+
+Usage:
+    from emacs.fleet_bridge import FleetBridge
+    
+    # Connect to the remote daemon
+    bridge = FleetBridge(host="root@bezalel", socket="/root/.emacs.d/server/bezalel")
+    
+    # Poll for tasks
+    tasks = bridge.poll(agent="timmy")
+    
+    # Claim a task
+    bridge.claim("Fix deployment bug", agent="timmy")
+    
+    # Append to fleet log
+    bridge.log("Completed deployment of v2.3")
+    
+    # Add a new task
+    bridge.add_task("Research Q3 benchmarks", assigned="claude", priority="high")
+    
+    # Complete a task
+    bridge.complete("Fix deployment bug", result="Merged PR #456")
+"""
+
+import json
+import logging
+import os
+import shlex
+import subprocess
+from dataclasses import dataclass, field
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+DEFAULT_SOCKET = "/root/.emacs.d/server/bezalel"
+DEFAULT_HOST = "root@bezalel"  # SSH target
+DEFAULT_TIMEOUT = 30  # seconds
+
+FLEET_APPEND_WRAPPER = "/usr/local/bin/fleet-append"
+DISPATCH_FILE = "/srv/fleet/workspace/dispatch.org"
+LOG_FILE = "/srv/fleet/workspace/fleet-log.org"
+
+
+@dataclass
+class FleetBridgeConfig:
+    """Configuration for the Emacs Fleet Bridge."""
+    host: str = DEFAULT_HOST
+    socket: str = DEFAULT_SOCKET
+    timeout: int = DEFAULT_TIMEOUT
+    ssh_key: Optional[str] = None
+    local_mode: bool = False  # If True, don't SSH — run emacsclient locally
+
+
+class FleetBridge:
+    """Python client for the Emacs Sovereign Control Plane.
+
+    Communicates with the shared Emacs daemon via emacsclient, either
+    directly (local) or over SSH (remote).
+    """
+
+    def __init__(
+        self,
+        host: str = DEFAULT_HOST,
+        socket: str = DEFAULT_SOCKET,
+        timeout: int = DEFAULT_TIMEOUT,
+        ssh_key: Optional[str] = None,
+        local_mode: bool = False,
+    ):
+        self.config = FleetBridgeConfig(
+            host=host,
+            socket=socket,
+            timeout=timeout,
+            ssh_key=ssh_key,
+            local_mode=local_mode,
+        )
+        self._validate_config()
+
+    def _validate_config(self) -> None:
+        """Validate configuration and check connectivity."""
+        pass  # Lazy validation — fail on first call if misconfigured
+
+    def _build_remote_prefix(self) -> list[str]:
+        """Build the SSH command prefix for remote execution."""
+        cmd = ["ssh", "-o", "StrictHostKeyChecking=no"]
+        if self.config.ssh_key:
+            cmd.extend(["-i", self.config.ssh_key])
+        cmd.extend(["-o", "ConnectTimeout=10"])
+        cmd.append(self.config.host)
+        return cmd
+
+    def _build_emacsclient_cmd(self, elisp: str) -> list[str]:
+        """Build the full emacsclient command."""
+        ec_cmd = ["emacsclient", "-s", self.config.socket, "-e", elisp]
+
+        if self.config.local_mode:
+            return ec_cmd
+        else:
+            return self._build_remote_prefix() + ["--"] + ec_cmd
+
+    def _run(self, elisp: str, timeout: Optional[int] = None) -> tuple[bool, str]:
+        """Execute an elisp expression via emacsclient.
+
+        Returns (success, output_or_error).
+        """
+        cmd = self._build_emacsclient_cmd(elisp)
+        t = timeout or self.config.timeout
+
+        try:
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=t,
+            )
+            if result.returncode != 0:
+                return False, result.stderr.strip() or f"emacsclient exited with code {result.returncode}"
+            return True, result.stdout.strip()
+        except subprocess.TimeoutExpired:
+            return False, f"Command timed out after {t}s"
+        except FileNotFoundError:
+            return False, "emacsclient not found — is emacs installed?"
+        except Exception as e:
+            return False, str(e)
+
+    def _run_wrapper(self, args: list[str]) -> tuple[bool, str]:
+        """Run the fleet-append wrapper script (simpler interface)."""
+        if self.config.local_mode:
+            cmd = [FLEET_APPEND_WRAPPER] + args
+        else:
+            cmd = self._build_remote_prefix() + ["--", FLEET_APPEND_WRAPPER] + args
+
+        try:
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=self.config.timeout,
+            )
+            if result.returncode != 0:
+                return False, result.stderr.strip() or f"wrapper exited with code {result.returncode}"
+            return True, result.stdout.strip()
+        except Exception as e:
+            return False, str(e)
+
+    def _escape_elisp_string(self, s: str) -> str:
+        """Escape a string for embedding in an elisp expression."""
+        return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
+
+    # -----------------------------------------------------------------------
+    # Public API
+    # -----------------------------------------------------------------------
+
+    def log(self, message: str) -> str:
+        """Append a message to the fleet log.
+
+        Returns the log entry created, or raises on failure.
+        """
+        escaped = self._escape_elisp_string(message)
+        ok, out = self._run(f'(fleet-append "{escaped}")')
+        if not ok:
+            raise FleetBridgeError(f"Failed to log message: {out}")
+        return out
+
+    def log_via_wrapper(self, message: str) -> str:
+        """Append via the fleet-append wrapper (simpler, less reliable)."""
+        ok, out = self._run_wrapper([message])
+        if not ok:
+            raise FleetBridgeError(f"Failed to log via wrapper: {out}")
+        return out
+
+    def add_task(
+        self,
+        title: str,
+        assigned: str = "all",
+        priority: str = "medium",
+        body: Optional[str] = None,
+    ) -> dict:
+        """Add a new task to the dispatch queue.
+
+        Args:
+            title: Task title
+            assigned: Agent name (e.g. 'timmy', 'allegro', 'all')
+            priority: 'high', 'medium', or 'low'
+            body: Optional task description
+
+        Returns:
+            Dict with task_id, title, assigned, priority.
+        """
+        esc_title = self._escape_elisp_string(title)
+        esc_body = self._escape_elisp_string(body) if body else ""
+
+        if body:
+            elisp = f'(fleet-add-task "{esc_title}" "{assigned}" "{priority}" "{esc_body}")'
+        else:
+            elisp = f'(fleet-add-task "{esc_title}" "{assigned}" "{priority}")'
+
+        ok, out = self._run(elisp)
+        if not ok:
+            raise FleetBridgeError(f"Failed to add task: {out}")
+
+        # Parse the response to extract task ID
+        return {
+            "raw_response": out,
+            "title": title,
+            "assigned": assigned,
+            "priority": priority,
+        }
+
+    def poll(self, agent: Optional[str] = None) -> list[dict]:
+        """Poll for available tasks.
+
+        Args:
+            agent: Filter to tasks assigned to this agent (or 'all').
+
+        Returns:
+            List of task dicts with title, assigned, priority, created.
+        """
+        if agent:
+            esc_agent = self._escape_elisp_string(agent)
+            elisp = f'(fleet-poll "{esc_agent}")'
+        else:
+            elisp = '(fleet-poll)'
+
+        ok, out = self._run(elisp)
+        if not ok:
+            raise FleetBridgeError(f"Failed to poll: {out}")
+
+        # Parse the response
+        tasks = []
+        for line in out.split("\n"):
+            line = line.strip()
+            if line.startswith("["):
+                # Format: [priority] title (assigned: X, created: Y)
+                tasks.append({"raw": line})
+        return tasks
+
+    def claim(self, task_title: str, agent: str) -> str:
+        """Claim a task by updating its status to IN_PROGRESS.
+
+        Args:
+            task_title: Substring of the task title to match
+            agent: Agent claiming the task
+
+        Returns:
+            Confirmation message.
+        """
+        esc_title = self._escape_elisp_string(task_title)
+        esc_agent = self._escape_elisp_string(agent)
+        elisp = f'(fleet-claim "{esc_title}" "{esc_agent}")'
+
+        ok, out = self._run(elisp)
+        if not ok:
+            raise FleetBridgeError(f"Failed to claim task: {out}")
+        return out
+
+    def complete(self, task_title: str, result: Optional[str] = None) -> str:
+        """Mark a task as completed.
+
+        Args:
+            task_title: Substring of the task title to match
+            result: Optional result string to append
+
+        Returns:
+            Confirmation message.
+        """
+        esc_title = self._escape_elisp_string(task_title)
+        if result:
+            esc_result = self._escape_elisp_string(result)
+            elisp = f'(fleet-complete "{esc_title}" "{esc_result}")'
+        else:
+            elisp = f'(fleet-complete "{esc_title}")'
+
+        ok, out = self._run(elisp)
+        if not ok:
+            raise FleetBridgeError(f"Failed to complete task: {out}")
+        return out
+
+    def block(self, task_title: str, reason: str) -> str:
+        """Mark a task as blocked.
+
+        Args:
+            task_title: Substring of the task title to match
+            reason: Why the task is blocked
+
+        Returns:
+            Confirmation message.
+        """
+        esc_title = self._escape_elisp_string(task_title)
+        esc_reason = self._escape_elisp_string(reason)
+        elisp = f'(fleet-block "{esc_title}" "{esc_reason}")'
+
+        ok, out = self._run(elisp)
+        if not ok:
+            raise FleetBridgeError(f"Failed to block task: {out}")
+        return out
+
+    def status(self) -> dict:
+        """Get fleet status summary.
+
+        Returns:
+            Dict with pending, in_progress, completed, blocked counts.
+        """
+        ok, out = self._run("(fleet-status)")
+        if not ok:
+            raise FleetBridgeError(f"Failed to get status: {out}")
+
+        # Parse: "Fleet Status: X pending, Y in-progress, Z completed, W blocked"
+        import re
+        match = re.search(r"(\d+) pending, (\d+) in-progress, (\d+) completed, (\d+) blocked", out)
+        if match:
+            return {
+                "pending": int(match.group(1)),
+                "in_progress": int(match.group(2)),
+                "completed": int(match.group(3)),
+                "blocked": int(match.group(4)),
+                "raw": out,
+            }
+        return {"raw": out}
+
+    def eval_elisp(self, expression: str) -> tuple[bool, str]:
+        """Execute an arbitrary elisp expression (escape hatch)."""
+        return self._run(expression)
+
+    def is_reachable(self) -> bool:
+        """Check if the Emacs daemon is reachable."""
+        ok, out = self._run("(+ 1 1)", timeout=10)
+        return ok and "2" in out
+
+
+class FleetBridgeError(Exception):
+    """Error communicating with the Emacs Fleet Bridge."""
+    pass
+
+
+# ---------------------------------------------------------------------------
+# Convenience: auto-detect connection mode
+# ---------------------------------------------------------------------------
+
+def auto_connect(
+    socket: str = DEFAULT_SOCKET,
+    host: str = DEFAULT_HOST,
+) -> FleetBridge:
+    """Auto-detect local vs remote mode and connect.
+
+    If the socket file exists locally, use local mode.
+    Otherwise, connect via SSH.
+    """
+    local_socket = os.path.expanduser(socket.replace("/root/", "~/"))
+    if os.path.exists(local_socket):
+        return FleetBridge(socket=local_socket, local_mode=True)
+    elif os.path.exists(socket):
+        return FleetBridge(socket=socket, local_mode=True)
+    else:
+        return FleetBridge(host=host, socket=socket, local_mode=False)

emacs/scripts/fleet-append

@@ -0,0 +1,31 @@
+#!/bin/bash
+# fleet-append — Simple wrapper to append messages to the fleet log.
+#
+# Usage: /usr/local/bin/fleet-append "Your message here"
+#
+# This is the fast-track wrapper for agents. For full control,
+# use emacsclient directly or the Python FleetBridge client.
+
+set -euo pipefail
+
+SOCKET="/root/.emacs.d/server/bezalel"
+LOG_FILE="/srv/fleet/workspace/fleet-log.org"
+
+if [ $# -eq 0 ]; then
+    echo "Usage: fleet-append \"message\"" >&2
+    exit 1
+fi
+
+MESSAGE="$1"
+TIMESTAMP="$(date '+[%Y-%m-%d %a %H:%M:%S]')"
+
+# Try emacsclient first, fall back to direct file append
+if emacsclient -s "$SOCKET" -e '(+ 1 1)' &>/dev/null; then
+    # Daemon is running — use it
+    RESULT=$(emacsclient -s "$SOCKET" -e "(fleet-append \"$(echo "$MESSAGE" | sed 's/"/\\"/g' | sed 's/\n/\\n/g')\")")
+    echo "$RESULT"
+else
+    # Daemon down — direct append (degraded mode)
+    echo "${TIMESTAMP} ${MESSAGE}" >> "$LOG_FILE"
+    echo "${TIMESTAMP} ${MESSAGE}"
+fi

emacs/scripts/fleet-claim

@@ -0,0 +1,28 @@
+#!/bin/bash
+# fleet-claim — Claim a task from the dispatch queue.
+#
+# Usage: fleet-claim "Task Title" agent_name
+
+set -euo pipefail
+
+SOCKET="/root/.emacs.d/server/bezalel"
+
+if [ $# -lt 2 ]; then
+    echo "Usage: fleet-claim \"Task Title\" agent_name" >&2
+    exit 1
+fi
+
+TASK_TITLE="$1"
+AGENT="$2"
+
+# Escape for elisp
+ESC_TITLE=$(echo "$TASK_TITLE" | sed 's/"/\\"/g')
+ESC_AGENT=$(echo "$AGENT" | sed 's/"/\\"/g')
+
+if emacsclient -s "$SOCKET" -e '(+ 1 1)' &>/dev/null; then
+    RESULT=$(emacsclient -s "$SOCKET" -e "(fleet-claim \"$ESC_TITLE\" \"$ESC_AGENT\")")
+    echo "$RESULT"
+else
+    echo "Error: Emacs daemon not reachable at $SOCKET" >&2
+    exit 1
+fi

emacs/scripts/fleet-poll

@@ -0,0 +1,39 @@
+#!/bin/bash
+# fleet-poll — Poll the dispatch queue for available tasks.
+#
+# Usage: fleet-poll [agent_name]
+#   If agent_name is provided, filters to tasks assigned to that agent.
+#   Without arguments, shows all pending tasks.
+
+set -euo pipefail
+
+SOCKET="/root/.emacs.d/server/bezalel"
+DISPATCH_FILE="/srv/fleet/workspace/dispatch.org"
+
+AGENT="${1:-}"
+
+if emacsclient -s "$SOCKET" -e '(+ 1 1)' &>/dev/null; then
+    if [ -n "$AGENT" ]; then
+        RESULT=$(emacsclient -s "$SOCKET" -e "(fleet-poll \"$AGENT\")")
+    else
+        RESULT=$(emacsclient -s "$SOCKET" -e '(fleet-poll)')
+    fi
+    echo "$RESULT"
+else
+    # Fallback: grep dispatch.org directly
+    if [ -n "$AGENT" ]; then
+        echo "Available tasks for $AGENT:"
+        grep -E "^\*\* PENDING" "$DISPATCH_FILE" 2>/dev/null | \
+            while read -r line; do
+                title="${line##** PENDING }"
+                echo "  - $title"
+            done
+    else
+        echo "Available tasks:"
+        grep -E "^\*\* PENDING" "$DISPATCH_FILE" 2>/dev/null | \
+            while read -r line; do
+                title="${line##** PENDING }"
+                echo "  - $title"
+            done
+    fi
+fi

emacs/scripts/install-control-plane.sh

@@ -0,0 +1,99 @@
+#!/bin/bash
+# install-control-plane.sh — Deploy the Emacs Sovereign Control Plane
+#
+# Run on the target VPS (Bezalel) to set up:
+# 1. Emacs daemon with sovereign-control-plane/init.el
+# 2. Fleet workspace (/srv/fleet/workspace/)
+# 3. Wrapper scripts in /usr/local/bin/
+# 4. systemd service for auto-start
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+EMACS_DIR="$SCRIPT_DIR/.."
+WORKSPACE="/srv/fleet/workspace"
+SOCKET_DIR="/root/.emacs.d/server"
+
+echo "=== Emacs Sovereign Control Plane — Installer ==="
+
+# 1. Ensure emacs is installed
+if ! command -v emacs &>/dev/null; then
+    echo "Installing emacs..."
+    if command -v apt-get &>/dev/null; then
+        apt-get update && apt-get install -y emacs-nox
+    elif command -v yum &>/dev/null; then
+        yum install -y emacs-nox
+    else
+        echo "Error: Cannot determine package manager. Install emacs manually." >&2
+        exit 1
+    fi
+fi
+
+# 2. Create workspace
+echo "Creating workspace at $WORKSPACE..."
+mkdir -p "$WORKSPACE"
+mkdir -p "$SOCKET_DIR"
+
+# 3. Install init.el
+echo "Installing init.el..."
+mkdir -p /root/.emacs.d
+cp "$EMACS_DIR/sovereign-control-plane/init.el" /root/.emacs.d/init.el
+
+# 4. Install wrapper scripts
+echo "Installing wrapper scripts..."
+cp "$EMACS_DIR/scripts/fleet-append" /usr/local/bin/fleet-append
+cp "$EMACS_DIR/scripts/fleet-poll" /usr/local/bin/fleet-poll
+cp "$EMACS_DIR/scripts/fleet-claim" /usr/local/bin/fleet-claim
+chmod +x /usr/local/bin/fleet-append
+chmod +x /usr/local/bin/fleet-poll
+chmod +x /usr/local/bin/fleet-claim
+
+# 5. Create systemd service
+echo "Creating systemd service..."
+cat > /etc/systemd/system/fleet-control-plane.service << 'EOF'
+[Unit]
+Description=Emacs Sovereign Control Plane Daemon
+After=network.target
+
+[Service]
+Type=forking
+ExecStart=/usr/bin/emacs --daemon=bezalel
+ExecStop=/usr/bin/emacsclient -s /root/.emacs.d/server/bezalel -e "(kill-emacs)"
+Restart=on-failure
+RestartSec=10
+User=root
+Environment=HOME=/root
+
+[Install]
+WantedBy=multi-user.target
+EOF
+
+systemctl daemon-reload
+systemctl enable fleet-control-plane.service
+
+# 6. Start or restart the daemon
+if pgrep -f "emacs.*daemon.*bezalel" > /dev/null; then
+    echo "Restarting existing daemon..."
+    emacsclient -s "$SOCKET_DIR/bezalel" -e '(kill-emacs)' 2>/dev/null || true
+    sleep 2
+fi
+
+echo "Starting daemon..."
+systemctl start fleet-control-plane.service
+
+# 7. Verify
+sleep 2
+if emacsclient -s "$SOCKET_DIR/bezalel" -e '(+ 1 1)' 2>/dev/null | grep -q "2"; then
+    echo "✓ Control plane daemon is running."
+    echo "  Socket: $SOCKET_DIR/bezalel"
+    echo "  Workspace: $WORKSPACE"
+    echo ""
+    echo "Test it:"
+    echo "  fleet-poll"
+    echo "  fleet-append 'Control plane online'"
+else
+    echo "⚠ Daemon may not be ready yet. Check: systemctl status fleet-control-plane"
+fi
+
+echo ""
+echo "=== Installation complete ==="

emacs/sovereign-control-plane/init.el

@@ -0,0 +1,233 @@
+;;; init.el — Emacs Sovereign Control Plane for Timmy Fleet
+;;; 
+;;; Shared daemon running on Bezalel. Provides a real-time, programmable
+;;; whiteboard and task queue for the entire fleet via org-mode dispatch.
+;;; 
+;;; Daemon: emacs --daemon=bezalel
+;;; Client: emacsclient -s /root/.emacs.d/server/bezalel -e "(your-elisp)"
+;;; Wrapper: /usr/local/bin/fleet-append "message"
+
+(require 'org)
+(require 'org-id)
+(require 'ox)
+(require 'server)
+
+;; ---------------------------------------------------------------------------
+;; Server configuration
+;; ---------------------------------------------------------------------------
+
+;; Start server on named socket
+(unless (server-running-p)
+  (setq server-socket-dir "~/.emacs.d/server")
+  (setq server-name "bezalel")
+  (server-start))
+
+;; ---------------------------------------------------------------------------
+;; Dispatch workspace
+;; ---------------------------------------------------------------------------
+
+(defvar fleet-workspace "/srv/fleet/workspace"
+  "Root directory for fleet workspace files.")
+
+(defvar fleet-dispatch-file "/srv/fleet/workspace/dispatch.org"
+  "Central dispatch hub — org file with task queue.")
+
+(defvar fleet-log-file "/srv/fleet/workspace/fleet-log.org"
+  "Append-only log of fleet events.")
+
+;; Ensure workspace exists
+(make-directory fleet-workspace t)
+
+;; ---------------------------------------------------------------------------
+;; Dispatch.org template
+;; ---------------------------------------------------------------------------
+
+(defun fleet--ensure-dispatch ()
+  "Create dispatch.org with template if it doesn't exist."
+  (unless (file-exists-p fleet-dispatch-file)
+    (with-temp-file fleet-dispatch-file
+      (insert "#+TITLE: Fleet Dispatch Hub\n")
+      (insert "#+AUTHOR: Timmy Fleet Control Plane\n")
+      (insert (format "#+DATE: %s\n" (format-time-string "%Y-%m-%d %H:%M")))
+      (insert "#+DESCRIPTION: Central task queue for the sovereign AI fleet\n\n")
+      (insert "* Tasks [0/0]\n")
+      (insert "** PENDING Welcome\n")
+      (insert "   :PROPERTIES:\n")
+      (insert (format "   :CREATED: %s\n" (format-time-string "[%Y-%m-%d %a %H:%M]")))
+      (insert "   :ASSIGNED: all\n")
+      (insert "   :PRIORITY: low\n")
+      (insert "   :END:\n")
+      (insert "   First task: Review the control plane and begin using dispatch.org.\n\n")
+      (insert "* In Progress\n\n")
+      (insert "* Completed\n\n")
+      (insert "* Blocked\n"))))
+
+(fleet--ensure-dispatch)
+
+;; ---------------------------------------------------------------------------
+;; Fleet log template
+;; ---------------------------------------------------------------------------
+
+(defun fleet--ensure-log ()
+  "Create fleet-log.org if it doesn't exist."
+  (unless (file-exists-p fleet-log-file)
+    (with-temp-file fleet-log-file
+      (insert "#+TITLE: Fleet Event Log\n")
+      (insert "#+DESCRIPTION: Append-only log of fleet actions and events\n\n")
+      (insert "* Log\n"))))
+
+(fleet--ensure-log)
+
+;; ---------------------------------------------------------------------------
+;; Core functions — callable via emacsclient -e
+;; ---------------------------------------------------------------------------
+
+(defun fleet-append (message)
+  "Append a message to the fleet log with timestamp.
+Returns the log entry created."
+  (let* ((timestamp (format-time-string "[%Y-%m-%d %a %H:%M:%S]"))
+         (entry (format "%s %s\n" timestamp message)))
+    (with-temp-buffer
+      (insert entry)
+      (append-to-file (point-min) (point-max) fleet-log-file))
+    entry))
+
+(defun fleet-append-task (title assigned priority &optional body)
+  "Add a new PENDING task to dispatch.org.
+TITLE: task title
+ASSIGNED: agent name (e.g. 'timmy', 'allegro', 'all')
+PRIORITY: 'high', 'medium', or 'low'
+BODY: optional task description"
+  (let* ((timestamp (format-time-string "[%Y-%m-%d %a %H:%M]"))
+         (task-id (org-id-new))
+         (entry (format "** PENDING %s\n   :PROPERTIES:\n   :CREATED: %s\n   :ASSIGNED: %s\n   :PRIORITY: %s\n   :ID: %s\n   :END:\n%s\n"
+                        title timestamp assigned priority task-id
+                        (if body (format "   %s\n" body) ""))))
+    (with-temp-buffer
+      (insert entry)
+      (append-to-file (point-min) (point-max) fleet-dispatch-file))
+    (format "Task '%s' assigned to %s (priority: %s, id: %s)" title assigned priority task-id)))
+
+(defun fleet-claim (task-title agent)
+  "Claim a task by updating its status to IN_PROGRESS.
+TASK-TITLE: substring of the task title to match
+AGENT: agent claiming the task"
+  (with-current-buffer (find-file-noselect fleet-dispatch-file)
+    (goto-char (point-min))
+    (if (re-search-forward (format "\\*\\* PENDING %s" (regexp-quote task-title)) nil t)
+        (progn
+          (beginning-of-line)
+          (org-todo 'right)  ;; PENDING -> IN_PROGRESS
+          ;; Update ASSIGNED property
+          (org-entry-put nil "ASSIGNED" agent)
+          (org-entry-put nil "CLAIMED" (format-time-string "[%Y-%m-%d %a %H:%M]"))
+          (save-buffer)
+          (format "Task '%s' claimed by %s" task-title agent))
+      (format "Task '%s' not found (already claimed or completed?)" task-title))))
+
+(defun fleet-complete (task-title &optional result)
+  "Mark a task as completed.
+TASK-TITLE: substring of the task title to match
+RESULT: optional result string to append"
+  (with-current-buffer (find-file-noselect fleet-dispatch-file)
+    (goto-char (point-min))
+    (if (re-search-forward (format "\\*\\* IN_PROGRESS %s" (regexp-quote task-title)) nil t)
+        (progn
+          (beginning-of-line)
+          (org-todo 'right)  ;; IN_PROGRESS -> COMPLETED
+          (org-entry-put nil "COMPLETED" (format-time-string "[%Y-%m-%d %a %H:%M]"))
+          (when result
+            (end-of-line)
+            (insert (format "\n   Result: %s" result)))
+          (save-buffer)
+          (format "Task '%s' completed" task-title))
+      (format "Task '%s' not found in IN_PROGRESS" task-title))))
+
+(defun fleet-block (task-title reason)
+  "Mark a task as blocked with a reason.
+TASK-TITLE: substring of the task title to match
+REASON: why the task is blocked"
+  (with-current-buffer (find-file-noselect fleet-dispatch-file)
+    (goto-char (point-min))
+    (if (re-search-forward (format "\\*\\* \\(PENDING\\|IN_PROGRESS\\) %s" (regexp-quote task-title)) nil t)
+        (progn
+          (beginning-of-line)
+          (org-todo "BLOCKED")
+          (org-entry-put nil "BLOCKED-REASON" reason)
+          (save-buffer)
+          (format "Task '%s' blocked: %s" task-title reason))
+      (format "Task '%s' not found" task-title))))
+
+(defun fleet-poll (&optional agent)
+  "List pending/available tasks.
+If AGENT is provided, filter to tasks assigned to that agent or 'all'.
+Returns task summaries as a string."
+  (with-current-buffer (find-file-noselect fleet-dispatch-file)
+    (goto-char (point-min))
+    (let ((tasks '()))
+      (while (re-search-forward "^\\*\\* PENDING \\(.+\\)$" nil t)
+        (let* ((title (match-string 1))
+               (assigned (org-entry-get nil "ASSIGNED"))
+               (priority (org-entry-get nil "PRIORITY"))
+               (created (org-entry-get nil "CREATED")))
+          (when (or (null agent)
+                    (string= assigned agent)
+                    (string= assigned "all"))
+            (push (format "  [%s] %s (assigned: %s, created: %s)"
+                          (or priority "?") title (or assigned "?") (or created "?"))
+                  tasks))))
+      (if tasks
+          (format "Available tasks:\n%s" (mapconcat #'identity (nreverse tasks) "\n"))
+        "No available tasks."))))
+
+(defun fleet-status ()
+  "Return a status summary of the dispatch queue."
+  (with-current-buffer (find-file-noselect fleet-dispatch-file)
+    (goto-char (point-min))
+    (let ((pending 0) (in-progress 0) (completed 0) (blocked 0))
+      (while (re-search-forward "^\\*\\* \\(PENDING\\|IN_PROGRESS\\|COMPLETED\\|BLOCKED\\)" nil t)
+        (let ((state (match-string 1)))
+          (cond
+           ((string= state "PENDING") (setq pending (1+ pending)))
+           ((string= state "IN_PROGRESS") (setq in-progress (1+ in-progress)))
+           ((string= state "COMPLETED") (setq completed (1+ completed)))
+           ((string= state "BLOCKED") (setq blocked (1+ blocked))))))
+      (format "Fleet Status: %d pending, %d in-progress, %d completed, %d blocked"
+              pending in-progress completed blocked))))
+
+;; ---------------------------------------------------------------------------
+;; Org-mode workflow customization
+;; ---------------------------------------------------------------------------
+
+(setq org-todo-keywords
+      '((sequence "PENDING" "IN_PROGRESS" "COMPLETED" "BLOCKED")))
+
+(setq org-todo-keyword-faces
+      '(("PENDING" . (:foreground "orange" :weight bold))
+        ("IN_PROGRESS" . (:foreground "deep sky blue" :weight bold))
+        ("COMPLETED" . (:foreground "green" :weight bold))
+        ("BLOCKED" . (:foreground "red" :weight bold))))
+
+;; ---------------------------------------------------------------------------
+;; Auto-reload dispatch.org on changes (for real-time sync)
+;; ---------------------------------------------------------------------------
+
+(defun fleet--auto-revert ()
+  "Auto-revert dispatch.org when it changes on disk."
+  (with-current-buffer (get-file-buffer fleet-dispatch-file)
+    (auto-revert-mode 1)
+    (setq auto-revert-interval 5)))
+
+;; Hook into dispatch.org buffer
+(add-hook 'find-file-hook
+          (lambda ()
+            (when (and buffer-file-name
+                       (string= buffer-file-name fleet-dispatch-file))
+              (fleet--auto-revert))))
+
+;; ---------------------------------------------------------------------------
+;; Startup message
+;; ---------------------------------------------------------------------------
+
+(fleet-append "Control plane daemon started.")
+(message "Fleet Control Plane ready. Socket: %s" server-socket-dir)

tests/fleet_control/test_fleet_bridge.py

@@ -0,0 +1,299 @@
+"""Tests for the Emacs Fleet Bridge client.
+
+Tests the Python interface to the Sovereign Control Plane, using mocks
+for the SSH/emacsclient subprocess calls.
+"""
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch, call
+
+import pytest
+
+# Ensure the project root is on sys.path
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from emacs.fleet_bridge import FleetBridge, FleetBridgeError, FleetBridgeConfig
+
+
+@pytest.fixture
+def bridge():
+    """Create a FleetBridge in local mode for testing."""
+    return FleetBridge(socket="/tmp/test-bezalel", local_mode=True)
+
+
+@pytest.fixture
+def remote_bridge():
+    """Create a FleetBridge in remote mode for testing."""
+    return FleetBridge(host="root@testhost", socket="/root/.emacs.d/server/bezalel", local_mode=False)
+
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+class TestFleetBridgeConfig:
+    def test_default_config(self):
+        bridge = FleetBridge()
+        assert bridge.config.host == "root@bezalel"
+        assert bridge.config.socket == "/root/.emacs.d/server/bezalel"
+        assert bridge.config.timeout == 30
+        assert bridge.config.local_mode is False
+
+    def test_local_mode_config(self):
+        bridge = FleetBridge(socket="/tmp/test", local_mode=True)
+        assert bridge.config.local_mode is True
+        assert bridge.config.socket == "/tmp/test"
+
+    def test_remote_config(self):
+        bridge = FleetBridge(host="root@1.2.3.4", socket="/custom/socket")
+        assert bridge.config.host == "root@1.2.3.4"
+        assert bridge.config.socket == "/custom/socket"
+
+
+# ---------------------------------------------------------------------------
+# Command building
+# ---------------------------------------------------------------------------
+
+class TestCommandBuilding:
+    def test_local_emacsclient_cmd(self, bridge):
+        cmd = bridge._build_emacsclient_cmd('(+ 1 1)')
+        assert cmd == ["emacsclient", "-s", "/tmp/test-bezalel", "-e", "(+ 1 1)"]
+
+    def test_remote_cmd_includes_ssh(self, remote_bridge):
+        cmd = remote_bridge._build_emacsclient_cmd('(+ 1 1)')
+        assert cmd[0] == "ssh"
+        assert "root@testhost" in cmd
+        assert "emacsclient" in cmd
+        assert "/root/.emacs.d/server/bezalel" in cmd
+
+    def test_remote_with_ssh_key(self):
+        bridge = FleetBridge(
+            host="root@host",
+            socket="/s",
+            ssh_key="/home/user/.ssh/id_rsa",
+            local_mode=False,
+        )
+        cmd = bridge._build_emacsclient_cmd('(+ 1 1)')
+        assert "-i" in cmd
+        assert "/home/user/.ssh/id_rsa" in cmd
+
+
+# ---------------------------------------------------------------------------
+# String escaping
+# ---------------------------------------------------------------------------
+
+class TestStringEscaping:
+    def test_escape_quotes(self, bridge):
+        assert bridge._escape_elisp_string('say "hello"') == 'say \\"hello\\"'
+
+    def test_escape_backslash(self, bridge):
+        assert bridge._escape_elisp_string("path\\to\\file") == "path\\\\to\\\\file"
+
+    def test_escape_newline(self, bridge):
+        assert bridge._escape_elisp_string("line1\nline2") == "line1\\nline2"
+
+    def test_escape_complex(self, bridge):
+        result = bridge._escape_elisp_string('He said "go\\there"\nok')
+        assert '\\"' in result
+        assert '\\\\' in result
+        assert '\\n' in result
+
+
+# ---------------------------------------------------------------------------
+# Core operations
+# ---------------------------------------------------------------------------
+
+class TestLog:
+    @patch("subprocess.run")
+    def test_log_success(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout='[2026-04-13 Sun 15:30:00] test message\n',
+            stderr="",
+        )
+        result = bridge.log("test message")
+        assert "test message" in result
+        mock_run.assert_called_once()
+
+    @patch("subprocess.run")
+    def test_log_failure(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="connection refused")
+        with pytest.raises(FleetBridgeError, match="connection refused"):
+            bridge.log("test")
+
+    @patch("subprocess.run")
+    def test_log_timeout(self, mock_run, bridge):
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="test", timeout=30)
+        with pytest.raises(FleetBridgeError, match="timed out"):
+            bridge.log("test")
+
+
+class TestAddTask:
+    @patch("subprocess.run")
+    def test_add_task_success(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout='Task \'Fix bug\' assigned to timmy (priority: high, id: abc123)\n',
+            stderr="",
+        )
+        result = bridge.add_task("Fix bug", assigned="timmy", priority="high")
+        assert result["title"] == "Fix bug"
+        assert result["assigned"] == "timmy"
+        assert result["priority"] == "high"
+
+    @patch("subprocess.run")
+    def test_add_task_with_body(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task created\n",
+            stderr="",
+        )
+        bridge.add_task("Fix bug", body="Detailed description here")
+        # Verify the elisp call includes the body
+        call_args = mock_run.call_args[0][0]
+        elisp_expr = call_args[-1]
+        assert "Detailed description" in elisp_expr
+
+    @patch("subprocess.run")
+    def test_add_task_failure(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=0, stdout="nil\n", stderr="")
+        # This should still work — emacs returns nil for no-ops
+        result = bridge.add_task("Test")
+        assert result is not None
+
+
+class TestPoll:
+    @patch("subprocess.run")
+    def test_poll_returns_tasks(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout='Available tasks:\n  [high] Fix bug (assigned: timmy, created: [2026-04-13])\n  [low] Research (assigned: all, created: [2026-04-12])\n',
+            stderr="",
+        )
+        tasks = bridge.poll(agent="timmy")
+        assert len(tasks) == 2
+        assert "[high]" in tasks[0]["raw"]
+
+    @patch("subprocess.run")
+    def test_poll_no_tasks(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="No available tasks.\n",
+            stderr="",
+        )
+        tasks = bridge.poll()
+        assert len(tasks) == 0
+
+    @patch("subprocess.run")
+    def test_poll_with_agent_filter(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=0, stdout="Available tasks:\n", stderr="")
+        bridge.poll(agent="timmy")
+        call_args = mock_run.call_args[0][0]
+        elisp_expr = call_args[-1]
+        assert "timmy" in elisp_expr
+
+
+class TestClaim:
+    @patch("subprocess.run")
+    def test_claim_success(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task 'Fix bug' claimed by timmy\n",
+            stderr="",
+        )
+        result = bridge.claim("Fix bug", agent="timmy")
+        assert "claimed" in result.lower()
+
+    @patch("subprocess.run")
+    def test_claim_not_found(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task 'Missing' not found\n",
+            stderr="",
+        )
+        result = bridge.claim("Missing", agent="timmy")
+        assert "not found" in result.lower()
+
+
+class TestComplete:
+    @patch("subprocess.run")
+    def test_complete_success(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task 'Fix bug' completed\n",
+            stderr="",
+        )
+        result = bridge.complete("Fix bug")
+        assert "completed" in result.lower()
+
+    @patch("subprocess.run")
+    def test_complete_with_result(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task 'Fix bug' completed\n",
+            stderr="",
+        )
+        bridge.complete("Fix bug", result="Merged PR #456")
+        call_args = mock_run.call_args[0][0]
+        elisp_expr = call_args[-1]
+        assert "PR #456" in elisp_expr
+
+
+class TestBlock:
+    @patch("subprocess.run")
+    def test_block_success(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Task 'Fix bug' blocked: waiting for API key\n",
+            stderr="",
+        )
+        result = bridge.block("Fix bug", reason="waiting for API key")
+        assert "blocked" in result.lower()
+
+
+class TestStatus:
+    @patch("subprocess.run")
+    def test_status_parses(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="Fleet Status: 3 pending, 1 in-progress, 5 completed, 0 blocked\n",
+            stderr="",
+        )
+        status = bridge.status()
+        assert status["pending"] == 3
+        assert status["in_progress"] == 1
+        assert status["completed"] == 5
+        assert status["blocked"] == 0
+
+
+class TestReachability:
+    @patch("subprocess.run")
+    def test_reachable(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=0, stdout="2\n", stderr="")
+        assert bridge.is_reachable() is True
+
+    @patch("subprocess.run")
+    def test_not_reachable(self, mock_run, bridge):
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="test", timeout=10)
+        assert bridge.is_reachable() is False
+
+    @patch("subprocess.run")
+    def test_not_reachable_connection_refused(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="connection refused")
+        assert bridge.is_reachable() is False
+
+
+# ---------------------------------------------------------------------------
+# Eval escape hatch
+# ---------------------------------------------------------------------------
+
+class TestEvalElisp:
+    @patch("subprocess.run")
+    def test_eval_custom(self, mock_run, bridge):
+        mock_run.return_value = MagicMock(returncode=0, stdout="42\n", stderr="")
+        ok, out = bridge.eval_elisp("(* 6 7)")
+        assert ok is True
+        assert "42" in out