web P3a: log windows — attach/detach/history + SSE log events

POST /api/logs attaches a streamed 'docker logs -f --tail=200' per
(cluster, node, recipe, rank) over a no-inactivity-timeout SSH session;
DELETE /api/logs detaches (stop flag + task abort); GET /api/logs returns
ring history for a fresh window. Idempotent per key (live stream reused,
finished stream replaced); max 8 attaches (409); 1000-line ring per
attach.

- src/web/ssh_stream.rs: stream_lines (russh channel.wait() data chunks)
  + split_lines — a chunk without a trailing newline is buffered until it
  completes (pure, unit-tested: partials, CRLF, blanks)
- src/web/logs.rs: LogAttach registry + push_line ring trim
- src/web/events.rs: Event::Log; Snapshot.attaches (AttachSnap) so the UI
  restores open log windows after reload (build_snapshot takes the view)
- API round-trip test boots the real server (attach/idempotent/GET/400/
  DELETE 204/404); registry tests (cap, dedup, reuse, replace, trim)
- Live-verified on the real fleet: attach to the running qwen38 serve on
  dgx1 -> 200 history lines + live 'type:log' SSE events, detach 204

src/web/api.rs

@@ -199,3 +199,95 @@ pub(crate) async fn recipes(app: State<AppState>) -> Json<serde_json::Value> {
     }
     Json(serde_json::json!({ "recipes": out }))
 }
+
+// ---------------------------------------------------------------------------
+// P3: log windows — attach a streaming `docker logs -f` per node+recipe+rank
+// (idempotent), detach, and fetch ring history for a fresh window.
+
+#[derive(Deserialize)]
+pub(crate) struct LogsReq {
+    pub cluster: String,
+    pub node: String,
+    pub recipe: String,
+    #[serde(default)]
+    pub rank: u32,
+    #[serde(default)]
+    pub lines: usize,
+}
+
+/// `POST /api/logs` — attach (reuses a live stream for the same key).
+pub(crate) async fn logs_attach(
+    app: State<AppState>,
+    Json(req): Json<LogsReq>,
+) -> Result<Json<serde_json::Value>, ApiError> {
+    let (id, history) = super::logs::attach(
+        &app.config,
+        &app.bus,
+        &app.log_attaches,
+        &req.cluster,
+        &req.node,
+        &req.recipe,
+        req.rank,
+    )
+    .map_err(|e| {
+        (
+            if e.starts_with("max") {
+                StatusCode::CONFLICT
+            } else {
+                StatusCode::BAD_REQUEST
+            },
+            e,
+        )
+    })?;
+    Ok(Json(
+        serde_json::json!({ "attach_id": id, "history": history }),
+    ))
+}
+
+/// `DELETE /api/logs` — stop the stream and drop the attach.
+#[derive(Deserialize)]
+pub(crate) struct DetachReq {
+    pub attach_id: String,
+}
+
+pub(crate) async fn logs_detach(
+    app: State<AppState>,
+    Json(req): Json<DetachReq>,
+) -> Result<StatusCode, ApiError> {
+    let id = req.attach_id;
+    if super::logs::detach(&app.log_attaches, &id) {
+        Ok(StatusCode::NO_CONTENT)
+    } else {
+        Err((StatusCode::NOT_FOUND, format!("no log attach '{id}'")))
+    }
+}
+
+/// `GET /api/logs?cluster=&node=&recipe=&rank=&lines=` — ring history for a
+/// fresh window (0 lines = `logs::TAIL_LINES`).
+pub(crate) async fn logs_history(
+    app: State<AppState>,
+    axum::extract::Query(q): axum::extract::Query<LogsReq>,
+) -> Result<Json<serde_json::Value>, ApiError> {
+    let lines = if q.lines == 0 {
+        super::logs::TAIL_LINES
+    } else {
+        q.lines
+    };
+    let (id, history) = super::logs::by_key(
+        &app.log_attaches,
+        &q.cluster,
+        &q.node,
+        &q.recipe,
+        q.rank,
+        lines,
+    )
+    .ok_or_else(|| {
+        (
+            StatusCode::NOT_FOUND,
+            "no log attach for that node/recipe".to_string(),
+        )
+    })?;
+    Ok(Json(
+        serde_json::json!({ "attach_id": id, "history": history }),
+    ))
+}

