feat: v1.0 — am serve : API HTTP + WebSocket authentifiée pour piloter am à distance (closes #80)

- src/serve.rs : tiny_http (zéro dépendance runtime) — token Bearer obligatoire (comptime-ish), rate limiting fenêtre 60s par IP, routes réutilisant les contrats --json (stats/sessions/ps/providers/run/start/stop/ask) en re-spawnant le binaire (pattern web #54), WebSocket RFC 6455 (handshake 101 + accept sha1, broadcast des events du journal par polling 500ms, première passe = indexation), TLS délégué au reverse proxy (documenté)
- am serve --token <TOKEN> [--host] [--port 8080] [--rate-limit 120] + REPL + help + man
- 9 tests : auth (sans/mauvais/bon token), routes (runner mocké), run args, 404, rate limit, handshake RFC 6455 (vecteur connu), broadcast journal, comparaison token
- E2E réel : curl 401/health/stats + POST run exécutant un agent, client WS brut recevant l'event run en temps réel
- version 1.0.0 (milestone v1.0 terminé) ; test version_table suit Cargo.toml (major plus hardcodé)

Cargo.lock

@@ -21,9 +21,10 @@ dependencies = [
 
 [[package]]
 name = "agent-manager"
-version = "0.7.0"
+version = "1.0.0"
 dependencies = [
  "anyhow",
+ "base64",
  "chrono",
  "clap",
  "clap_complete",
@@ -39,6 +40,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_yaml",
+ "sha1",
  "sha2",
  "shell-words",
  "tar",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "agent-manager"
-version = "0.7.0"
+version = "1.0.0"
 edition = "2021"
 description = "Manage local AI coding agents: list, install, start, stop, update — with automatic dependency handling and a YAML-driven catalog."
 license = "MIT"
@@ -32,6 +32,8 @@ shell-words = "1"
 tar = "0.4"
 tempfile = "3"
 tiny_http = "0.12"
+sha1 = "0.10"
+base64 = "0.22"
 toml = "0.8"
 ureq = { version = "2", default-features = false, features = ["tls"] }
 wait-timeout = "0.2"

ROADMAP.md

@@ -478,7 +478,7 @@ alerte).
 | [#77](https://git.dracodev.net/Projets/agent-manager/issues/77) | ✅ Registre communautaire — am registry (publication + recherche sur Gitea) | L |
 | [#78](https://git.dracodev.net/Projets/agent-manager/issues/78) | ✅ am ask — langage naturel → commande am (fournisseur LLM optionnel) | L |
 | [#79](https://git.dracodev.net/Projets/agent-manager/issues/79) | ✅ Profils sandbox par agent (commandes/répertoires autorisés) | L |
-| [#80](https://git.dracodev.net/Projets/agent-manager/issues/80) | am serve --token — API HTTP + WebSocket pour piloter à distance | XL |
+| [#80](https://git.dracodev.net/Projets/agent-manager/issues/80) | ✅ am serve --token — API HTTP + WebSocket pour piloter à distance | XL |
 
 Critère de sortie du jalon : dashboard web complet + API distante.
 

man/am-serve.1

@@ -0,0 +1,25 @@
+.ie \n(.g .ds Aq \(aq
+.el .ds Aq '
+.TH am-serve 1  "serve " 
+.SH NAME
+serve \- Authenticated HTTP + WebSocket API to drive am remotely (issue #80)
+.SH SYNOPSIS
+\fBserve\fR [\fB\-\-token\fR] [\fB\-\-host\fR] [\fB\-\-port\fR] [\fB\-\-rate\-limit\fR] [\fB\-h\fR|\fB\-\-help\fR] 
+.SH DESCRIPTION
+Authenticated HTTP + WebSocket API to drive am remotely (issue #80)
+.SH OPTIONS
+.TP
+\fB\-\-token\fR \fI<TOKEN>\fR
+Bearer token required on every request (mandatory)
+.TP
+\fB\-\-host\fR \fI<HOST>\fR
+Listen address (default 127.0.0.1)
+.TP
+\fB\-\-port\fR \fI<PORT>\fR
+Listening port (default 8080)
+.TP
+\fB\-\-rate\-limit\fR \fI<N>\fR
+Per\-IP request budget per minute (default 120)
+.TP
+\fB\-h\fR, \fB\-\-help\fR
+Print help

man/am.1

@@ -1,6 +1,6 @@
 .ie \n(.g .ds Aq \(aq
 .el .ds Aq '
-.TH am 1  "am 0.7.0" 
+.TH am 1  "am 1.0.0" 
 .SH NAME
 am \- agent\-manager (am) — manage local AI coding agents
 .SH SYNOPSIS
@@ -157,6 +157,9 @@ Real\-time monitor of the managed processes (issue #50)
 am\-web(1)
 Local read\-only web dashboard with charts (issue #54)
 .TP
+am\-serve(1)
+Authenticated HTTP + WebSocket API to drive am remotely (issue #80)
+.TP
 am\-sync(1)
 Push the state into the configured git repository (issue #66)
 .TP
@@ -241,4 +244,4 @@ Export the configuration and installation state (backup)
 am\-import(1)
 Import a previously exported configuration and state
 .SH VERSION
-v0.7.0
+v1.0.0

src/cli.rs

@@ -291,6 +291,8 @@ pub enum Command {
     },
     /// Local read-only web dashboard with charts (issue #54)
     Web(WebArgs),
+    /// Authenticated HTTP + WebSocket API to drive am remotely (issue #80)
+    Serve(ServeArgs),
     /// Push the state into the configured git repository (issue #66)
     Sync {
         /// Commit message (default: "am sync — state update")
@@ -698,6 +700,23 @@ pub struct WebArgs {
     pub no_open: bool,
 }
 
+/// Arguments of am serve (issue #80).
+#[derive(Args, Debug, Clone, Default)]
+pub struct ServeArgs {
+    /// Bearer token required on every request (mandatory)
+    #[arg(long, value_name = "TOKEN")]
+    pub token: Option<String>,
+    /// Listen address (default 127.0.0.1)
+    #[arg(long, value_name = "HOST")]
+    pub host: Option<String>,
+    /// Listening port (default 8080)
+    #[arg(long, value_name = "PORT")]
+    pub port: Option<u16>,
+    /// Per-IP request budget per minute (default 120)
+    #[arg(long, value_name = "N")]
+    pub rate_limit: Option<u32>,
+}
+
 /// Remote catalog subcommands (issues #60 #67).
 #[derive(Subcommand, Debug, Clone)]
 pub enum CatalogCmd {

src/commands/mod.rs

@@ -30,10 +30,11 @@ pub mod registry_cmd;
 pub mod run_cmd;
 pub mod schedule_cmd;
 pub mod search_cmd;
-pub mod service_cmd;
 pub mod secret_cmd;
 pub mod self_uninstall;
 pub mod self_update;
+pub mod serve_cmd;
+pub mod service_cmd;
 pub mod sessions_cmd;
 pub mod stats_cmd;
 pub mod status_cmd;
@@ -124,6 +125,7 @@ pub fn execute_command(app: &App, cmd: &Command) -> Result<i32> {
             monitor_cmd::run(app, *interval, *json)
         }
         Command::Web(args) => web_cmd::run(app, args.port, args.no_open),
+        Command::Serve(args) => serve_cmd::run(app, args),
         Command::Sync { message } => sync_cmd::run(app, message.as_deref()),
         Command::Migrate { export, output, bundle } => {
             migrate_cmd::run(app, *export, output.as_deref(), bundle.as_deref())

src/commands/serve_cmd.rs

@@ -0,0 +1,17 @@
+//! am serve — authenticated HTTP + WebSocket API (issue #80).
+
+use super::*;
+use crate::cli::ServeArgs;
+
+pub fn run(app: &App, args: &ServeArgs) -> Result<i32> {
+    let Some(token) = args.token.clone() else {
+        anyhow::bail!("am serve exige --token <token> — le serveur refuse toute requête sans ce token");
+    };
+    crate::serve::run(
+        app,
+        &token,
+        args.host.as_deref().unwrap_or("127.0.0.1"),
+        args.port.unwrap_or(crate::serve::DEFAULT_PORT),
+        args.rate_limit.unwrap_or(crate::serve::DEFAULT_RATE_LIMIT),
+    )
+}

src/help.rs

@@ -902,6 +902,32 @@ pub static HELP_SPECS: &[HelpSpec] = &[
             HelpExample { desc: "Installer un catalogue en faisant confiance à la source.", code: "registry install https://git.dracodev.net/bruno/am-catalog/raw/branch/main/am-catalog.yaml --yes" },
         ],
     },
+    HelpSpec {
+        name: "serve",
+        category: "Commands",
+        usage: "serve --token <TOKEN> [--host HOST] [--port PORT] [--rate-limit N]",
+        about: "API HTTP + WebSocket authentifiée pour piloter am à distance (issue #80) : token obligatoire sur chaque requête, contrats --json réutilisés (stats, sessions, ps, run, start, stop, ask), WebSocket diffusant les événements du journal en temps réel, rate limiting par IP. TLS : placez le serveur derrière un reverse proxy (caddy/nginx).",
+        search_terms: &["api", "http", "websocket", "token", "remote", "serveur"],
+        flags: &[
+            HelpFlag { short: "", long: "--token", value: "TOKEN", desc: "Bearer token requis sur chaque requête (obligatoire)" },
+            HelpFlag { short: "", long: "--host", value: "HOST", desc: "Adresse d'écoute (défaut 127.0.0.1)" },
+            HelpFlag { short: "", long: "--port", value: "PORT", desc: "Port d'écoute (défaut 8080)" },
+            HelpFlag { short: "", long: "--rate-limit", value: "N", desc: "Budget de requêtes par minute et par IP (défaut 120)" },
+        ],
+        subcommands: &[
+            ("/api/health", "GET — {ok, version}"),
+            ("/api/stats", "GET — am stats --json"),
+            ("/api/run", "POST {agent, model?, provider?, args?} — am run --json"),
+            ("/ws", "GET — WebSocket : événements du journal en temps réel"),
+        ],
+        parameters: &[],
+        io: None,
+        examples: &[
+            HelpExample { desc: "Démarrer le serveur.", code: "serve --token MonTokenSecret --port 8080" },
+            HelpExample { desc: "Appeler une route (depuis une autre machine).", code: "curl -H \"Authorization: Bearer MonTokenSecret\" http://serveur:8080/api/stats" },
+            HelpExample { desc: "Lancer un agent à distance.", code: "curl -X POST -H \"Authorization: Bearer MonTokenSecret\" -d '{\"agent\":\"claude-code\"}' http://serveur:8080/api/run" },
+        ],
+    },
     HelpSpec {
         name: "start",
         category: "Commands",
@@ -2239,7 +2265,14 @@ mod tests {
         assert!(text.contains("build_target"));
         assert!(text.contains("allocator"));
         assert_eq!(info.version, env!("CARGO_PKG_VERSION"));
-        assert_eq!(info.major, 0);
+        // The major/minor version follows Cargo.toml: parse instead of pinning.
+        let major: u64 = env!("CARGO_PKG_VERSION")
+            .split('.')
+            .next()
+            .unwrap()
+            .parse()
+            .unwrap();
+        assert_eq!(info.major, major);
         // The minor version follows Cargo.toml: parse it instead of pinning it.
         let minor: u64 = env!("CARGO_PKG_VERSION")
             .split('.')

src/lib.rs

@@ -48,6 +48,7 @@ pub mod repl;
 pub mod runner;
 pub mod sandbox;
 pub mod secrets;
+pub mod serve;
 pub mod sessions;
 pub mod shell;
 pub mod state;

src/repl.rs

@@ -122,6 +122,7 @@ const COMMAND_DESCRIPTIONS: &[(&str, &str)] = &[
     ("playbook", "replay a saved history sequence step by step (am history --save)"),
     ("lab", "benchmark the same task on several agents (duration, cost, --json)"),
     ("web", "local read-only web dashboard with charts (issue #54)"),
+    ("serve", "authenticated HTTP + WebSocket API to drive am remotely (issue #80)"),
     ("plugins", "list the event plugins and test one (JSON contract on stdin/stdout)"),
     ("shell", "show or switch the system shell"),
     ("theme", "show or switch the color theme"),
@@ -238,7 +239,7 @@ impl AmCompleter {
                 "self-update", "self-uninstall", "export", "import", "shell", "theme", "lang",
                 "tip", "dashboard", "favorite", "unfavorite", "note", "tag", "untag", "tags",
                 "profile", "man", "models", "catalog", "providers", "suggest", "ask", "registry", "audit",
-                "service", "schedule", "monitor", "web", "sync", "migrate", "playbook", "lab", "plugins",
+                "service", "schedule", "monitor", "web", "serve", "sync", "migrate", "playbook", "lab", "plugins",
                 "ls", "dir", "cd", "ps", "where", "get", "help", "version", "exit",
             ],
             config_sub: vec!["show", "path", "edit", "validate", "add"],
@@ -780,6 +781,7 @@ pub fn banner_box(
     }));
     rows.push(inner("  catalog    list · status · search · info · init".to_string()));
     rows.push(inner("  registry   publish · search · install · list".to_string()));
+    rows.push(inner("  serve      API HTTP+WS · token · stats · run · start · stop · ask".to_string()));
     rows.push(inner(
         "  activity   sessions · stats · top · report · projects · timeline · log · logs · history"
             .to_string(),
@@ -1677,6 +1679,12 @@ fn handle_line(
             port: opt_value("--port").and_then(|v| v.parse().ok()),
             no_open: flag("--no-open"),
         }),
+        "serve" => Command::Serve(crate::cli::ServeArgs {
+            token: opt_value("--token"),
+            host: opt_value("--host"),
+            port: opt_value("--port").and_then(|v| v.parse().ok()),
+            rate_limit: opt_value("--rate-limit").and_then(|v| v.parse().ok()),
+        }),
         "sync" => Command::Sync {
             message: opt_value("--message"),
         },

src/serve.rs

@@ -0,0 +1,626 @@
+//! am serve (issue #80): authenticated HTTP + WebSocket API to drive am
+//! remotely. Zero new runtime dependency (tiny_http is already there).
+//!
+//! - Every route requires the bearer token (Authorization: Bearer <token>,
+//!   or ?token= for WebSocket clients which cannot set headers).
+//! - Commands reuse the existing `--json` contracts by re-spawning the am
+//!   binary (the same pattern as the web dashboard, issue #54): stats,
+//!   sessions, ps, run, start, stop, ask.
+//! - The WebSocket broadcasts journal events (journal, sessions, alerts)
+//!   in real time by polling the events journal.
+//! - Rate limiting: a sliding window per client IP.
+//! - TLS is delegated to a reverse proxy (caddy/nginx) — documented.
+
+use crate::app::App;
+use anyhow::{anyhow, Context, Result};
+use base64::Engine;
+use sha2::Digest;
+use std::collections::HashMap;
+use std::io::{Read, Write};
+use std::path::PathBuf;
+use std::sync::mpsc::Sender;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+/// Default listening port.
+pub const DEFAULT_PORT: u16 = 8080;
+/// Default per-IP request budget per minute.
+pub const DEFAULT_RATE_LIMIT: u32 = 120;
+/// WebSocket GUID mandated by RFC 6455.
+const WS_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+pub struct ServeState {
+    pub token: String,
+    pub rate_limit_per_min: u32,
+    hits: Mutex<HashMap<String, Vec<Instant>>>,
+    pub clients: Mutex<Vec<(usize, Sender<String>)>>,
+    pub events_offsets: Mutex<HashMap<PathBuf, u64>>,
+    next_client_id: std::sync::atomic::AtomicUsize,
+}
+
+impl ServeState {
+    pub fn new(token: &str, rate_limit_per_min: u32) -> Self {
+        ServeState {
+            token: token.to_string(),
+            rate_limit_per_min,
+            hits: Mutex::new(HashMap::new()),
+            clients: Mutex::new(Vec::new()),
+            events_offsets: Mutex::new(HashMap::new()),
+            next_client_id: std::sync::atomic::AtomicUsize::new(1),
+        }
+    }
+
+    /// Constant-time-ish token check (no early length shortcut).
+    pub fn token_ok(&self, candidate: &str) -> bool {
+        let a = self.token.as_bytes();
+        let b = candidate.as_bytes();
+        if a.len() != b.len() {
+            return false;
+        }
+        a.iter()
+            .zip(b.iter())
+            .fold(0u8, |acc, (x, y)| acc | (x ^ y))
+            == 0
+    }
+
+    /// Sliding window rate limit: returns true when the request is allowed.
+    pub fn allow(&self, ip: &str) -> bool {
+        let now = Instant::now();
+        let mut hits = self.hits.lock().unwrap();
+        let window = hits.entry(ip.to_string()).or_default();
+        window.retain(|t| now.duration_since(*t) < Duration::from_secs(60));
+        if window.len() >= self.rate_limit_per_min as usize {
+            return false;
+        }
+        window.push(now);
+        true
+    }
+}
+
+/// Execute an am command by re-spawning the current binary (isolated App,
+/// same guarantees as the CLI). Returns stdout when the exit code is 0.
+pub fn run_am(config: &Option<PathBuf>, args: &[&str]) -> Result<(i32, String)> {
+    let exe = std::env::current_exe().context("cannot locate the am binary")?;
+    let mut cmd = std::process::Command::new(&exe);
+    if let Some(cfg) = config {
+        cmd.arg("--config").arg(cfg);
+    }
+    cmd.arg("--no-color");
+    cmd.args(args);
+    let out = cmd
+        .output()
+        .with_context(|| format!("failed to run {}", exe.display()))?;
+    let text = String::from_utf8_lossy(&out.stdout).to_string();
+    Ok((out.status.code().unwrap_or(1), text))
+}
+
+/// Reusable test seam: the handler calls this for every command route.
+pub type Runner = dyn Fn(&[&str]) -> Result<(i32, String)> + Send + Sync;
+
+/// Serve until the process is killed. Every request is handled on its own
+/// thread; the broadcast loop pushes journal events to WebSocket clients.
+pub fn run(app: &App, token: &str, host: &str, port: u16, rate_limit: u32) -> Result<i32> {
+    let server = tiny_http::Server::http((host, port))
+        .map_err(|e| anyhow!("cannot listen on {host}:{port}: {e}"))?;
+    let state = Arc::new(ServeState::new(token, rate_limit));
+    let config = Arc::new(app.cli.config.clone());
+
+    // Broadcast loop: poll the journal, push new events to WS clients.
+    {
+        let state = state.clone();
+        let events_dir = app.events_dir();
+        std::thread::spawn(move || loop {
+            broadcast_events(&state, &events_dir);
+            std::thread::sleep(Duration::from_millis(500));
+        });
+    }
+
+    app.log.info(&format!(
+        "am serve sur http://{host}:{port} (token requis, {} req/min/IP) — Ctrl+C pour arrêter",
+        rate_limit
+    ));
+    let runner: Arc<Runner> = Arc::new({
+        let config = config.clone();
+        Box::new(move |args: &[&str]| run_am(config.as_ref(), args))
+    });
+    for request in server.incoming_requests() {
+        let state = state.clone();
+        let config = config.clone();
+        let runner = runner.clone();
+        std::thread::spawn(move || {
+            let _ = handle(request, &state, config.as_ref(), runner.as_ref());
+        });
+    }
+    Ok(0)
+}
+
+/// Route table. `runner` is the command executor (re-spawn in production,
+/// a mock in tests).
+pub fn handle(
+    mut request: tiny_http::Request,
+    state: &ServeState,
+    config: &Option<PathBuf>,
+    runner: &Runner,
+) -> Result<()> {
+    let ip = request
+        .remote_addr()
+        .map(|a| a.ip().to_string())
+        .unwrap_or_else(|| "unknown".to_string());
+    if !state.allow(&ip) {
+        return json(request, 429, r#"{"error":"rate limit exceeded"}"#);
+    }
+    let url = request.url().to_string();
+    let (path, query) = match url.split_once('?') {
+        Some((p, q)) => (p.to_string(), q.to_string()),
+        None => (url, String::new()),
+    };
+
+    // Authentication: Authorization: Bearer <token>, or ?token= (WebSocket
+    // clients and simple clients).
+    let auth_ok = header_token(&request)
+        .or_else(|| query_token(&query))
+        .map(|t| state.token_ok(&t))
+        .unwrap_or(false);
+    if !auth_ok {
+        return json(request, 401, r#"{"error":"token required"}"#);
+    }
+
+    match (request.method(), path.as_str()) {
+        (tiny_http::Method::Get, "/api/health") => json(
+            request,
+            200,
+            &format!(
+                r#"{{"ok":true,"version":"{}","time":"{}"}}"#,
+                env!("CARGO_PKG_VERSION"),
+                crate::installers::now_rfc3339()
+            ),
+        ),
+        (tiny_http::Method::Get, "/api/stats") => {
+            command(request, config, runner, &["stats", "--json"])
+        }
+        (tiny_http::Method::Get, "/api/sessions") => {
+            command(request, config, runner, &["sessions", "--json"])
+        }
+        (tiny_http::Method::Get, "/api/ps") => command(request, config, runner, &["ps", "--json"]),
+        (tiny_http::Method::Get, "/api/providers") => {
+            command(request, config, runner, &["providers", "list", "--json"])
+        }
+        (tiny_http::Method::Post, "/api/run") => {
+            let body = read_body(&mut request);
+            match serde_json::from_str::<serde_json::Value>(&body) {
+                Ok(v) => {
+                    let mut args = vec![
+                        "run".to_string(),
+                        v.get("agent")
+                            .and_then(|a| a.as_str())
+                            .unwrap_or_default()
+                            .to_string(),
+                    ];
+                    if let Some(m) = v.get("model").and_then(|m| m.as_str()) {
+                        args.push("--model".to_string());
+                        args.push(m.to_string());
+                    }
+                    if let Some(p) = v.get("provider").and_then(|p| p.as_str()) {
+                        args.push("--provider".to_string());
+                        args.push(p.to_string());
+                    }
+                    if let Some(extra) = v.get("args").and_then(|a| a.as_array()) {
+                        for a in extra {
+                            if let Some(s) = a.as_str() {
+                                args.push(s.to_string());
+                            }
+                        }
+                    }
+                    args.push("--json".to_string());
+                    let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+                    command(request, config, runner, &refs)
+                }
+                Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#),
+            }
+        }
+        (tiny_http::Method::Post, "/api/start") => {
+            let body = read_body(&mut request);
+            match serde_json::from_str::<serde_json::Value>(&body) {
+                Ok(v) => {
+                    let agent = v.get("agent").and_then(|a| a.as_str()).unwrap_or_default();
+                    command(
+                        request,
+                        config,
+                        runner,
+                        &["start", agent, "--background", "--json"],
+                    )
+                }
+                Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#),
+            }
+        }
+        (tiny_http::Method::Post, "/api/stop") => {
+            let body = read_body(&mut request);
+            match serde_json::from_str::<serde_json::Value>(&body) {
+                Ok(v) => {
+                    let agent = v.get("agent").and_then(|a| a.as_str()).unwrap_or_default();
+                    command(request, config, runner, &["stop", agent, "--json"])
+                }
+                Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#),
+            }
+        }
+        (tiny_http::Method::Post, "/api/ask") => {
+            let body = read_body(&mut request);
+            match serde_json::from_str::<serde_json::Value>(&body) {
+                Ok(v) => {
+                    let q = v.get("query").and_then(|a| a.as_str()).unwrap_or_default();
+                    command(request, config, runner, &["ask", q, "--yes"])
+                }
+                Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#),
+            }
+        }
+        (tiny_http::Method::Get, "/ws") => websocket(request, state),
+        _ => json(request, 404, r#"{"error":"not found"}"#),
+    }
+}
+
+fn header_token(request: &tiny_http::Request) -> Option<String> {
+    request
+        .headers()
+        .iter()
+        .find(|h| h.field.equiv("Authorization"))
+        .and_then(|h| h.value.as_str().strip_prefix("Bearer "))
+        .map(|t| t.trim().to_string())
+}
+
+fn query_token(query: &str) -> Option<String> {
+    query
+        .split('&')
+        .find_map(|kv| kv.strip_prefix("token="))
+        .map(|t| t.to_string())
+}
+
+fn command(
+    request: tiny_http::Request,
+    config: &Option<PathBuf>,
+    runner: &Runner,
+    args: &[&str],
+) -> Result<()> {
+    match runner(args) {
+        Ok((0, out)) => json(request, 200, &out),
+        Ok((code, out)) => json(
+            request,
+            422,
+            &format!(
+                r#"{{"error":"command failed ({code})","output":{}}}"#,
+                serde_json::to_string(&out).unwrap_or_else(|_| "\"\"".into())
+            ),
+        ),
+        Err(e) => json(
+            request,
+            500,
+            &format!(
+                r#"{{"error":{}}}"#,
+                serde_json::to_string(&format!("{e:#}")).unwrap_or_else(|_| "\"internal\"".into())
+            ),
+        ),
+    }
+}
+
+fn json(request: tiny_http::Request, status: u32, body: &str) -> Result<()> {
+    let response = tiny_http::Response::from_string(body.to_string())
+        .with_status_code(status)
+        .with_header(
+            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap(),
+        );
+    request.respond(response)?;
+    Ok(())
+}
+
+fn read_body(request: &mut tiny_http::Request) -> String {
+    let mut body = String::new();
+    let _ = request
+        .as_reader()
+        .take(1_000_000)
+        .read_to_string(&mut body);
+    body
+}
+
+// ---------------------------------------------------------------------------
+// WebSocket (RFC 6455): handshake + text frames + journal broadcast.
+// ---------------------------------------------------------------------------
+
+/// Serve the /ws endpoint: upgrade the request, register the client and push
+/// broadcast frames until the client disconnects.
+pub fn websocket(request: tiny_http::Request, state: &ServeState) -> Result<()> {
+    let key = request
+        .headers()
+        .iter()
+        .find(|h| h.field.equiv("Sec-WebSocket-Key"))
+        .map(|h| h.value.as_str().to_string())
+        .ok_or_else(|| anyhow!("missing Sec-WebSocket-Key"))?;
+    let accept = ws_accept(&key);
+
+    let response = tiny_http::Response::empty(101).with_header(
+        tiny_http::Header::from_bytes(&b"Sec-WebSocket-Accept"[..], accept.as_bytes()).unwrap(),
+    );
+    let mut writer = request.upgrade("websocket", response);
+    let (tx, rx) = std::sync::mpsc::channel::<String>();
+    let client_id = state
+        .next_client_id
+        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    state.clients.lock().unwrap().push((client_id, tx));
+
+    // Write loop: frames come from the broadcast channel; a failed write
+    // means the client is gone. (Pings/close frames are not answered in
+    // this version — the write failure on the closed socket ends the loop.)
+    while let Ok(frame) = rx.recv_timeout(Duration::from_millis(500)) {
+        if write_frame(writer.as_mut(), 1, frame.as_bytes()).is_err() {
+            break;
+        }
+    }
+    // The client is gone: remove our sender from the broadcast list.
+    drop(rx);
+    state
+        .clients
+        .lock()
+        .unwrap()
+        .retain(|(id, _)| *id != client_id);
+    Ok(())
+}
+
+fn ws_accept(key: &str) -> String {
+    let mut hasher = sha1::Sha1::new();
+    hasher.update(key.as_bytes());
+    hasher.update(WS_GUID.as_bytes());
+    base64::engine::general_purpose::STANDARD.encode(hasher.finalize())
+}
+
+/// Write a single server frame (unmasked): opcode + 7-bit length + payload.
+fn write_frame<W: Write + ?Sized>(w: &mut W, opcode: u8, payload: &[u8]) -> Result<()> {
+    w.write_all(&[0x80 | opcode, payload.len() as u8])?;
+    w.write_all(payload)?;
+    w.flush()?;
+    Ok(())
+}
+
+/// Broadcast the new journal events to every WebSocket client.
+pub fn broadcast_events(state: &ServeState, events_dir: &std::path::Path) {
+    let files = crate::events::journal_files(events_dir);
+    let Some(latest) = files.last() else {
+        return;
+    };
+    let Ok(mut file) = std::fs::File::open(latest) else {
+        return;
+    };
+    let mut offsets = state.events_offsets.lock().unwrap();
+    let known = offsets.contains_key(latest);
+    let start = offsets.get(latest).copied().unwrap_or(0);
+    use std::io::Seek;
+    if file.seek(std::io::SeekFrom::Start(start)).is_err() {
+        return;
+    }
+    let mut tail = String::new();
+    if file.read_to_string(&mut tail).is_err() {
+        return;
+    }
+    let new_len = start + tail.len() as u64;
+    if tail.is_empty() {
+        return;
+    }
+    offsets.insert(latest.clone(), new_len);
+    drop(offsets);
+    // First pass only indexes the journal: nothing is broadcast until the
+    // next poll sees new lines.
+    if !known {
+        return;
+    }
+    let mut clients = state.clients.lock().unwrap();
+    for line in tail.lines() {
+        let frame = format!("{line}\n");
+        for (_, client) in clients.iter() {
+            let _ = client.send(frame.clone());
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use anyhow::Context;
+
+    fn test_state() -> ServeState {
+        ServeState::new("s3cr3t", 3)
+    }
+
+    fn mock_runner(captured: &Arc<Mutex<Vec<String>>>) -> Box<Runner> {
+        let captured = captured.clone();
+        Box::new(move |args: &[&str]| {
+            captured.lock().unwrap().push(args.join(" "));
+            Ok((0, r#"{"mocked":true}"#.to_string()))
+        })
+    }
+
+    /// Drive a real tiny_http request through handle() and return the body.
+    fn hit(port: u16, method: &str, path: &str, token: Option<&str>) -> (u32, String) {
+        let url = format!("http://127.0.0.1:{port}{path}");
+        let client = ureq::AgentBuilder::new()
+            .timeout(Duration::from_secs(5))
+            .build();
+        let mut req = client.request(method, &url);
+        if let Some(t) = token {
+            req = req.set("Authorization", &format!("Bearer {t}"));
+        }
+        let res = req.call();
+        match res {
+            Ok(r) => (r.status() as u32, r.into_string().unwrap_or_default()),
+            Err(ureq::Error::Status(code, r)) => (code as u32, r.into_string().unwrap_or_default()),
+            Err(_) => (0, String::new()),
+        }
+    }
+
+    #[test]
+    fn without_token_is_401() {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let port = server.server_addr().to_ip().unwrap().port();
+        let handle_thread = std::thread::spawn(move || {
+            let req = server
+                .recv_timeout(Duration::from_secs(5))
+                .unwrap()
+                .unwrap();
+            let state = test_state();
+            let captured = Arc::new(Mutex::new(Vec::new()));
+            let runner = mock_runner(&captured);
+            handle(req, &state, &None, runner.as_ref()).unwrap();
+        });
+        let (status, body) = hit(port, "GET", "/api/health", None);
+        handle_thread.join().unwrap();
+        assert_eq!(status, 401);
+        assert!(body.contains("token required"), "{body}");
+    }
+
+    #[test]
+    fn wrong_token_is_401_and_right_token_health_ok() {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let port = server.server_addr().to_ip().unwrap().port();
+        let handle_thread = std::thread::spawn(move || {
+            for _ in 0..2 {
+                let req = server
+                    .recv_timeout(Duration::from_secs(5))
+                    .unwrap()
+                    .unwrap();
+                let state = test_state();
+                let captured = Arc::new(Mutex::new(Vec::new()));
+                let runner = mock_runner(&captured);
+                handle(req, &state, &None, runner.as_ref()).unwrap();
+            }
+        });
+        let (s1, _) = hit(port, "GET", "/api/health", Some("wrong"));
+        let (s2, body) = hit(port, "GET", "/api/health", Some("s3cr3t"));
+        handle_thread.join().unwrap();
+        assert_eq!(s1, 401);
+        assert_eq!(s2, 200);
+        assert!(body.contains("\"ok\":true"), "{body}");
+    }
+
+    #[test]
+    fn stats_route_invokes_the_runner_with_json() {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let port = server.server_addr().to_ip().unwrap().port();
+        let captured = Arc::new(Mutex::new(Vec::new()));
+        let captured2 = captured.clone();
+        let handle_thread = std::thread::spawn(move || {
+            let req = server
+                .recv_timeout(Duration::from_secs(5))
+                .unwrap()
+                .unwrap();
+            let state = test_state();
+            let runner = mock_runner(&captured2);
+            handle(req, &state, &None, runner.as_ref()).unwrap();
+        });
+        let (status, body) = hit(port, "GET", "/api/stats", Some("s3cr3t"));
+        handle_thread.join().unwrap();
+        assert_eq!(status, 200);
+        assert!(body.contains("mocked"), "{body}");
+        assert_eq!(captured.lock().unwrap()[0], "stats --json");
+    }
+
+    #[test]
+    fn run_route_passes_agent_model_and_args() {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let port = server.server_addr().to_ip().unwrap().port();
+        let captured = Arc::new(Mutex::new(Vec::new()));
+        let captured2 = captured.clone();
+        let handle_thread = std::thread::spawn(move || {
+            let req = server
+                .recv_timeout(Duration::from_secs(5))
+                .unwrap()
+                .unwrap();
+            let state = test_state();
+            let runner = mock_runner(&captured2);
+            handle(req, &state, &None, runner.as_ref()).unwrap();
+        });
+        let body = r#"{"agent":"claude-code","model":"claude-sonnet-4-5","args":["--help"]}"#;
+        let client = ureq::AgentBuilder::new()
+            .timeout(Duration::from_secs(5))
+            .build();
+        let res = client
+            .post(&format!("http://127.0.0.1:{port}/api/run"))
+            .set("Authorization", "Bearer s3cr3t")
+            .send_string(body)
+            .unwrap();
+        let status = res.status();
+        handle_thread.join().unwrap();
+        assert_eq!(status, 200);
+        let called = captured.lock().unwrap()[0].clone();
+        assert!(called.contains("run claude-code"), "{called}");
+        assert!(called.contains("--model claude-sonnet-4-5"), "{called}");
+        assert!(called.contains("--help"), "{called}");
+        assert!(called.ends_with("--json"), "{called}");
+    }
+
+    #[test]
+    fn unknown_route_is_404() {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let port = server.server_addr().to_ip().unwrap().port();
+        let handle_thread = std::thread::spawn(move || {
+            let req = server
+                .recv_timeout(Duration::from_secs(5))
+                .unwrap()
+                .unwrap();
+            let state = test_state();
+            let captured = Arc::new(Mutex::new(Vec::new()));
+            let runner = mock_runner(&captured);
+            handle(req, &state, &None, runner.as_ref()).unwrap();
+        });
+        let (status, body) = hit(port, "GET", "/api/nope", Some("s3cr3t"));
+        handle_thread.join().unwrap();
+        assert_eq!(status, 404);
+        assert!(body.contains("not found"), "{body}");
+    }
+
+    #[test]
+    fn rate_limit_blocks_after_the_budget() {
+        let state = test_state(); // budget = 3/min
+        assert!(state.allow("10.0.0.1"));
+        assert!(state.allow("10.0.0.1"));
+        assert!(state.allow("10.0.0.1"));
+        assert!(!state.allow("10.0.0.1"), "4th hit must be blocked");
+        // A different IP is unaffected.
+        assert!(state.allow("10.0.0.2"));
+    }
+
+    #[test]
+    fn ws_handshake_accept_is_rfc6455() {
+        // Known RFC 6455 example: key "dGhlIHNhbXBsZSBub25jZQ=="
+        // -> accept "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
+        assert_eq!(
+            ws_accept("dGhlIHNhbXBsZSBub25jZQ=="),
+            "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
+        );
+    }
+
+    #[test]
+    fn broadcast_pushes_new_journal_lines() {
+        let dir = tempfile::tempdir().unwrap();
+        let state = test_state();
+        let (tx, rx) = std::sync::mpsc::channel::<String>();
+        state.clients.lock().unwrap().push((1, tx));
+        let file = dir.path().join("events-202608.jsonl");
+        std::fs::write(&file, "{\"kind\":\"start\"}\n").unwrap();
+        // First pass indexes the existing line (no broadcast).
+        broadcast_events(&state, dir.path());
+        // New line -> broadcast.
+        let mut f = std::fs::OpenOptions::new()
+            .append(true)
+            .open(&file)
+            .unwrap();
+        writeln!(f, "{{\"kind\":\"run\"}}").unwrap();
+        broadcast_events(&state, dir.path());
+        let msg = rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        assert!(msg.contains("\"kind\":\"run\""), "{msg}");
+    }
+
+    #[test]
+    fn token_compare_is_exact() {
+        let state = test_state();
+        assert!(state.token_ok("s3cr3t"));
+        assert!(!state.token_ok("s3cr3tX"));
+        assert!(!state.token_ok(""));
+        assert!(!state.token_ok("S3CR3T"));
+    }
+}