src/web/events.rs

@@ -1,8 +1,8 @@
 //! SSE event bus + state snapshot for the dashboard.
 //!
 //! One multiplexed `broadcast` channel; `/events` forwards every event and
-//! the UI routes by `type`. P2 emits `state` (supervisor + job finishes)
-//! and `job` (registry transitions).
+//! the UI routes by `type`. Emits `state` (supervisor + job finishes),
+//! `job` (registry transitions), and `log` (attached window lines).
 
 use crate::config::Config;
 use crate::web::probe::ContainerSnap;
@@ -26,6 +26,15 @@ pub enum Event {
         recipe: Option<String>,
         message: String,
     },
+    Log {
+        /// Attach id (matches `POST /api/logs` response).
+        attach: String,
+        cluster: String,
+        node: String,
+        recipe: String,
+        rank: u32,
+        line: String,
+    },
 }
 
 /// Per-node view for the LED grid.
@@ -53,12 +62,26 @@ pub struct ClusterSnap {
     pub free: u32,
 }
 
+/// One log window attach (for state snapshots — the UI restores open
+/// log windows from `/api/state` after a reload).
+#[derive(Serialize, Clone, Debug)]
+pub struct AttachSnap {
+    pub id: String,
+    pub cluster: String,
+    pub node: String,
+    pub recipe: String,
+    pub rank: u32,
+    /// Stream ended on its own (container stopped, SSH drop).
+    pub done: bool,
+}
+
 /// The full dashboard snapshot (also the `GET /api/state` body).
 #[derive(Serialize, Clone, Debug)]
 pub struct Snapshot {
     pub ts: u64,
     pub clusters: Vec<ClusterSnap>,
     pub jobs: Vec<crate::web::jobs::Job>,
+    pub attaches: Vec<AttachSnap>,
 }
 
 fn now_ts() -> u64 {
@@ -77,6 +100,7 @@ pub fn build_snapshot(
     config: &Config,
     fleet: &crate::web::probe::Fleet,
     jobs: &[crate::web::jobs::Job],
+    attaches: &[AttachSnap],
 ) -> Snapshot {
     let nodes = &fleet.nodes;
     let switch_off = &fleet.switch_off;
@@ -125,6 +149,7 @@ pub fn build_snapshot(
         ts: now_ts(),
         clusters,
         jobs: jobs.to_vec(),
+        attaches: attaches.to_vec(),
     }
 }
 
@@ -220,7 +245,7 @@ mod tests {
             switch_off: std::collections::HashMap::new(),
             soft_off: std::collections::HashMap::new(),
         };
-        let snap = build_snapshot(&config, &fleet, &[]);
+        let snap = build_snapshot(&config, &fleet, &[], &[]);
         assert_eq!(snap.clusters.len(), 2);
         let a = &snap.clusters[0];
         assert!(a.up);
@@ -262,7 +287,7 @@ mod tests {
             switch_off: std::collections::HashMap::new(),
             soft_off: std::collections::HashMap::new(),
         };
-        let snap = build_snapshot(&config, &fleet, &[]);
+        let snap = build_snapshot(&config, &fleet, &[], &[]);
         let a = &snap.clusters[0];
         assert!(!a.up, "one offline node takes the cluster down");
         assert_eq!(a.free, 0);
@@ -294,7 +319,7 @@ mod tests {
             switch_off: so,
             soft_off: std::collections::HashMap::new(),
         };
-        let snap = build_snapshot(&config, &fleet, &[]);
+        let snap = build_snapshot(&config, &fleet, &[], &[]);
         assert!(!snap.clusters[0].up);
         assert!(snap.clusters[0].switch_off);
         assert_eq!(snap.clusters[0].free, 0);
@@ -325,7 +350,7 @@ mod tests {
             switch_off: HashMap::new(),
             soft_off: soft,
         };
-        let snap = build_snapshot(&config, &fleet, &[]);
+        let snap = build_snapshot(&config, &fleet, &[], &[]);
         assert!(
             snap.clusters[0].up,
             "non-blocking off must not down the cluster"

src/web/logs.rs

@@ -0,0 +1,341 @@
+//! Log window attachments — tracked SSH `docker logs -f` streams with a
+//! ring-buffer history per (cluster, node, recipe, rank). `POST /api/logs`
+//! attaches (idempotent per key), `DELETE /api/logs` detaches, `GET
+//! /api/logs` returns ring history for a fresh window. Lines stream to the
+//! SSE bus as `Event::Log`.
+
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+
+use tokio::task::JoinHandle;
+
+use crate::config::Config;
+use crate::ssh::shq;
+use crate::web::events::Event;
+use crate::web::ssh_stream::stream_lines;
+
+/// Max concurrent log streams (each holds an SSH connection).
+pub const MAX_ATTACHES: usize = 8;
+/// Ring-buffer depth per attach (lines).
+pub const RING_LINES: usize = 1000;
+/// Lines of history a fresh window starts with (docker `--tail`).
+pub const TAIL_LINES: usize = 200;
+
+/// One live (or finished) log stream.
+pub struct LogAttach {
+    pub id: String,
+    pub cluster: String,
+    /// Node name (SSH target).
+    pub node: String,
+    pub recipe: String,
+    pub rank: u32,
+    /// Bounded history for `GET /api/logs` and window re-opens.
+    pub history: Arc<Mutex<VecDeque<String>>>,
+    /// Set on detach; the stream task ends at its next line.
+    pub stop: Arc<AtomicBool>,
+    /// True once the stream ended on its own (container stopped, SSH drop).
+    pub done: Arc<AtomicBool>,
+    /// The streaming task (aborted on detach).
+    pub task: JoinHandle<()>,
+}
+
+type AttachKey = (String, String, String, u32);
+
+impl LogAttach {
+    fn key(&self) -> AttachKey {
+        (
+            self.cluster.clone(),
+            self.node.clone(),
+            self.recipe.clone(),
+            self.rank,
+        )
+    }
+
+    /// Last `n` history lines, oldest first.
+    pub fn history_last(&self, n: usize) -> Vec<String> {
+        let ring = self.history.lock().unwrap();
+        ring.iter()
+            .rev()
+            .take(n)
+            .map(String::clone)
+            .collect::<Vec<_>>()
+            .into_iter()
+            .rev()
+            .collect()
+    }
+}
+
+/// `docker logs -f --tail=N --timestamps` for one recipe+rank's container(s)
+/// on the node — seeds the window with history, then follows.
+pub fn logs_stream_cmd(recipe: &str, rank: u32, tail: usize) -> String {
+    format!(
+        "docker ps -q --filter {} --filter label=sm121.rank={} \
+         | xargs -r docker logs -f --tail={tail} --timestamps 2>&1",
+        shq(&format!("label=sm121.recipe={}", recipe)),
+        rank
+    )
+}
+
+/// Append a line to a ring, evicting the oldest at capacity.
+pub(crate) fn push_line(ring: &Mutex<VecDeque<String>>, line: &str) {
+    let mut r = ring.lock().unwrap();
+    if r.len() >= RING_LINES {
+        r.pop_front();
+    }
+    r.push_back(line.to_string());
+}
+
+/// Idempotent per (cluster, node, recipe, rank): a live stream for the same
+/// key is reused; a finished one is replaced.
+pub fn attach(
+    config: &Config,
+    bus: &tokio::sync::broadcast::Sender<Event>,
+    attaches: &Mutex<Vec<LogAttach>>,
+    cluster: &str,
+    node: &str,
+    recipe: &str,
+    rank: u32,
+) -> Result<(String, Vec<String>), String> {
+    let cl = config
+        .clusters
+        .iter()
+        .find(|c| c.name == cluster)
+        .ok_or_else(|| format!("unknown cluster '{cluster}'"))?;
+    let node_cfg = cl
+        .nodes
+        .iter()
+        .find(|n| n.name == node)
+        .ok_or_else(|| format!("unknown node '{node}' in cluster '{cluster}'"))?;
+    let key: AttachKey = (
+        cluster.to_string(),
+        node.to_string(),
+        recipe.to_string(),
+        rank,
+    );
+
+    let mut list = attaches.lock().unwrap();
+    if let Some(a) = list.iter().find(|a| a.key() == key) {
+        if !a.done.load(Ordering::Relaxed) {
+            return Ok((a.id.clone(), a.history_last(TAIL_LINES)));
+        }
+        // Finished stream — drop it and start fresh below.
+        let pos = list.iter().position(|x| x.id == a.id).unwrap();
+        list.remove(pos);
+    }
+    if list.len() >= MAX_ATTACHES {
+        return Err(format!(
+            "max {MAX_ATTACHES} log attaches — close a window first"
+        ));
+    }
+
+    let id = format!("l{}", list.len() + 1);
+    let history: Arc<Mutex<VecDeque<String>>> =
+        Arc::new(Mutex::new(VecDeque::with_capacity(RING_LINES)));
+    let stop = Arc::new(AtomicBool::new(false));
+    let done = Arc::new(AtomicBool::new(false));
+    let cmd = logs_stream_cmd(recipe, rank, TAIL_LINES);
+    let (h, s, d) = (history.clone(), stop.clone(), done.clone());
+    let (cfg, node_obj, bus) = (config.clone(), node_cfg.clone(), bus.clone());
+    let (idc, cluster_s, node_s, recipe_s) = (
+        id.clone(),
+        cl.name.clone(),
+        node_cfg.name.clone(),
+        recipe.to_string(),
+    );
+    let task = tokio::spawn(async move {
+        let _ = stream_lines(&cfg, &node_obj, cmd, s, |line| {
+            push_line(&h, line);
+            let _ = bus.send(Event::Log {
+                attach: idc.clone(),
+                cluster: cluster_s.clone(),
+                node: node_s.clone(),
+                recipe: recipe_s.clone(),
+                rank,
+                line: line.to_string(),
+            });
+        })
+        .await;
+        // Stream errors (node offline, container gone) just end the window.
+        d.store(true, Ordering::Relaxed);
+    });
+
+    let attach = LogAttach {
+        id: id.clone(),
+        cluster: cluster.to_string(),
+        node: node.to_string(),
+        recipe: recipe.to_string(),
+        rank,
+        history: history.clone(),
+        stop,
+        done,
+        task,
+    };
+    list.push(attach);
+    Ok((id, vec![]))
+}
+
+/// Detach: stop the stream, abort the task, drop the entry.
+pub fn detach(attaches: &Mutex<Vec<LogAttach>>, attach_id: &str) -> bool {
+    let mut list = attaches.lock().unwrap();
+    if let Some(pos) = list.iter().position(|a| a.id == attach_id) {
+        list[pos].stop.store(true, Ordering::Relaxed);
+        list[pos].task.abort();
+        list.remove(pos);
+        true
+    } else {
+        false
+    }
+}
+
+/// Find an attach by key (for `GET /api/logs?cluster=&node=&recipe=&rank=`).
+pub fn by_key(
+    attaches: &Mutex<Vec<LogAttach>>,
+    cluster: &str,
+    node: &str,
+    recipe: &str,
+    rank: u32,
+    lines: usize,
+) -> Option<(String, Vec<String>)> {
+    let list = attaches.lock().unwrap();
+    list.iter()
+        .find(|a| {
+            a.key()
+                == (
+                    cluster.to_string(),
+                    node.to_string(),
+                    recipe.to_string(),
+                    rank,
+                )
+        })
+        .map(|a| (a.id.clone(), a.history_last(lines)))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::web::events::Event;
+
+    fn ring() -> Arc<Mutex<VecDeque<String>>> {
+        Arc::new(Mutex::new(VecDeque::new()))
+    }
+
+    #[test]
+    fn push_line_trims_to_capacity_keeping_newest() {
+        let r = ring();
+        for i in 0..=RING_LINES {
+            push_line(&r, &format!("line-{i}"));
+        }
+        let g = r.lock().unwrap();
+        assert_eq!(g.len(), RING_LINES, "ring stays at capacity");
+        assert_eq!(g.front().unwrap(), &"line-1".to_string(), "oldest evicted");
+        assert_eq!(
+            g.back().unwrap(),
+            &format!("line-{}", RING_LINES),
+            "newest kept"
+        );
+    }
+
+    #[tokio::test]
+    async fn history_last_returns_oldest_first() {
+        let a = LogAttach {
+            id: "l1".into(),
+            cluster: "c".into(),
+            node: "n".into(),
+            recipe: "r".into(),
+            rank: 0,
+            history: ring(),
+            stop: Arc::new(AtomicBool::new(false)),
+            done: Arc::new(AtomicBool::new(false)),
+            task: tokio::spawn(async {}),
+        };
+        for i in 0..5 {
+            push_line(&a.history, &format!("x{i}"));
+        }
+        assert_eq!(a.history_last(3), ["x2", "x3", "x4"]);
+        assert_eq!(a.history_last(99), ["x0", "x1", "x2", "x3", "x4"]);
+    }
+
+    #[test]
+    fn stream_cmd_carries_recipe_rank_and_tail() {
+        let cmd = logs_stream_cmd("glm52-nf4-lloyd", 1, 200);
+        assert!(cmd.contains("label=sm121.recipe=glm52-nf4-lloyd"));
+        assert!(cmd.contains("label=sm121.rank=1"));
+        assert!(cmd.contains("--tail=200"));
+        assert!(cmd.contains("docker logs -f"));
+    }
+
+    #[tokio::test]
+    async fn attach_is_idempotent_and_bounded() {
+        let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
+        let attaches: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
+        let cfg = crate::web::tests::fake_config();
+        // 8 attaches fill the cap; the 9th is refused.
+        for i in 0..MAX_ATTACHES {
+            let (id, _) = attach(&cfg, &tx, &attaches, "fake", "n0", &format!("r{i}"), 0).unwrap();
+            assert_eq!(id, format!("l{}", i + 1));
+        }
+        let err = attach(&cfg, &tx, &attaches, "fake", "n0", "overflow", 0).unwrap_err();
+        assert!(err.contains("max"), "{err}");
+        // Unknown cluster / node are 400-class errors.
+        assert!(attach(&cfg, &tx, &attaches, "nope", "n0", "r", 0).is_err());
+        assert!(attach(&cfg, &tx, &attaches, "fake", "nope", "r", 0).is_err());
+        // by_key finds what exists, misses what doesn't.
+        assert!(by_key(&attaches, "fake", "n0", "r0", 0, 10).is_some());
+        assert!(by_key(&attaches, "fake", "n0", "r0", 7, 10).is_none());
+        // Detach removes + frees a slot; re-attach after detach works.
+        assert!(detach(&attaches, "l1"));
+        assert!(!detach(&attaches, "l1"));
+        attach(&cfg, &tx, &attaches, "fake", "n0", "new", 0).unwrap();
+        assert_eq!(attaches.lock().unwrap().len(), MAX_ATTACHES);
+    }
+
+    /// A fake attach with a controlled done flag (no real SSH involved).
+    fn fake_attach(id: &str, done: bool) -> LogAttach {
+        LogAttach {
+            id: id.into(),
+            cluster: "fake".into(),
+            node: "n0".into(),
+            recipe: "r".into(),
+            rank: 0,
+            history: Arc::new(Mutex::new(VecDeque::new())),
+            stop: Arc::new(AtomicBool::new(false)),
+            done: Arc::new(AtomicBool::new(done)),
+            // Long-lived task so the attach stays "live" deterministically.
+            task: tokio::spawn(async {
+                tokio::time::sleep(std::time::Duration::from_secs(60)).await
+            }),
+        }
+    }
+
+    #[tokio::test]
+    async fn attach_reuses_live_stream_for_same_key() {
+        let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
+        let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
+        list.lock().unwrap().push(fake_attach("l1", false));
+        let cfg = crate::web::tests::fake_config();
+        // Same key, live stream -> the existing attach is reused, no new task.
+        let (id, history) = attach(&cfg, &tx, &list, "fake", "n0", "r", 0).unwrap();
+        assert_eq!(id, "l1");
+        assert!(history.is_empty());
+        assert_eq!(list.lock().unwrap().len(), 1, "no second attach spawned");
+    }
+
+    #[tokio::test]
+    async fn attach_replaces_finished_stream_for_same_key() {
+        let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
+        let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
+        list.lock().unwrap().push(fake_attach("l1", true));
+        let cfg = crate::web::tests::fake_config();
+        // Same key, finished stream -> dropped and replaced with a new task.
+        attach(&cfg, &tx, &list, "fake", "n0", "r", 0).unwrap();
+        let g = list.lock().unwrap();
+        assert_eq!(g.len(), 1, "old attach dropped");
+        assert_eq!(g[0].id, "l1", "slot id may be reused");
+        assert!(
+            !g[0].done.load(Ordering::Relaxed),
+            "fresh stream not yet done"
+        );
+        g[0].stop.store(true, Ordering::Relaxed); // end the fake task
+    }
+}

src/web/mod.rs

@@ -1,16 +1,19 @@
 //! `sm121 web` — axum + SSE dashboard.
 //!
-//! P2: read-only LED board (state snapshots every 30 s + on job finish)
-//! plus actions — serve/stop/power run as tracked jobs calling the same
-//! handlers the CLI uses. Data endpoints are token-gated when `--token`
-//! is set (required when binding off-loopback).
+//! Read-only LED board (state snapshots every 30 s + on job finish),
+//! actions — serve/stop/power run as tracked jobs calling the same
+//! handlers the CLI uses — and live log windows (SSH `docker logs -f`
+//! attaches with ring-buffer history). Data endpoints are token-gated
+//! when `--token` is set (required when binding off-loopback).
 //!
 //! Design: `specs/tech-architecture/web-dashboard.md`.
 
 mod api;
 pub mod events;
 pub mod jobs;
+pub mod logs;
 pub mod probe;
+pub mod ssh_stream;
 
 use crate::config::Config;
 use crate::web::events::{Event, build_snapshot};
@@ -43,6 +46,8 @@ pub struct AppState {
     pub last: Arc<Mutex<Option<events::Snapshot>>>,
     /// Tracked action jobs (serve/stop/power); capped at `jobs::MAX_JOBS`.
     pub jobs: Arc<Mutex<Vec<jobs::Job>>>,
+    /// Tracked log streams; capped at `logs::MAX_ATTACHES`.
+    pub log_attaches: Arc<Mutex<Vec<logs::LogAttach>>>,
 }
 
 /// Probe the fleet and broadcast a fresh state snapshot (supervisor tick
@@ -50,7 +55,21 @@ pub struct AppState {
 pub(crate) async fn push_state(app: &AppState) {
     let fleet = probe::probe(&app.config).await;
     let jobs = app.jobs.lock().unwrap().clone();
-    let snap = build_snapshot(&app.config, &fleet, &jobs);
+    let attaches: Vec<events::AttachSnap> = app
+        .log_attaches
+        .lock()
+        .unwrap()
+        .iter()
+        .map(|a| events::AttachSnap {
+            id: a.id.clone(),
+            cluster: a.cluster.clone(),
+            node: a.node.clone(),
+            recipe: a.recipe.clone(),
+            rank: a.rank,
+            done: a.done.load(std::sync::atomic::Ordering::Relaxed),
+        })
+        .collect();
+    let snap = build_snapshot(&app.config, &fleet, &jobs, &attaches);
     let _ = app.bus.send(Event::State {
         ts: snap.ts,
         snapshot: snap.clone(),
@@ -131,6 +150,12 @@ pub fn router(app: AppState) -> Router {
         .route("/api/stop", post(api::stop))
         .route("/api/power", post(api::power))
         .route("/api/recipes", get(api::recipes))
+        .route(
+            "/api/logs",
+            get(api::logs_history)
+                .post(api::logs_attach)
+                .delete(api::logs_detach),
+        )
         .route("/events", get(events_stream))
         .layer(middleware::from_fn_with_state(app.clone(), token_guard));
     Router::new()
@@ -160,6 +185,7 @@ pub async fn run(
         token,
         last: Arc::new(Mutex::new(None)),
         jobs: Arc::new(Mutex::new(Vec::new())),
+        log_attaches: Arc::new(Mutex::new(Vec::new())),
     };
     tokio::spawn(supervisor(app.clone()));
     let listener = tokio::net::TcpListener::bind(addr).await?;

src/web/ssh_stream.rs

@@ -0,0 +1,93 @@
+//! SSH output streaming for web log windows — `channel.wait()` data chunks
+//! split into complete lines (a chunk without a trailing newline is
+//! buffered until it completes; vLLM banner lines are long but single-line).
+
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+
+use anyhow::Result;
+use russh::ChannelMsg;
+
+use crate::config::{Config, Node};
+use crate::ssh::ssh_session_follow;
+
+/// Split `chunk` into complete lines, carrying `pending` across chunks.
+/// `on_line` receives each complete line (no trailing newline).
+pub(crate) fn split_lines(pending: &mut String, chunk: &str, on_line: &mut impl FnMut(&str)) {
+    pending.push_str(chunk);
+    while let Some(nl) = pending.find('\n') {
+        let line: String = pending.drain(..=nl).collect();
+        // drain(..=nl) keeps the newline in `line` — strip it, then a CR.
+        on_line(line.trim_end_matches('\n').trim_end_matches('\r'));
+    }
+}
+
+/// Stream `command` output via SSH (no inactivity timeout — `docker logs -f`
+/// is long-lived), calling `on_line` per complete line. Returns when the
+/// channel closes (container stopped / docker exited) or `stop` is set.
+pub async fn stream_lines(
+    config: &Config,
+    node: &Node,
+    command: String,
+    stop: Arc<AtomicBool>,
+    mut on_line: impl FnMut(&str) + Send,
+) -> Result<()> {
+    let session = ssh_session_follow(config, node).await?;
+    let mut channel = session.channel_open_session().await?;
+    channel.exec(true, command.as_bytes()).await?;
+    let mut pending = String::new();
+    while let Some(msg) = channel.wait().await {
+        if stop.load(Ordering::Relaxed) {
+            break;
+        }
+        match msg {
+            ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => {
+                split_lines(&mut pending, &String::from_utf8_lossy(&data), &mut on_line);
+            }
+            _ => {}
+        }
+    }
+    // Flush a trailing partial line (stream ended without a final newline).
+    if !pending.is_empty() {
+        let line = pending.clone();
+        on_line(&line);
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::split_lines;
+
+    #[test]
+    fn splits_complete_lines_and_buffers_partials() {
+        let mut pending = String::new();
+        let mut lines: Vec<String> = vec![];
+        // Only the newline-terminated part is a line; "de" has no trailing
+        // newline yet and stays pending.
+        split_lines(&mut pending, "abc\nde", &mut |l| lines.push(l.to_string()));
+        assert_eq!(lines, ["abc"]);
+        assert_eq!(pending, "de");
+        // More partial data: still no line.
+        split_lines(&mut pending, "par", &mut |l| lines.push(l.to_string()));
+        assert_eq!(lines, ["abc"], "no line emitted for a partial");
+        // …and completes on the next chunk, with a new partial left behind.
+        split_lines(&mut pending, "tial\nnext", &mut |l| {
+            lines.push(l.to_string())
+        });
+        assert_eq!(lines, ["abc", "departial"]);
+        assert_eq!(pending, "next");
+    }
+
+    #[test]
+    fn strips_cr_and_handles_blank_lines() {
+        let mut pending = String::new();
+        let mut lines: Vec<String> = vec![];
+        split_lines(&mut pending, "a\r\n\nb\n", &mut |l| {
+            lines.push(l.to_string())
+        });
+        assert_eq!(lines, ["a", "", "b"]);
+    }
+}

src/web/tests.rs

@@ -18,7 +18,7 @@ async fn call(addr: SocketAddr, path: &str) -> reqwest::Response {
 
 /// Config with one node at 127.0.0.1:2222 — connection refused instantly,
 /// machine-independent (no real SSH, no hardware).
-fn fake_config() -> Config {
+pub(crate) fn fake_config() -> Config {
     Config {
         ssh_user: "u".into(),
         ssh_key_path: "/nonexistent/key".into(),
@@ -61,6 +61,7 @@ fn app_state_with(config: Config, token: Option<&str>) -> AppState {
         token: token.map(String::from),
         last: std::sync::Arc::new(std::sync::Mutex::new(None)),
         jobs: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
+        log_attaches: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
     }
 }
 
@@ -130,6 +131,7 @@ async fn bus_delivers_state_events() {
         ts: 1,
         clusters: Vec::new(),
         jobs: Vec::new(),
+        attaches: Vec::new(),
     };
     tx.send(Event::State {
         ts: snap.ts,
@@ -143,6 +145,7 @@ async fn bus_delivers_state_events() {
             assert!(snapshot.clusters.is_empty());
         }
         Event::Job { .. } => panic!("unexpected job event"),
+        Event::Log { .. } => panic!("unexpected log event"),
     }
 }
 
@@ -158,6 +161,7 @@ async fn snapshot_serializes_with_type_tag() {
                 soft_off: std::collections::HashMap::new(),
             },
             &[],
+            &[],
         ),
     };
     let json = serde_json::to_value(&ev).unwrap();
@@ -355,3 +359,94 @@ async fn ui_playwright_checks() {
     );
     eprintln!("{stdout}");
 }
+
+// --- P3: log windows -----------------------------------------------------------
+
+#[tokio::test]
+async fn log_attach_get_detach_roundtrip() {
+    let (addr, _h) = boot().await;
+    let base = format!("http://{addr}");
+    let client = reqwest::Client::new();
+
+    // Attach on the fake cluster (node n0 at a refused port: the stream task
+    // dies immediately, but the registry wiring is what's under test).
+    let res = client
+        .post(format!("{base}/api/logs"))
+        .json(
+            &serde_json::json!({ "cluster": "fake", "node": "n0", "recipe": "deepseek-v4-flash" }),
+        )
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(res.status(), 200, "attach ok");
+    let body: serde_json::Value = res.json().await.unwrap();
+    let attach_id = body["attach_id"].as_str().unwrap().to_string();
+    assert_eq!(attach_id, "l1");
+    assert!(body["history"].is_array());
+
+    // Idempotent: same key -> same attach.
+    let res = client
+        .post(format!("{base}/api/logs"))
+        .json(
+            &serde_json::json!({ "cluster": "fake", "node": "n0", "recipe": "deepseek-v4-flash" }),
+        )
+        .send()
+        .await
+        .unwrap();
+    let body: serde_json::Value = res.json().await.unwrap();
+    assert_eq!(body["attach_id"], attach_id);
+
+    // History endpoint (empty ring — the SSH stream failed on the refused port).
+    let res = client
+        .get(format!(
+            "{base}/api/logs?cluster=fake&node=n0&recipe=deepseek-v4-flash"
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(res.status(), 200);
+    let body: serde_json::Value = res.json().await.unwrap();
+    assert_eq!(body["attach_id"], attach_id);
+    assert!(body["history"].as_array().unwrap().is_empty());
+
+    // Unknown cluster -> 400.
+    let res = client
+        .post(format!("{base}/api/logs"))
+        .json(&serde_json::json!({ "cluster": "nope", "node": "n0", "recipe": "r" }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(res.status(), 400);
+
+    // Detach -> 204, then 404.
+    let res = client
+        .delete(format!("{base}/api/logs"))
+        .json(&serde_json::json!({ "attach_id": attach_id }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(res.status(), 204);
+    let res = client
+        .delete(format!("{base}/api/logs"))
+        .json(&serde_json::json!({ "attach_id": attach_id }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(res.status(), 404);
+}
+
+#[tokio::test]
+async fn log_event_serializes_with_type_tag() {
+    let ev = Event::Log {
+        attach: "l1".into(),
+        cluster: "fake".into(),
+        node: "n0".into(),
+        recipe: "r".into(),
+        rank: 0,
+        line: "hello".into(),
+    };
+    let json = serde_json::to_value(&ev).unwrap();
+    assert_eq!(json["type"], "log");
+    assert_eq!(json["attach"], "l1");
+    assert_eq!(json["line"], "hello");
+}