From bdfeac80a7e4e0d2755793e994f0b1a5d31b7b96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 5 Jul 2026 20:40:07 +0200 Subject: [PATCH] refactor(web_ui): split into a module dir by concern, serve stays in mod.rs --- hive-ag3nt/src/web_ui.rs | 1232 ------------------------------ hive-ag3nt/src/web_ui/actions.rs | 172 +++++ hive-ag3nt/src/web_ui/auth.rs | 141 ++++ hive-ag3nt/src/web_ui/mod.rs | 267 +++++++ hive-ag3nt/src/web_ui/screen.rs | 96 +++ hive-ag3nt/src/web_ui/state.rs | 434 +++++++++++ hive-ag3nt/src/web_ui/stats.rs | 131 ++++ hive-ag3nt/src/web_ui/stream.rs | 70 ++ 8 files changed, 1311 insertions(+), 1232 deletions(-) delete mode 100644 hive-ag3nt/src/web_ui.rs create mode 100644 hive-ag3nt/src/web_ui/actions.rs create mode 100644 hive-ag3nt/src/web_ui/auth.rs create mode 100644 hive-ag3nt/src/web_ui/mod.rs create mode 100644 hive-ag3nt/src/web_ui/screen.rs create mode 100644 hive-ag3nt/src/web_ui/state.rs create mode 100644 hive-ag3nt/src/web_ui/stats.rs create mode 100644 hive-ag3nt/src/web_ui/stream.rs diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs deleted file mode 100644 index 34a36700..00000000 --- a/hive-ag3nt/src/web_ui.rs +++ /dev/null @@ -1,1232 +0,0 @@ -//! Per-container HTTP UI. SPA shape: `GET /` returns a static shell; -//! `GET /static/*` serves CSS + JS; `GET /api/state` returns the page -//! state as JSON; the JS app renders. Live events stream on -//! `/events/stream`. Action POSTs (`/send`, `/login/*`) return either a -//! 303 Redirect (for browsers that submit the form normally) or just -//! 200 OK — the JS app re-fetches `/api/state` afterwards. - -use std::convert::Infallible; -use std::net::SocketAddr; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; - -use anyhow::{Context, Result}; -use axum::{ - Form, Router, - extract::State, - http::StatusCode, - response::{ - IntoResponse, Response, - sse::{Event, KeepAlive, Sse}, - }, - routing::{get, post}, -}; -use serde::{Deserialize, Serialize}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream}; -use tower_http::services::ServeDir; - -use crate::client; -use crate::events::Bus; -use crate::login::LoginState; -use crate::login_session::{LoginSession, drop_if_finished}; - -/// Deadline for broker-backed fetches on web-UI request paths. The -/// page's critical fields (status, turn state, usage) are all -/// in-memory; a busy or stalled hive-c0re must degrade the -/// socket-backed extras (inbox rows, loose ends, reminder stats) -/// instead of hanging the whole response — an unbounded await here is -/// what let `/api/state` stall long enough to bork the terminal. -const SOCKET_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3); - -/// Live login state for the web UI. The harness updates this in place as it -/// transitions between `NeedsLogin` and `Online`; the UI reads on each -/// render. -pub type LoginStateCell = Arc>; - -#[derive(Clone)] -struct AppState { - label: String, - login: LoginStateCell, - session: Arc>>>, - bus: Bus, - socket: PathBuf, - /// VNC port from the `HIVE_GUI_VNC_PORT` env var at startup. - /// `None` when unset (gui not enabled for this agent). - gui_vnc_port: Option, -} - -/// Bind the per-container web listener and serve the SPA. -/// -/// `HIVE_WEB_SOCKET` opt-in selects unix-socket vs TCP binding; the -/// dual-mode transition + gateway-side consumer live in -/// [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md) and -/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md). -/// -/// # Errors -/// -/// Returns an error if neither the TCP listener (default) nor the -/// unix-socket bind (`HIVE_WEB_SOCKET`, if set) can be acquired, or -/// if `HIVE_STATIC_DIR` is missing. -pub async fn serve( - label: String, - port: u16, - login: LoginStateCell, - bus: Bus, - socket: PathBuf, -) -> Result<()> { - let gui_vnc_port = read_gui_vnc_port(); - let static_dir: PathBuf = std::env::var_os("HIVE_STATIC_DIR") - .map(PathBuf::from) - .context( - "HIVE_STATIC_DIR env var not set — point it at the merged \ - per-agent dist (see hyperhive.frontend.mergedDist in nix)", - )?; - if !static_dir.is_dir() { - anyhow::bail!( - "HIVE_STATIC_DIR ({}) is not a directory", - static_dir.display() - ); - } - tracing::info!(static_dir = %static_dir.display(), "web UI static dir resolved"); - let state = AppState { - label, - login, - session: Arc::new(Mutex::new(None)), - bus, - socket, - gui_vnc_port, - }; - let app = Router::new() - .route("/api/state", get(api_state)) - .route("/api/dashboard-state", get(api_dashboard_state)) - .route("/events/stream", get(events_stream)) - .route("/events/history", get(events_history)) - .route("/send", post(post_send)) - .route("/login/start", post(post_login_start)) - .route("/login/code", post(post_login_code)) - .route("/login/cancel", post(post_login_cancel)) - .route("/api/cancel", post(post_cancel_turn)) - .route("/api/compact", post(post_compact)) - .route("/api/model", post(post_set_model)) - .route("/api/effort", post(post_set_effort)) - .route("/api/new-session", post(post_new_session)) - .route("/api/logout", post(post_logout)) - .route("/api/loose-ends", get(api_loose_ends)) - .route("/api/bash-tasks", get(api_bash_tasks)) - .route("/api/stats", get(api_stats)) - .route("/screen/ws", get(screen_ws)) - .route("/icon", get(serve_icon)) - // Anything else (`/`, `/stats`, `/screen`, `/static/*`) - // falls through to the merged dist. ServeDir auto-appends - // `.html` when the URL is a bare path that matches a file - // (so `/stats` → `dist/stats.html`, `/screen` → `dist/ - // screen.html`). Per-agent `extraFiles` additions are - // already layered into this same directory (see - // hyperhive.frontend.mergedDist in nix). - .fallback_service(ServeDir::new(&static_dir)) - .with_state(state); - // `HIVE_WEB_SOCKET` opt-in: when set + non-empty, bind a - // `UnixListener` at the given path. Empty string treated as - // unset so a stray `HIVE_WEB_SOCKET=` doesn't trap us into an - // un-bindable empty path. Falls through to the TCP path below - // otherwise. See docs/gateway.md::Per-agent unix-socket upstream - // for the gateway-side consumer. - if let Some(socket_path) = std::env::var_os("HIVE_WEB_SOCKET") - && !socket_path.is_empty() - { - let path = PathBuf::from(socket_path); - let listener = bind_unix(&path)?; - tracing::info!(socket = %path.display(), "web UI listening on unix socket"); - axum::serve(listener, app).await?; - return Ok(()); - } - let addr = SocketAddr::from(([0, 0, 0, 0], port)); - let listener = bind_with_retry(addr, "web UI").await?; - tracing::info!(%port, "web UI listening on tcp"); - axum::serve(listener, app).await?; - Ok(()) -} - -/// Bind a `UnixListener` at `path` and drop a `.bound` marker next -/// to it so c0re's gateway-map writer knows the socket is live. -/// Best-effort unlinks any stale socket left from a crashed previous -/// harness (clean exit removes it, but `bind(2)` refuses to overwrite -/// an existing file) and `mkdir -p`s the parent for first-boot. Mode -/// `0o666` — world-accessible so the gateway container's nginx process -/// can `connect(2)` without sharing a group with the agent user. -/// The per-agent subdir (`/run/hive-agent//`) is only accessible -/// to containers that have it bind-mounted, so world-accessible sockets -/// are not a material risk. -/// -/// Marker-gating + the gateway-side consumer: see -/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md). -fn bind_unix(path: &Path) -> Result { - use std::os::unix::fs::PermissionsExt; - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("create socket parent dir {}", parent.display()))?; - } - // Best-effort: ENOENT is fine (no stale file); any other error - // surfaces via the bind below with a clearer "AddrInUse" / perms - // message than a partial cleanup would. - let _ = std::fs::remove_file(path); - let listener = tokio::net::UnixListener::bind(path) - .with_context(|| format!("bind unix socket at {}", path.display()))?; - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666)) - .with_context(|| format!("set perms on {}", path.display()))?; - // Best-effort ready marker: failed write isn't fatal (the harness - // still binds + serves), it just means the gateway side keeps the - // TCP upstream for one more sync tick. - if let Some(parent) = path.parent() { - let marker = parent.join("hyperhive-socket-bound"); - if let Err(e) = std::fs::write(&marker, b"") { - tracing::warn!( - marker = %marker.display(), error = %e, - "failed to write hyperhive-socket-bound marker — gateway may keep TCP upstream" - ); - } - } - Ok(listener) -} - -// --------------------------------------------------------------------------- -// Static assets + state snapshot -// --------------------------------------------------------------------------- - -/// Bind a TCP listener with `SO_REUSEADDR` set, retrying on -/// `AddrInUse` indefinitely with exponential backoff capped at 2s. -/// First 12 attempts log at WARN; subsequent attempts log at INFO so -/// a long-held stale socket doesn't flood the journal. -/// -/// Uncapped retry + dashboard-banner-on-real-collision rationale: -/// see [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md). -async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result { - let mut delay_ms = 250u64; - let mut attempts = 0u32; - loop { - match try_bind(addr) { - Ok(l) => { - if attempts > 0 { - tracing::info!( - %addr, attempts, - "{label}: bind succeeded after retry" - ); - } - return Ok(l); - } - Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { - let attempt = attempts + 1; - if attempt <= 12 { - tracing::warn!( - %addr, attempt, - "{label}: AddrInUse, retrying in {delay_ms}ms" - ); - } else { - tracing::info!( - %addr, attempt, - "{label}: AddrInUse still holding, retrying in {delay_ms}ms" - ); - } - tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; - attempts += 1; - delay_ms = (delay_ms * 2).min(2000); - } - Err(e) => { - return Err(e).with_context(|| format!("bind {label} on {addr}")); - } - } - } -} - -fn try_bind(addr: SocketAddr) -> std::io::Result { - let sock = match addr { - SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?, - SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?, - }; - sock.set_reuseaddr(true)?; - sock.bind(addr)?; - sock.listen(1024) -} - -/// This agent's icon. Serves the operator-configured SVG from -/// `/etc/hyperhive/icon.svg` (set via the `hyperhive.icon` agent.nix -/// option) when present, otherwise the bundled default hyperhive logo. -/// Always returns an image, so consumers (dashboard, favicon) can hit -/// `/icon` unconditionally without probing whether one is configured. -async fn serve_icon() -> impl IntoResponse { - // Per-agent icon overrides go through `/etc/hyperhive/icon.svg` - // (set via the `hyperhive.icon` agent.nix option); the bundled - // default is resolved at runtime from - // `$HIVE_ASSETS_DIR/branding/hyperhive.svg`. If neither file can - // be read we serve an empty body — keeps the response a valid SVG - // content-type without a panic on a misconfigured container. - let body = std::fs::read_to_string("/etc/hyperhive/icon.svg").unwrap_or_else(|_| { - std::fs::read_to_string(hive_sh4re::assets::branding_svg()).unwrap_or_default() - }); - ([("content-type", "image/svg+xml")], body) -} - -/// The fixed VNC port weston bound, from the `HIVE_GUI_VNC_PORT` env var -/// the harness service sets when gui is enabled (see weston-vnc.nix). -/// `None` when unset (gui not enabled for this agent) or unparseable. -/// The port is a fixed, container-local value — no per-agent hashing, no -/// marker file — because network isolation is unconditional (each agent -/// has its own netns, so the port can't collide across containers). -fn read_gui_vnc_port() -> Option { - std::env::var("HIVE_GUI_VNC_PORT").ok()?.parse().ok() -} - -/// WebSocket handler: upgrade then pump bytes between the WS client and -/// the VNC server on `127.0.0.1:`. Returns 404 when gui is not -/// enabled for this agent. -async fn screen_ws( - ws: axum::extract::ws::WebSocketUpgrade, - State(state): State, -) -> Response { - let Some(vnc_port) = state.gui_vnc_port else { - return (StatusCode::NOT_FOUND, "gui not enabled for this agent").into_response(); - }; - ws.on_upgrade(move |socket| relay_ws_vnc(socket, vnc_port)) -} - -/// Pure byte pump: forwards raw bytes between the WebSocket client and -/// the VNC TCP stream. Transparent to any RFB variant (plain, `VeNCrypt`). -async fn relay_ws_vnc(socket: axum::extract::ws::WebSocket, vnc_port: u16) { - // Import futures traits locally so they don't conflict with - // tokio_stream::StreamExt used at module scope. - use axum::extract::ws::Message; - use futures_util::{SinkExt, StreamExt as _}; - - let addr = format!("127.0.0.1:{vnc_port}"); - let Ok(tcp) = tokio::net::TcpStream::connect(&addr).await else { - tracing::warn!(%addr, "screen/ws: could not connect to VNC server"); - return; - }; - let (mut tcp_rx, mut tcp_tx) = tcp.into_split(); - let (mut ws_tx, mut ws_rx) = socket.split(); - - // WS → TCP - let ws_to_tcp = tokio::spawn(async move { - while let Some(Ok(msg)) = futures_util::StreamExt::next(&mut ws_rx).await { - match msg { - Message::Binary(data) if tcp_tx.write_all(&data).await.is_err() => { - break; - } - Message::Close(_) => break, - _ => {} // ping/pong/text: ignore - } - } - }); - - // TCP → WS - let tcp_to_ws = tokio::spawn(async move { - let mut buf = vec![0u8; 8192]; - loop { - match tcp_rx.read(&mut buf).await { - Ok(0) | Err(_) => break, - Ok(n) => { - if ws_tx - .send(Message::Binary(buf[..n].to_vec().into())) - .await - .is_err() - { - break; - } - } - } - } - }); - - // Wait for either direction to close, then let both tasks drop. - tokio::select! { - _ = ws_to_tcp => {} - _ = tcp_to_ws => {} - } -} - -#[derive(Deserialize)] -struct StatsQuery { - window: Option, -} - -async fn api_stats( - State(state): State, - axum::extract::Query(q): axum::extract::Query, -) -> axum::Json { - let window = crate::stats::Window::parse(q.window.as_deref().unwrap_or("24h")); - let mut snapshot = crate::stats::snapshot_default(window); - // Pass the window span to the reminder-stats RPC so the broker - // filters its counts to the same time range as the chart data. - let window_secs = window.span_secs(); - let window_secs_u = u64::try_from(window_secs).unwrap_or(0); - snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await; - axum::Json(snapshot) -} - -#[derive(Serialize)] -struct StateSnapshot { - /// Bus seq at the moment this snapshot was assembled. Clients dedupe - /// their buffered SSE traffic against this value: events with - /// `seq <= snapshot.seq` are already reflected (or pre-date the - /// snapshot); `seq > snapshot.seq` is post-snapshot. Reset to 0 on - /// harness restart — clients treat reconnect as a fresh world. - seq: u64, - label: String, - /// Hive-qualified long name (`${label}@${hyperhive.domain}`) when - /// the host has been configured for a multi-hive swarm; falls back - /// to the short label when the hive domain env var is unset. - /// The frontend uses this for the page title / agent self-introduction; - /// when it equals `label`, the page renders the short form unchanged. - qualified_label: String, - dashboard_port: u16, - /// `"online"` | `"rate_limited"` | `"needs_login_idle"` | `"needs_login_in_progress"`. - status: &'static str, - /// Present when `status == "needs_login_in_progress"`. - session: Option, - /// Last N messages addressed to this agent, newest-first. Pulled - /// from the broker via the per-agent socket on each render. - /// Empty on transport failure. - inbox: Vec, - /// Authoritative turn-loop state from the harness and the unix - /// timestamp the state was entered. The JS computes the age - /// client-side off this rather than tracking it from SSE events. - turn_state: crate::events::TurnState, - turn_state_since: i64, - /// Currently-active claude model name. Reflected on the page so - /// the operator can see what they just switched to (and what's - /// in flight). Mutable at runtime via `POST /api/model`. - model: String, - /// Effective context-window token budget for the current model. - /// Primary source: API-reported `modelUsage.*.contextWindow` from - /// the last result event (authoritative per-inference active window). - /// Falls back to `HIVE_CONTEXT_WINDOW_TOKENS_*` env vars, then 200 000. - /// Consumers (e.g. dashboard badge) use this to render ctx-usage %. - context_window_tokens: u64, - /// Last-inference token usage from the most recent completed - /// turn — represents the current context-window size at turn-end. - /// `null` until the first turn finishes. - ctx_usage: Option, - /// Cumulative token usage across the most recent turn's inferences - /// (cost signal). `null` until the first turn finishes. - cost_usage: Option, - /// Navigation links for this agent page. Also served via - /// `DashboardState.links` (`GET /api/dashboard-state`) for the - /// dashboard card's icon strip. Both are produced by `agent_links()` - /// — single source of truth. See [`docs/web-ui/dashboard.md::Container row`] - /// for the frontend resolver + which links appear in which conditions. - links: Vec, - /// Public URL of the forge served by hive-gateway (e.g. - /// `"https://forge.pr1ma.darkest.space"`). Sourced from - /// `HIVE_FORGE_PUBLIC_URL`; `None` when `forge.behindGateway=false` - /// or the env var is absent. The frontend uses this to build forge - /// nav-strip links instead of hardcoding `:3000`. - forge_public_url: Option, - /// Human name of this hive instance (e.g. `"pr1ma"`). Sourced - /// from `HYPERHIVE_HIVE_NAME`; `None` when unset. The frontend - /// uses this for the page `` and header label so browser - /// tabs disambiguate when multiple hives are open in parallel. - hive_name: Option<String>, - /// Human name of the swarm (e.g. `"constellat1on"`). Sourced from - /// `HYPERHIVE_SWARM_NAME`; `None` when unset. - swarm_name: Option<String>, - /// Ordered list of model short-names the operator has declared as - /// available on this hive. Sourced from `HIVE_AVAILABLE_MODELS` - /// (comma-separated, set by `services.hyperhive.availableModels`). - /// Falls back to `["haiku", "sonnet", "opus"]` when the env var is - /// absent or empty. The frontend model quick-picker renders one button - /// per entry in this list, so operators can add new models or drop - /// ones they don't want without touching the frontend code. - available_models: Vec<String>, - /// Currently-active claude effort level. Reflected on the page so the - /// operator's effort picker shows the live selection. Mutable at - /// runtime via `POST /api/effort`; applies on the next session. - effort: String, - /// Selectable effort levels for the picker, ascending. Fixed set - /// (`low`, `medium`, `high`, `xhigh`, `max`) — sourced from - /// [`crate::events::EFFORT_LEVELS`], not operator-configurable like - /// `available_models`. The frontend renders one button per entry. - available_efforts: Vec<String>, -} - -/// One navigation link in the agent page header row. The same JSON -/// shape appears in both `StateSnapshot.links` (`GET /api/state`, -/// per-agent page) and `DashboardState.links` (`GET /api/dashboard-state`, -/// dashboard card icon strip). `agent_links()` is the single source -/// of truth for what links an agent exposes. -#[derive(Serialize)] -struct AgentLink { - /// `kind = Container | Forge` → path; `kind = External` → full URL. - /// The frontend prepends the right base before rendering. - url: String, - icon: String, - label: String, - kind: AgentLinkKind, -} - -/// Resolution hint for `AgentLink.url`. The agent backend can't know -/// which hostname the browser sees (especially when the dashboard -/// proxies the call from a different origin), so it labels each link -/// and lets the frontend prepend the right base. -#[derive(Serialize, Clone, Copy)] -#[serde(rename_all = "snake_case")] -enum AgentLinkKind { - /// `url` is a path on the agent's container web UI (`/stats`, - /// `/screen`). Agent page: same-origin path. Dashboard: - /// `http://<host>:<container.port><url>`. - Container, - /// `url` is a path on the local Forgejo (`/<label>`, - /// `/agent-configs/<label>`). Both surfaces: - /// `http://<host>:3000<url>`. - Forge, - /// `url` is already a fully-qualified absolute URL — use as-is. - /// Agent-declared `hyperhive.dashboardLinks` extras arrive here. - External, -} - -#[derive(Serialize)] -struct SessionView { - /// First `https://…` claude emitted on stdout, if any. - url: Option<String>, - /// Accumulated stdout + stderr. - output: String, - finished: bool, - exit_note: Option<String>, -} - -/// Proxy this agent's loose-ends list via the per-agent socket. The -/// web UI surfaces the result as a collapsible section in the page -/// so the operator can see at a glance what's pending against the -/// agent (questions asked by it, peer questions targeting it, -/// reminders it scheduled, approvals for the manager). Same data -/// the `mcp__hyperhive__get_loose_ends` tool sees from inside the -/// container. -async fn api_loose_ends(State(state): State<AppState>) -> Response { - let loose_ends: Vec<hive_sh4re::LooseEnd> = match tokio::time::timeout( - SOCKET_FETCH_TIMEOUT, - client::request::<_, hive_sh4re::Response>( - &state.socket, - &hive_sh4re::Request::GetLooseEnds { agent: None }, - ), - ) - .await - { - Ok(Ok(hive_sh4re::Response::LooseEnds { loose_ends })) => loose_ends, - Ok(Ok(hive_sh4re::Response::Err { message })) => { - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("get_loose_ends: {message}"), - ); - } - Ok(Ok(other)) => { - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("unexpected response: {other:?}"), - ); - } - Ok(Err(e)) => { - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("transport: {e:#}"), - ); - } - Err(_) => { - return error_response( - StatusCode::CONFLICT, - "get_loose_ends: timed out — hive-c0re busy, retry", - ); - } - }; - axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response() -} - -/// `GET /api/bash-tasks` — snapshot of this agent's in-flight bash tasks. -/// -/// The `hive-bash-mcp` daemon runs in this same container and writes one -/// `<id>.json` ([`hive_sh4re::TaskFile`]) per task under the harness -/// `bash-tasks/` dir. This reads that dir and returns the tasks still -/// `Pending` or `Running`, so the agent page can show what's running without -/// going through the broker. Snapshot only — the page polls/refreshes it like -/// `/api/loose-ends`; there's no live SSE push for task state yet. Unreadable -/// or malformed files (incl. the daemon's `.json.tmp` scratch writes, which -/// don't match the `.json` extension) are skipped so one stray file can't -/// fail the whole list. -async fn api_bash_tasks() -> Response { - let dir = crate::paths::harness_dir().join("bash-tasks"); - // The dir scan + per-file reads are blocking fs I/O; run them off the - // async executor so a slow or large tasks dir can't stall other requests. - let tasks = tokio::task::spawn_blocking(move || { - let mut tasks: Vec<hive_sh4re::TaskFile> = Vec::new(); - let Ok(rd) = std::fs::read_dir(&dir) else { - return tasks; - }; - for entry in rd.flatten() { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some("json") { - continue; - } - let Ok(text) = std::fs::read_to_string(&path) else { - continue; - }; - let Ok(task) = serde_json::from_str::<hive_sh4re::TaskFile>(&text) else { - continue; - }; - if matches!( - task.status, - hive_sh4re::TaskStatus::Pending | hive_sh4re::TaskStatus::Running - ) { - tasks.push(task); - } - } - // Running before Pending, then oldest-first so a long-runner sits on top. - tasks.sort_by(|a, b| { - let rank = |s: &hive_sh4re::TaskStatus| match s { - hive_sh4re::TaskStatus::Running => 0, - _ => 1, - }; - rank(&a.status) - .cmp(&rank(&b.status)) - .then(a.created_at.cmp(&b.created_at)) - }); - tasks - }) - .await - .unwrap_or_default(); - axum::Json(serde_json::json!({ "tasks": tasks })).into_response() -} - -async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> { - // Capture seq *before* any reads so the dedupe contract is - // "events with seq > snapshot.seq are post-snapshot, never missed." - let seq = state.bus.current_seq(); - drop_if_finished(&state.session); - let login = *state.login.lock().unwrap(); - let session_snapshot = state.session.lock().unwrap().clone(); - let (status, session_view) = match (login, session_snapshot) { - (LoginState::Online, _) if state.bus.is_rate_limited() => ("rate_limited", None), - (LoginState::Online, _) => ("online", None), - (LoginState::NeedsLogin, None) => ("needs_login_idle", None), - (LoginState::NeedsLogin, Some(s)) => ( - "needs_login_in_progress", - Some(SessionView { - url: s.url(), - output: s.output(), - finished: s.finished(), - exit_note: s.exit_note(), - }), - ), - }; - let dashboard_port = std::env::var("HIVE_DASHBOARD_PORT") - .ok() - .and_then(|s| s.parse::<u16>().ok()) - .unwrap_or(7000); - let inbox = recent_inbox(&state.socket).await; - let (turn_state, turn_state_since) = state.bus.state_snapshot(); - let model = state.bus.model(); - let context_window_tokens = state - .bus - .api_context_window() - .unwrap_or_else(|| crate::events::context_window_tokens(&model)); - let ctx_usage = state.bus.last_ctx_usage(); - let cost_usage = state.bus.last_cost_usage(); - let effort = state.bus.effort(); - axum::Json(StateSnapshot { - seq, - label: state.label.clone(), - qualified_label: crate::identity::qualify(&state.label), - dashboard_port, - status, - session: session_view, - inbox, - turn_state, - turn_state_since, - model, - context_window_tokens, - ctx_usage, - cost_usage, - links: agent_links(&state.label, state.gui_vnc_port.is_some()), - forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") - .ok() - .filter(|s| !s.is_empty()), - hive_name: crate::identity::hive_name(), - swarm_name: crate::identity::swarm_name(), - available_models: available_models(), - effort, - available_efforts: crate::events::EFFORT_LEVELS - .iter() - .map(ToString::to_string) - .collect(), - }) -} - -/// Lean snapshot of the agent-owned fields that the dashboard card -/// needs. Served at `GET /api/dashboard-state` (accessible through the -/// gateway at `/agent/<name>/api/dashboard-state`). The dashboard -/// fetches this once per running agent to get fresh, agent-authoritative -/// values instead of relying on hive-c0re's periodic file-reads. -/// -/// Structural fields (running, `needs_update`, `deployed_sha`, parent, …) -/// continue to come from hive-c0re's `/api/state`; this endpoint covers -/// only the fields the agent itself is the source of truth for. -#[derive(serde::Serialize)] -struct DashboardState { - /// Free-text status set by `set_status`, read directly from the - /// `hyperhive-status` file the harness writes. `None` when unset. - #[serde(skip_serializing_if = "Option::is_none")] - status_text: Option<String>, - /// Unix timestamp (seconds) when the status file was last written. - /// `None` when no status is set. - #[serde(skip_serializing_if = "Option::is_none")] - status_set_at: Option<i64>, - /// Full context-window size from the most recent completed turn - /// (`ctx_usage.context_tokens()` = input + cache-read + cache-creation). - /// `None` until the first turn finishes. Drives the `ctx·Nk` card badge. - #[serde(skip_serializing_if = "Option::is_none")] - ctx_tokens: Option<u64>, - /// Effective context-window budget for the current model. Same - /// derivation as `StateSnapshot::context_window_tokens`. - context_window_tokens: u64, - /// True while the harness is parked after a rate-limit response. - rate_limited: bool, - /// Navigation links for the dashboard card's icon strip. This is - /// the authoritative source — includes the screen link (GUI agents) - /// which hive-c0re's disk-based fallback cannot determine. - links: Vec<AgentLink>, -} - -/// Read the agent's own free-text status and the timestamp when it was -/// set, directly from the `hyperhive-status` file in the state dir. -/// Mirrors `hive_c0re::container_view::read_agent_status` but runs -/// inside the agent container using its own state dir. -fn read_own_status() -> (Option<String>, Option<i64>) { - let path = crate::paths::state_dir().join("hyperhive-status"); - let meta = std::fs::metadata(&path).ok(); - let text = std::fs::read_to_string(&path) - .ok() - .as_deref() - .map(str::trim) - .filter(|t| !t.is_empty()) - .map(str::to_owned); - let mtime = meta.and_then(|m| { - m.modified().ok().and_then(|t| { - t.duration_since(std::time::UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) - }) - }); - if text.is_none() { - (None, None) - } else { - (text, mtime) - } -} - -async fn api_dashboard_state(State(state): State<AppState>) -> axum::Json<DashboardState> { - let (status_text, status_set_at) = read_own_status(); - let rate_limited = state.bus.is_rate_limited(); - let model = state.bus.model(); - let context_window_tokens = state - .bus - .api_context_window() - .unwrap_or_else(|| crate::events::context_window_tokens(&model)); - // Full context-window size = input + cache-read + cache-creation. Using - // raw `input_tokens` here reported only the *uncached* sliver, which is - // ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match - // the agent page (which ships the whole `ctx_usage` and sums it) and the - // cache-TTL logic in turn.rs, both of which use `context_tokens()`. - let ctx_tokens = state.bus.last_ctx_usage().map(|u| u.context_tokens()); - axum::Json(DashboardState { - status_text, - status_set_at, - ctx_tokens, - context_window_tokens, - rate_limited, - links: agent_links(&state.label, state.gui_vnc_port.is_some()), - }) -} - -/// Build the navigation link list for the agent page header. URLs -/// are paths (relative) for `Container`/`Forge` targets and absolute -/// for `External`; the frontend resolves each against its `kind` -/// against the right base so the backend never has to guess the -/// operator's browser host. See -/// [`docs/web-ui/dashboard.md::Container row`](../../../docs/web-ui/dashboard.md) for -/// the resolver + how `deployed:<sha>` ships alongside. -fn agent_links(label: &str, gui_enabled: bool) -> Vec<AgentLink> { - let mut links = Vec::new(); - - links.push(AgentLink { - url: "stats.html".to_owned(), - icon: "📊".to_owned(), - label: "stats".to_owned(), - kind: AgentLinkKind::Container, - }); - - if gui_enabled { - links.push(AgentLink { - url: "screen.html".to_owned(), - icon: "🖥".to_owned(), - label: "screen".to_owned(), - kind: AgentLinkKind::Container, - }); - } - - if crate::paths::state_dir().join("forge-token").is_file() { - links.push(AgentLink { - url: format!("/{label}"), - icon: "⬡".to_owned(), - label: "forge".to_owned(), - kind: AgentLinkKind::Forge, - }); - links.push(AgentLink { - url: format!("/agent-configs/{label}"), - icon: "↳".to_owned(), - label: "config".to_owned(), - kind: AgentLinkKind::Forge, - }); - } - - // Agent-declared extras (`hyperhive.dashboardLinks` → the - // `hive-dashboard-links` NixOS oneshot writes them to - // `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk - // is `{label, icon, url}` with absolute URLs — those become - // `kind = External` links, passed through verbatim. - let extras_path = crate::paths::state_dir().join("hyperhive-dashboard-links.json"); - if let Ok(text) = std::fs::read_to_string(&extras_path) - && !text.trim().is_empty() - && let Ok(extras) = serde_json::from_str::<Vec<ExtraLink>>(&text) - { - for e in extras { - links.push(AgentLink { - url: e.url, - icon: e.icon, - label: e.label, - kind: AgentLinkKind::External, - }); - } - } - - links -} - -/// On-disk shape of `hyperhive-dashboard-links.json` (the -/// `hive-dashboard-links` NixOS oneshot's output). Mapped to -/// `AgentLink { kind: External }` inside `agent_links`. -#[derive(serde::Deserialize)] -struct ExtraLink { - label: String, - #[serde(default)] - icon: String, - url: String, -} - -/// Best-effort: pull the last 30 messages addressed to us via the -/// per-agent / manager socket. Empty list on any transport / decode -/// failure — the inbox section is decorative, not authoritative. -async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> { - const LIMIT: u64 = 30; - // Deadline-bounded: `/api/state` must render even when hive-c0re is - // busy — an empty inbox section beats a hung snapshot. - match tokio::time::timeout( - SOCKET_FETCH_TIMEOUT, - client::request::<_, hive_sh4re::Response>( - socket, - &hive_sh4re::Request::Recent { limit: LIMIT }, - ), - ) - .await - { - Ok(Ok(hive_sh4re::Response::Recent { rows })) => rows, - _ => Vec::new(), - } -} - -/// Fetch reminder activity stats from the broker via the per-agent / -/// manager socket. Returns None on any transport / decode failure — the -/// stats are decorative, not authoritative. -async fn fetch_reminder_stats( - socket: &std::path::Path, - window_secs: u64, -) -> Option<hive_sh4re::ReminderStats> { - match tokio::time::timeout( - SOCKET_FETCH_TIMEOUT, - client::request::<_, hive_sh4re::Response>( - socket, - &hive_sh4re::Request::ReminderRollup { - since_secs: window_secs, - agent: None, - }, - ), - ) - .await - { - Ok(Ok(hive_sh4re::Response::ReminderRollup(stats))) => Some(stats), - _ => None, - } -} - -// --------------------------------------------------------------------------- -// Action handlers -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct SendForm { - body: String, -} - -async fn post_send(State(state): State<AppState>, Form(form): Form<SendForm>) -> Response { - let body = form.body.trim().to_owned(); - if body.is_empty() { - return error_response(StatusCode::BAD_REQUEST, "send: `body` required"); - } - let result = match tokio::time::timeout( - SOCKET_FETCH_TIMEOUT, - client::request::<_, hive_sh4re::Response>( - &state.socket, - &hive_sh4re::Request::OperatorMsg { body }, - ), - ) - .await - { - Ok(Ok(hive_sh4re::Response::Ok)) => Ok(()), - Ok(Ok(hive_sh4re::Response::Err { message })) => Err(message), - Ok(Ok(other)) => Err(format!("unexpected response: {other:?}")), - Ok(Err(e)) => Err(format!("transport: {e:#}")), - Err(_) => Err("timed out — hive-c0re busy, retry".to_owned()), - }; - match result { - // 200 instead of 303 → the client doesn't refetch /api/state. - // The operator message becomes a broker `Sent` (already shown - // server-side in the dashboard); on the agent side, the - // resulting `TurnStart` SSE event drives the terminal + the - // inbox row gets consumed by the time `TurnEnd` fires the - // existing turn-end refresh. - Ok(()) => (axum::http::StatusCode::OK, "ok").into_response(), - Err(e) => error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("send failed: {e}"), - ), - } -} - -/// Query params for the paginated history endpoint. -#[derive(Debug, Deserialize)] -struct HistoryParams { - /// Cursor: only return events with sqlite row id < `before`. - /// Omit for the initial (most-recent) page. - before: Option<i64>, - /// Page size (default 100, capped at `HISTORY_CAPACITY`). - limit: Option<usize>, -} - -async fn events_history( - State(state): State<AppState>, - axum::extract::Query(params): axum::extract::Query<HistoryParams>, -) -> axum::Json<serde_json::Value> { - use crate::events::HISTORY_CAPACITY; - let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY); - let before = params.before; - let is_initial = before.is_none(); - - // Capture seq *before* the read on initial loads so the SSE dedupe - // window is "drop buffered events you've already seen in history", - // never "lose an event that fired between the read and the seq." - // On paginated loads (`before` is set) seq is not needed. - let seq = if is_initial { - Some(state.bus.current_seq()) - } else { - None - }; - - let (events, min_id, has_more) = state.bus.history_page(before, limit); - let mut resp = serde_json::json!({ - "events": events, - "min_id": min_id, - "has_more": has_more, - }); - if let Some(s) = seq { - resp["seq"] = serde_json::json!(s); - } - axum::Json(resp) -} - -async fn events_stream( - State(state): State<AppState>, -) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { - tracing::info!("sse: client subscribed"); - let rx = state.bus.subscribe(); - // Drop a "hello" note into the bus so every new subscriber sees at - // least one event immediately and can clear the connecting placeholder. - state.bus.emit(crate::events::LiveEvent::Note { - text: "live stream attached".into(), - }); - let stream = BroadcastStream::new(rx).filter_map(|res| { - let ev = res.ok()?; - let json = serde_json::to_string(&ev).ok()?; - Some(Ok(Event::default().data(json))) - }); - Sse::new(stream).keep_alive(KeepAlive::default()) -} - -async fn post_login_start(State(state): State<AppState>) -> Response { - drop_if_finished(&state.session); - { - let guard = state.session.lock().unwrap(); - if guard.is_some() { - return (axum::http::StatusCode::OK, "ok").into_response(); - } - } - match LoginSession::start() { - Ok(session) => { - *state.session.lock().unwrap() = Some(Arc::new(session)); - // Flip status from needs_login_idle → needs_login_in_progress - // so the web UI's badge + polling kick in (polling is still - // the right tool for the streaming session output during - // the login flow itself; events drop the poll for - // *everything else*). - state.bus.emit_status("needs_login_in_progress"); - (axum::http::StatusCode::OK, "ok").into_response() - } - Err(e) => error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("login start failed: {e:#}"), - ), - } -} - -#[derive(Deserialize)] -struct CodeForm { - code: String, -} - -async fn post_login_code(State(state): State<AppState>, Form(form): Form<CodeForm>) -> Response { - let session = state.session.lock().unwrap().clone(); - let Some(session) = session else { - return error_response(StatusCode::CONFLICT, "no login session running"); - }; - if let Err(e) = session.submit_code(&form.code).await { - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("submit code failed: {e:#}"), - ); - } - (axum::http::StatusCode::OK, "ok").into_response() -} - -async fn post_login_cancel(State(state): State<AppState>) -> Response { - let session = state.session.lock().unwrap().take(); - if let Some(session) = session { - session.close_stdin().await; - session.kill(); - } - // Back to needs_login_idle (LoginState unchanged, session gone). - state.bus.emit_status("needs_login_idle"); - (axum::http::StatusCode::OK, "ok").into_response() -} - -#[derive(Deserialize)] -struct ModelForm { - model: String, -} - -/// Switch the model for future turns. The current turn (if any) -/// keeps its model; `/model <name>` applies starting with the next -/// `recv` cycle. Empty / whitespace-only inputs are rejected. No -/// claude-side validation — we just hand the string through to -/// `claude --model <name>`; an unknown model surfaces as a turn -/// failure in the live panel and the operator can revert. -async fn post_set_model(State(state): State<AppState>, Form(form): Form<ModelForm>) -> Response { - let name = form.model.trim(); - if name.is_empty() { - return error_response(StatusCode::BAD_REQUEST, "model: name required"); - } - state.bus.set_model(name); - state.bus.emit(crate::events::LiveEvent::Note { - text: format!("operator: /model — claude model set to '{name}' for future turns"), - }); - tracing::info!(%name, "operator set model"); - (axum::http::StatusCode::OK, "ok").into_response() -} - -#[derive(Deserialize)] -struct EffortForm { - effort: String, -} - -/// Switch the claude effort level for future sessions. Operator-only -/// (the dashboard picker POSTs here through the gateway). Validated -/// server-side against [`crate::events::EFFORT_LEVELS`] — an out-of-set -/// value is rejected rather than handed to `claude --effort`, since an -/// unknown level would fail every subsequent launch. Applies on the next -/// session start (no mid-session swap). -async fn post_set_effort(State(state): State<AppState>, Form(form): Form<EffortForm>) -> Response { - let level = form.effort.trim(); - if !crate::events::is_valid_effort(level) { - return error_response( - StatusCode::BAD_REQUEST, - &format!( - "effort: level must be one of {}", - crate::events::EFFORT_LEVELS.join(", ") - ), - ); - } - state.bus.set_effort(level); - state.bus.emit(crate::events::LiveEvent::Note { - text: format!("operator: /effort — claude effort set to '{level}' for future sessions"), - }); - tracing::info!(%level, "operator set effort"); - (axum::http::StatusCode::OK, "ok").into_response() -} - -/// Operator-initiated `/compact`. Deferred: sets the `compact_pending` flag -/// that `turn::drive_turn` consumes at the end of the current/next turn, so it -/// works while a turn is in flight (a mid-turn compaction would race the live -/// claude process) rather than only when the agent is idle. Returns 200 -/// immediately; the compaction stream lands in the live panel when it runs. -async fn post_compact(State(state): State<AppState>) -> Response { - state.bus.request_compact(); - state.bus.emit(crate::events::LiveEvent::Note { - text: "operator: /compact queued — runs at the end of the current turn".into(), - }); - (axum::http::StatusCode::OK, "ok").into_response() -} - -/// Request a session reset. The current session is archived (its backing -/// `<uuid>.jsonl` renamed out of claude's resolution glob) at the next turn -/// boundary, so the following turn's `--resume` misses and self-heals into a -/// freshly-named session. History is preserved on disk, not deleted. -/// -/// Deferred (a one-shot flag consumed by `drive_turn`) rather than applied -/// here: renaming the session file while a claude turn is mid-write would -/// race the live process. Between turns there is no open session file (one -/// claude per container, serialized by the serve loop), so the archive is -/// safe there. Useful when the session-resume context is poisoned (claude -/// went off the rails, hit an unrecoverable refusal, etc.) and a full reset -/// is cheaper than asking claude to forget mid-stream. -async fn post_new_session(State(state): State<AppState>) -> Response { - state.bus.request_session_reset(); - state.bus.emit(crate::events::LiveEvent::Note { - text: "operator: session reset queued — takes effect at the next turn".into(), - }); - (axum::http::StatusCode::OK, "ok").into_response() -} - -/// OAuth credential filenames inside `paths::claude_dir()`. Wiping -/// only these (and not the rest of `~/.claude/`) preserves session -/// history so `claude --continue` keeps working after a fresh login. -/// Rationale + the previous wholesale-wipe shape we replaced live in -/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md) -/// (the `/api/logout` bullet). -const CRED_FILE_NAMES: &[&str] = &[".credentials.json", "mcp-needs-auth-cache.json"]; - -/// Operator-driven `/logout`: SIGINT claude, delete the credential -/// files in `CRED_FILE_NAMES`, flip `LoginState::NeedsLogin`. The -/// turn loop's next iteration parks into `wait_for_login` which -/// resumes when a fresh credentials file appears via `/login/code`. -/// Always returns 200 with a body describing what happened. See -/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md) -/// (the `/api/logout` bullet) for the three-step rationale + -/// preservation invariants. -async fn post_logout(State(state): State<AppState>) -> Response { - // Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`). - let _ = tokio::process::Command::new("pkill") - .args(["-INT", "claude"]) - .output() - .await; - // Step 2: delete OAuth credential files only — preserve session - // history files alongside them. - let dir = crate::paths::claude_dir(); - let mut warnings: Vec<String> = Vec::new(); - let mut wiped: Vec<&str> = Vec::new(); - for name in CRED_FILE_NAMES { - let path = dir.join(name); - match tokio::fs::remove_file(&path).await { - Ok(()) => wiped.push(name), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => { - // Already gone — operator clicked /logout while - // already logged out, or the file simply didn't exist - // for this agent. Idempotent. - } - Err(e) => warnings.push(format!("{name}: {e}")), - } - } - let wipe_summary = if wiped.is_empty() { - "no credential files present (already logged out)".to_owned() - } else { - format!("wiped {}", wiped.join(", ")) - }; - let warn_suffix = if warnings.is_empty() { - String::new() - } else { - format!(" (warnings: {})", warnings.join("; ")) - }; - // Step 3: flip LoginState + emit Note. Turn loop sees the flip on - // its next iteration and parks into wait_for_login. - *state.login.lock().unwrap() = LoginState::NeedsLogin; - state.bus.emit(crate::events::LiveEvent::Note { - text: format!( - "operator: /logout — {wipe_summary} in {}{warn_suffix}", - dir.display() - ), - }); - state.bus.emit_status("needs_login_idle"); - ( - axum::http::StatusCode::OK, - format!("ok: {wipe_summary} in {}{warn_suffix}", dir.display()), - ) - .into_response() -} - -async fn post_cancel_turn(State(state): State<AppState>) -> Response { - let out = tokio::process::Command::new("pkill") - .args(["-INT", "claude"]) - .output() - .await; - let note = match out { - Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(), - Ok(o) if o.status.code() == Some(1) => { - "operator: /cancel — no claude process to interrupt".to_owned() - } - Ok(o) => format!( - "operator: /cancel — pkill exited {} stderr={}", - o.status, - String::from_utf8_lossy(&o.stderr).trim() - ), - Err(e) => format!("operator: /cancel — pkill failed: {e}"), - }; - state - .bus - .emit(crate::events::LiveEvent::Note { text: note }); - (axum::http::StatusCode::OK, "ok").into_response() -} - -fn error_response(status: StatusCode, message: &str) -> Response { - // Plain text — JS app surfaces in `alert()`, HTML wrapping would just - // be noise. Status is per-caller: 400 for bad input, 409 for a - // retryable state conflict (turn in flight / hive-c0re busy), 500 only - // for a genuine server/transport failure — the frontend shows the code - // in its alert, so a benign "busy, retry" must not read as a 500. - (status, message.to_owned()).into_response() -} - -/// Read `HIVE_AVAILABLE_MODELS` (comma-separated short names injected by -/// `services.hyperhive.availableModels`) and return the parsed list. -/// Falls back to `["haiku", "sonnet", "opus"]` when the env var is absent -/// or resolves to an empty list after trimming. -fn available_models() -> Vec<String> { - const DEFAULT: &[&str] = &["haiku", "sonnet", "opus"]; - let raw = match std::env::var("HIVE_AVAILABLE_MODELS") { - Ok(v) if !v.trim().is_empty() => v, - _ => return DEFAULT.iter().map(ToString::to_string).collect(), - }; - let models: Vec<String> = raw - .split(',') - .map(|s| s.trim().to_string()) - .filter(|s| !s.is_empty()) - .collect(); - if models.is_empty() { - DEFAULT.iter().map(ToString::to_string).collect() - } else { - models - } -} diff --git a/hive-ag3nt/src/web_ui/actions.rs b/hive-ag3nt/src/web_ui/actions.rs new file mode 100644 index 00000000..49fce257 --- /dev/null +++ b/hive-ag3nt/src/web_ui/actions.rs @@ -0,0 +1,172 @@ +//! Operator action POST handlers (send, cancel, compact, model, effort, reset). + +use axum::{ + Form, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::Deserialize; + +use crate::client; + +use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response}; + +#[derive(Deserialize)] +pub(super) struct SendForm { + body: String, +} + +pub(super) async fn post_send( + State(state): State<AppState>, + Form(form): Form<SendForm>, +) -> Response { + let body = form.body.trim().to_owned(); + if body.is_empty() { + return error_response(StatusCode::BAD_REQUEST, "send: `body` required"); + } + let result = match tokio::time::timeout( + SOCKET_FETCH_TIMEOUT, + client::request::<_, hive_sh4re::Response>( + &state.socket, + &hive_sh4re::Request::OperatorMsg { body }, + ), + ) + .await + { + Ok(Ok(hive_sh4re::Response::Ok)) => Ok(()), + Ok(Ok(hive_sh4re::Response::Err { message })) => Err(message), + Ok(Ok(other)) => Err(format!("unexpected response: {other:?}")), + Ok(Err(e)) => Err(format!("transport: {e:#}")), + Err(_) => Err("timed out — hive-c0re busy, retry".to_owned()), + }; + match result { + // 200 instead of 303 → the client doesn't refetch /api/state. + // The operator message becomes a broker `Sent` (already shown + // server-side in the dashboard); on the agent side, the + // resulting `TurnStart` SSE event drives the terminal + the + // inbox row gets consumed by the time `TurnEnd` fires the + // existing turn-end refresh. + Ok(()) => (axum::http::StatusCode::OK, "ok").into_response(), + Err(e) => error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("send failed: {e}"), + ), + } +} + +pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response { + let out = tokio::process::Command::new("pkill") + .args(["-INT", "claude"]) + .output() + .await; + let note = match out { + Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(), + Ok(o) if o.status.code() == Some(1) => { + "operator: /cancel — no claude process to interrupt".to_owned() + } + Ok(o) => format!( + "operator: /cancel — pkill exited {} stderr={}", + o.status, + String::from_utf8_lossy(&o.stderr).trim() + ), + Err(e) => format!("operator: /cancel — pkill failed: {e}"), + }; + state + .bus + .emit(crate::events::LiveEvent::Note { text: note }); + (axum::http::StatusCode::OK, "ok").into_response() +} + +/// Operator-initiated `/compact`. Deferred: sets the `compact_pending` flag +/// that `turn::drive_turn` consumes at the end of the current/next turn, so it +/// works while a turn is in flight (a mid-turn compaction would race the live +/// claude process) rather than only when the agent is idle. Returns 200 +/// immediately; the compaction stream lands in the live panel when it runs. +pub(super) async fn post_compact(State(state): State<AppState>) -> Response { + state.bus.request_compact(); + state.bus.emit(crate::events::LiveEvent::Note { + text: "operator: /compact queued — runs at the end of the current turn".into(), + }); + (axum::http::StatusCode::OK, "ok").into_response() +} + +/// Request a session reset. The current session is archived (its backing +/// `<uuid>.jsonl` renamed out of claude's resolution glob) at the next turn +/// boundary, so the following turn's `--resume` misses and self-heals into a +/// freshly-named session. History is preserved on disk, not deleted. +/// +/// Deferred (a one-shot flag consumed by `drive_turn`) rather than applied +/// here: renaming the session file while a claude turn is mid-write would +/// race the live process. Between turns there is no open session file (one +/// claude per container, serialized by the serve loop), so the archive is +/// safe there. Useful when the session-resume context is poisoned (claude +/// went off the rails, hit an unrecoverable refusal, etc.) and a full reset +/// is cheaper than asking claude to forget mid-stream. +pub(super) async fn post_new_session(State(state): State<AppState>) -> Response { + state.bus.request_session_reset(); + state.bus.emit(crate::events::LiveEvent::Note { + text: "operator: session reset queued — takes effect at the next turn".into(), + }); + (axum::http::StatusCode::OK, "ok").into_response() +} + +#[derive(Deserialize)] +pub(super) struct ModelForm { + model: String, +} + +/// Switch the model for future turns. The current turn (if any) +/// keeps its model; `/model <name>` applies starting with the next +/// `recv` cycle. Empty / whitespace-only inputs are rejected. No +/// claude-side validation — we just hand the string through to +/// `claude --model <name>`; an unknown model surfaces as a turn +/// failure in the live panel and the operator can revert. +pub(super) async fn post_set_model( + State(state): State<AppState>, + Form(form): Form<ModelForm>, +) -> Response { + let name = form.model.trim(); + if name.is_empty() { + return error_response(StatusCode::BAD_REQUEST, "model: name required"); + } + state.bus.set_model(name); + state.bus.emit(crate::events::LiveEvent::Note { + text: format!("operator: /model — claude model set to '{name}' for future turns"), + }); + tracing::info!(%name, "operator set model"); + (axum::http::StatusCode::OK, "ok").into_response() +} + +#[derive(Deserialize)] +pub(super) struct EffortForm { + effort: String, +} + +/// Switch the claude effort level for future sessions. Operator-only +/// (the dashboard picker POSTs here through the gateway). Validated +/// server-side against [`crate::events::EFFORT_LEVELS`] — an out-of-set +/// value is rejected rather than handed to `claude --effort`, since an +/// unknown level would fail every subsequent launch. Applies on the next +/// session start (no mid-session swap). +pub(super) async fn post_set_effort( + State(state): State<AppState>, + Form(form): Form<EffortForm>, +) -> Response { + let level = form.effort.trim(); + if !crate::events::is_valid_effort(level) { + return error_response( + StatusCode::BAD_REQUEST, + &format!( + "effort: level must be one of {}", + crate::events::EFFORT_LEVELS.join(", ") + ), + ); + } + state.bus.set_effort(level); + state.bus.emit(crate::events::LiveEvent::Note { + text: format!("operator: /effort — claude effort set to '{level}' for future sessions"), + }); + tracing::info!(%level, "operator set effort"); + (axum::http::StatusCode::OK, "ok").into_response() +} diff --git a/hive-ag3nt/src/web_ui/auth.rs b/hive-ag3nt/src/web_ui/auth.rs new file mode 100644 index 00000000..f3e175e0 --- /dev/null +++ b/hive-ag3nt/src/web_ui/auth.rs @@ -0,0 +1,141 @@ +//! Login / logout flow handlers (`/login/*`, `/api/logout`). + +use std::sync::Arc; + +use axum::{ + Form, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::Deserialize; + +use crate::login::LoginState; +use crate::login_session::{LoginSession, drop_if_finished}; + +use super::{AppState, error_response}; + +pub(super) async fn post_login_start(State(state): State<AppState>) -> Response { + drop_if_finished(&state.session); + { + let guard = state.session.lock().unwrap(); + if guard.is_some() { + return (axum::http::StatusCode::OK, "ok").into_response(); + } + } + match LoginSession::start() { + Ok(session) => { + *state.session.lock().unwrap() = Some(Arc::new(session)); + // Flip status from needs_login_idle → needs_login_in_progress + // so the web UI's badge + polling kick in (polling is still + // the right tool for the streaming session output during + // the login flow itself; events drop the poll for + // *everything else*). + state.bus.emit_status("needs_login_in_progress"); + (axum::http::StatusCode::OK, "ok").into_response() + } + Err(e) => error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("login start failed: {e:#}"), + ), + } +} + +#[derive(Deserialize)] +pub(super) struct CodeForm { + code: String, +} + +pub(super) async fn post_login_code( + State(state): State<AppState>, + Form(form): Form<CodeForm>, +) -> Response { + let session = state.session.lock().unwrap().clone(); + let Some(session) = session else { + return error_response(StatusCode::CONFLICT, "no login session running"); + }; + if let Err(e) = session.submit_code(&form.code).await { + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("submit code failed: {e:#}"), + ); + } + (axum::http::StatusCode::OK, "ok").into_response() +} + +pub(super) async fn post_login_cancel(State(state): State<AppState>) -> Response { + let session = state.session.lock().unwrap().take(); + if let Some(session) = session { + session.close_stdin().await; + session.kill(); + } + // Back to needs_login_idle (LoginState unchanged, session gone). + state.bus.emit_status("needs_login_idle"); + (axum::http::StatusCode::OK, "ok").into_response() +} + +/// OAuth credential filenames inside `paths::claude_dir()`. Wiping +/// only these (and not the rest of `~/.claude/`) preserves session +/// history so `claude --continue` keeps working after a fresh login. +/// Rationale + the previous wholesale-wipe shape we replaced live in +/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md) +/// (the `/api/logout` bullet). +const CRED_FILE_NAMES: &[&str] = &[".credentials.json", "mcp-needs-auth-cache.json"]; + +/// Operator-driven `/logout`: SIGINT claude, delete the credential +/// files in `CRED_FILE_NAMES`, flip `LoginState::NeedsLogin`. The +/// turn loop's next iteration parks into `wait_for_login` which +/// resumes when a fresh credentials file appears via `/login/code`. +/// Always returns 200 with a body describing what happened. See +/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md) +/// (the `/api/logout` bullet) for the three-step rationale + +/// preservation invariants. +pub(super) async fn post_logout(State(state): State<AppState>) -> Response { + // Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`). + let _ = tokio::process::Command::new("pkill") + .args(["-INT", "claude"]) + .output() + .await; + // Step 2: delete OAuth credential files only — preserve session + // history files alongside them. + let dir = crate::paths::claude_dir(); + let mut warnings: Vec<String> = Vec::new(); + let mut wiped: Vec<&str> = Vec::new(); + for name in CRED_FILE_NAMES { + let path = dir.join(name); + match tokio::fs::remove_file(&path).await { + Ok(()) => wiped.push(name), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + // Already gone — operator clicked /logout while + // already logged out, or the file simply didn't exist + // for this agent. Idempotent. + } + Err(e) => warnings.push(format!("{name}: {e}")), + } + } + let wipe_summary = if wiped.is_empty() { + "no credential files present (already logged out)".to_owned() + } else { + format!("wiped {}", wiped.join(", ")) + }; + let warn_suffix = if warnings.is_empty() { + String::new() + } else { + format!(" (warnings: {})", warnings.join("; ")) + }; + // Step 3: flip LoginState + emit Note. Turn loop sees the flip on + // its next iteration and parks into wait_for_login. + *state.login.lock().unwrap() = LoginState::NeedsLogin; + state.bus.emit(crate::events::LiveEvent::Note { + text: format!( + "operator: /logout — {wipe_summary} in {}{warn_suffix}", + dir.display() + ), + }); + state.bus.emit_status("needs_login_idle"); + ( + axum::http::StatusCode::OK, + format!("ok: {wipe_summary} in {}{warn_suffix}", dir.display()), + ) + .into_response() +} diff --git a/hive-ag3nt/src/web_ui/mod.rs b/hive-ag3nt/src/web_ui/mod.rs new file mode 100644 index 00000000..63c25e26 --- /dev/null +++ b/hive-ag3nt/src/web_ui/mod.rs @@ -0,0 +1,267 @@ +//! Per-container HTTP UI. SPA shape: `GET /` returns a static shell; +//! `GET /static/*` serves CSS + JS; `GET /api/state` returns the page +//! state as JSON; the JS app renders. Live events stream on +//! `/events/stream`. Action POSTs (`/send`, `/login/*`) return either a +//! 303 Redirect (for browsers that submit the form normally) or just +//! 200 OK — the JS app re-fetches `/api/state` afterwards. +//! +//! Handlers are split by concern into the submodules below; this file owns the +//! shared [`AppState`], the listener + router wiring in [`serve`], and a couple +//! of small shared helpers ([`error_response`], [`SOCKET_FETCH_TIMEOUT`]). + +mod actions; +mod auth; +mod screen; +mod state; +mod stats; +mod stream; + +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context, Result}; +use axum::{ + Router, + http::StatusCode, + response::{IntoResponse, Response}, + routing::{get, post}, +}; +use tower_http::services::ServeDir; + +use crate::events::Bus; +use crate::login::LoginState; +use crate::login_session::LoginSession; + +/// Deadline for broker-backed fetches on web-UI request paths. The +/// page's critical fields (status, turn state, usage) are all +/// in-memory; a busy or stalled hive-c0re must degrade the +/// socket-backed extras (inbox rows, loose ends, reminder stats) +/// instead of hanging the whole response — an unbounded await here is +/// what let `/api/state` stall long enough to bork the terminal. +const SOCKET_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3); + +/// Live login state for the web UI. The harness updates this in place as it +/// transitions between `NeedsLogin` and `Online`; the UI reads on each +/// render. +pub type LoginStateCell = Arc<Mutex<LoginState>>; + +#[derive(Clone)] +struct AppState { + label: String, + login: LoginStateCell, + session: Arc<Mutex<Option<Arc<LoginSession>>>>, + bus: Bus, + socket: PathBuf, + /// VNC port from the `HIVE_GUI_VNC_PORT` env var at startup. + /// `None` when unset (gui not enabled for this agent). + gui_vnc_port: Option<u16>, +} + +/// Bind the per-container web listener and serve the SPA. +/// +/// `HIVE_WEB_SOCKET` opt-in selects unix-socket vs TCP binding; the +/// dual-mode transition + gateway-side consumer live in +/// [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md) and +/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md). +/// +/// # Errors +/// +/// Returns an error if neither the TCP listener (default) nor the +/// unix-socket bind (`HIVE_WEB_SOCKET`, if set) can be acquired, or +/// if `HIVE_STATIC_DIR` is missing. +pub async fn serve( + label: String, + port: u16, + login: LoginStateCell, + bus: Bus, + socket: PathBuf, +) -> Result<()> { + let gui_vnc_port = read_gui_vnc_port(); + let static_dir: PathBuf = std::env::var_os("HIVE_STATIC_DIR") + .map(PathBuf::from) + .context( + "HIVE_STATIC_DIR env var not set — point it at the merged \ + per-agent dist (see hyperhive.frontend.mergedDist in nix)", + )?; + if !static_dir.is_dir() { + anyhow::bail!( + "HIVE_STATIC_DIR ({}) is not a directory", + static_dir.display() + ); + } + tracing::info!(static_dir = %static_dir.display(), "web UI static dir resolved"); + let state = AppState { + label, + login, + session: Arc::new(Mutex::new(None)), + bus, + socket, + gui_vnc_port, + }; + let app = Router::new() + .route("/api/state", get(state::api_state)) + .route("/api/dashboard-state", get(state::api_dashboard_state)) + .route("/events/stream", get(stream::events_stream)) + .route("/events/history", get(stream::events_history)) + .route("/send", post(actions::post_send)) + .route("/login/start", post(auth::post_login_start)) + .route("/login/code", post(auth::post_login_code)) + .route("/login/cancel", post(auth::post_login_cancel)) + .route("/api/cancel", post(actions::post_cancel_turn)) + .route("/api/compact", post(actions::post_compact)) + .route("/api/model", post(actions::post_set_model)) + .route("/api/effort", post(actions::post_set_effort)) + .route("/api/new-session", post(actions::post_new_session)) + .route("/api/logout", post(auth::post_logout)) + .route("/api/loose-ends", get(stats::api_loose_ends)) + .route("/api/bash-tasks", get(stats::api_bash_tasks)) + .route("/api/stats", get(stats::api_stats)) + .route("/screen/ws", get(screen::screen_ws)) + .route("/icon", get(screen::serve_icon)) + // Anything else (`/`, `/stats`, `/screen`, `/static/*`) + // falls through to the merged dist. ServeDir auto-appends + // `.html` when the URL is a bare path that matches a file + // (so `/stats` → `dist/stats.html`, `/screen` → `dist/ + // screen.html`). Per-agent `extraFiles` additions are + // already layered into this same directory (see + // hyperhive.frontend.mergedDist in nix). + .fallback_service(ServeDir::new(&static_dir)) + .with_state(state); + // `HIVE_WEB_SOCKET` opt-in: when set + non-empty, bind a + // `UnixListener` at the given path. Empty string treated as + // unset so a stray `HIVE_WEB_SOCKET=` doesn't trap us into an + // un-bindable empty path. Falls through to the TCP path below + // otherwise. See docs/gateway.md::Per-agent unix-socket upstream + // for the gateway-side consumer. + if let Some(socket_path) = std::env::var_os("HIVE_WEB_SOCKET") + && !socket_path.is_empty() + { + let path = PathBuf::from(socket_path); + let listener = bind_unix(&path)?; + tracing::info!(socket = %path.display(), "web UI listening on unix socket"); + axum::serve(listener, app).await?; + return Ok(()); + } + let addr = SocketAddr::from(([0, 0, 0, 0], port)); + let listener = bind_with_retry(addr, "web UI").await?; + tracing::info!(%port, "web UI listening on tcp"); + axum::serve(listener, app).await?; + Ok(()) +} + +/// Bind a `UnixListener` at `path` and drop a `.bound` marker next +/// to it so c0re's gateway-map writer knows the socket is live. +/// Best-effort unlinks any stale socket left from a crashed previous +/// harness (clean exit removes it, but `bind(2)` refuses to overwrite +/// an existing file) and `mkdir -p`s the parent for first-boot. Mode +/// `0o666` — world-accessible so the gateway container's nginx process +/// can `connect(2)` without sharing a group with the agent user. +/// The per-agent subdir (`/run/hive-agent/<name>/`) is only accessible +/// to containers that have it bind-mounted, so world-accessible sockets +/// are not a material risk. +/// +/// Marker-gating + the gateway-side consumer: see +/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md). +fn bind_unix(path: &Path) -> Result<tokio::net::UnixListener> { + use std::os::unix::fs::PermissionsExt; + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create socket parent dir {}", parent.display()))?; + } + // Best-effort: ENOENT is fine (no stale file); any other error + // surfaces via the bind below with a clearer "AddrInUse" / perms + // message than a partial cleanup would. + let _ = std::fs::remove_file(path); + let listener = tokio::net::UnixListener::bind(path) + .with_context(|| format!("bind unix socket at {}", path.display()))?; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666)) + .with_context(|| format!("set perms on {}", path.display()))?; + // Best-effort ready marker: failed write isn't fatal (the harness + // still binds + serves), it just means the gateway side keeps the + // TCP upstream for one more sync tick. + if let Some(parent) = path.parent() { + let marker = parent.join("hyperhive-socket-bound"); + if let Err(e) = std::fs::write(&marker, b"") { + tracing::warn!( + marker = %marker.display(), error = %e, + "failed to write hyperhive-socket-bound marker — gateway may keep TCP upstream" + ); + } + } + Ok(listener) +} + +/// Bind a TCP listener with `SO_REUSEADDR` set, retrying on +/// `AddrInUse` indefinitely with exponential backoff capped at 2s. +/// First 12 attempts log at WARN; subsequent attempts log at INFO so +/// a long-held stale socket doesn't flood the journal. +/// +/// Uncapped retry + dashboard-banner-on-real-collision rationale: +/// see [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md). +async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result<tokio::net::TcpListener> { + let mut delay_ms = 250u64; + let mut attempts = 0u32; + loop { + match try_bind(addr) { + Ok(l) => { + if attempts > 0 { + tracing::info!( + %addr, attempts, + "{label}: bind succeeded after retry" + ); + } + return Ok(l); + } + Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { + let attempt = attempts + 1; + if attempt <= 12 { + tracing::warn!( + %addr, attempt, + "{label}: AddrInUse, retrying in {delay_ms}ms" + ); + } else { + tracing::info!( + %addr, attempt, + "{label}: AddrInUse still holding, retrying in {delay_ms}ms" + ); + } + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + attempts += 1; + delay_ms = (delay_ms * 2).min(2000); + } + Err(e) => { + return Err(e).with_context(|| format!("bind {label} on {addr}")); + } + } + } +} + +fn try_bind(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> { + let sock = match addr { + SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?, + SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?, + }; + sock.set_reuseaddr(true)?; + sock.bind(addr)?; + sock.listen(1024) +} + +/// The fixed VNC port weston bound, from the `HIVE_GUI_VNC_PORT` env var +/// the harness service sets when gui is enabled (see weston-vnc.nix). +/// `None` when unset (gui not enabled for this agent) or unparseable. +/// The port is a fixed, container-local value — no per-agent hashing, no +/// marker file — because network isolation is unconditional (each agent +/// has its own netns, so the port can't collide across containers). +fn read_gui_vnc_port() -> Option<u16> { + std::env::var("HIVE_GUI_VNC_PORT").ok()?.parse().ok() +} + +fn error_response(status: StatusCode, message: &str) -> Response { + // Plain text — JS app surfaces in `alert()`, HTML wrapping would just + // be noise. Status is per-caller: 400 for bad input, 409 for a + // retryable state conflict (turn in flight / hive-c0re busy), 500 only + // for a genuine server/transport failure — the frontend shows the code + // in its alert, so a benign "busy, retry" must not read as a 500. + (status, message.to_owned()).into_response() +} diff --git a/hive-ag3nt/src/web_ui/screen.rs b/hive-ag3nt/src/web_ui/screen.rs new file mode 100644 index 00000000..e8a9babc --- /dev/null +++ b/hive-ag3nt/src/web_ui/screen.rs @@ -0,0 +1,96 @@ +//! VNC screen websocket relay + agent icon. + +use axum::{ + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use super::AppState; + +/// This agent's icon. Serves the operator-configured SVG from +/// `/etc/hyperhive/icon.svg` (set via the `hyperhive.icon` agent.nix +/// option) when present, otherwise the bundled default hyperhive logo. +/// Always returns an image, so consumers (dashboard, favicon) can hit +/// `/icon` unconditionally without probing whether one is configured. +pub(super) async fn serve_icon() -> impl IntoResponse { + // Per-agent icon overrides go through `/etc/hyperhive/icon.svg` + // (set via the `hyperhive.icon` agent.nix option); the bundled + // default is resolved at runtime from + // `$HIVE_ASSETS_DIR/branding/hyperhive.svg`. If neither file can + // be read we serve an empty body — keeps the response a valid SVG + // content-type without a panic on a misconfigured container. + let body = std::fs::read_to_string("/etc/hyperhive/icon.svg").unwrap_or_else(|_| { + std::fs::read_to_string(hive_sh4re::assets::branding_svg()).unwrap_or_default() + }); + ([("content-type", "image/svg+xml")], body) +} + +/// WebSocket handler: upgrade then pump bytes between the WS client and +/// the VNC server on `127.0.0.1:<vnc_port>`. Returns 404 when gui is not +/// enabled for this agent. +pub(super) async fn screen_ws( + ws: axum::extract::ws::WebSocketUpgrade, + State(state): State<AppState>, +) -> Response { + let Some(vnc_port) = state.gui_vnc_port else { + return (StatusCode::NOT_FOUND, "gui not enabled for this agent").into_response(); + }; + ws.on_upgrade(move |socket| relay_ws_vnc(socket, vnc_port)) +} + +/// Pure byte pump: forwards raw bytes between the WebSocket client and +/// the VNC TCP stream. Transparent to any RFB variant (plain, `VeNCrypt`). +async fn relay_ws_vnc(socket: axum::extract::ws::WebSocket, vnc_port: u16) { + // Import futures traits locally so they don't conflict with + // tokio_stream::StreamExt used at module scope. + use axum::extract::ws::Message; + use futures_util::{SinkExt, StreamExt as _}; + + let addr = format!("127.0.0.1:{vnc_port}"); + let Ok(tcp) = tokio::net::TcpStream::connect(&addr).await else { + tracing::warn!(%addr, "screen/ws: could not connect to VNC server"); + return; + }; + let (mut tcp_rx, mut tcp_tx) = tcp.into_split(); + let (mut ws_tx, mut ws_rx) = socket.split(); + + // WS → TCP + let ws_to_tcp = tokio::spawn(async move { + while let Some(Ok(msg)) = futures_util::StreamExt::next(&mut ws_rx).await { + match msg { + Message::Binary(data) if tcp_tx.write_all(&data).await.is_err() => { + break; + } + Message::Close(_) => break, + _ => {} // ping/pong/text: ignore + } + } + }); + + // TCP → WS + let tcp_to_ws = tokio::spawn(async move { + let mut buf = vec![0u8; 8192]; + loop { + match tcp_rx.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + if ws_tx + .send(Message::Binary(buf[..n].to_vec().into())) + .await + .is_err() + { + break; + } + } + } + } + }); + + // Wait for either direction to close, then let both tasks drop. + tokio::select! { + _ = ws_to_tcp => {} + _ = tcp_to_ws => {} + } +} diff --git a/hive-ag3nt/src/web_ui/state.rs b/hive-ag3nt/src/web_ui/state.rs new file mode 100644 index 00000000..f67ea0e2 --- /dev/null +++ b/hive-ag3nt/src/web_ui/state.rs @@ -0,0 +1,434 @@ +//! `/api/state` + `/api/dashboard-state` snapshot builders. + +use axum::extract::State; +use serde::Serialize; + +use crate::client; +use crate::login::LoginState; +use crate::login_session::drop_if_finished; + +use super::{AppState, SOCKET_FETCH_TIMEOUT}; + +pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> { + // Capture seq *before* any reads so the dedupe contract is + // "events with seq > snapshot.seq are post-snapshot, never missed." + let seq = state.bus.current_seq(); + drop_if_finished(&state.session); + let login = *state.login.lock().unwrap(); + let session_snapshot = state.session.lock().unwrap().clone(); + let (status, session_view) = match (login, session_snapshot) { + (LoginState::Online, _) if state.bus.is_rate_limited() => ("rate_limited", None), + (LoginState::Online, _) => ("online", None), + (LoginState::NeedsLogin, None) => ("needs_login_idle", None), + (LoginState::NeedsLogin, Some(s)) => ( + "needs_login_in_progress", + Some(SessionView { + url: s.url(), + output: s.output(), + finished: s.finished(), + exit_note: s.exit_note(), + }), + ), + }; + let dashboard_port = std::env::var("HIVE_DASHBOARD_PORT") + .ok() + .and_then(|s| s.parse::<u16>().ok()) + .unwrap_or(7000); + let inbox = recent_inbox(&state.socket).await; + let (turn_state, turn_state_since) = state.bus.state_snapshot(); + let model = state.bus.model(); + let context_window_tokens = state + .bus + .api_context_window() + .unwrap_or_else(|| crate::events::context_window_tokens(&model)); + let ctx_usage = state.bus.last_ctx_usage(); + let cost_usage = state.bus.last_cost_usage(); + let effort = state.bus.effort(); + axum::Json(StateSnapshot { + seq, + label: state.label.clone(), + qualified_label: crate::identity::qualify(&state.label), + dashboard_port, + status, + session: session_view, + inbox, + turn_state, + turn_state_since, + model, + context_window_tokens, + ctx_usage, + cost_usage, + links: agent_links(&state.label, state.gui_vnc_port.is_some()), + forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") + .ok() + .filter(|s| !s.is_empty()), + hive_name: crate::identity::hive_name(), + swarm_name: crate::identity::swarm_name(), + available_models: available_models(), + effort, + available_efforts: crate::events::EFFORT_LEVELS + .iter() + .map(ToString::to_string) + .collect(), + }) +} + +/// Lean snapshot of the agent-owned fields that the dashboard card +/// needs. Served at `GET /api/dashboard-state` (accessible through the +/// gateway at `/agent/<name>/api/dashboard-state`). The dashboard +/// fetches this once per running agent to get fresh, agent-authoritative +/// values instead of relying on hive-c0re's periodic file-reads. +/// +/// Structural fields (running, `needs_update`, `deployed_sha`, parent, …) +/// continue to come from hive-c0re's `/api/state`; this endpoint covers +/// only the fields the agent itself is the source of truth for. +pub(super) async fn api_dashboard_state( + State(state): State<AppState>, +) -> axum::Json<DashboardState> { + let (status_text, status_set_at) = read_own_status(); + let rate_limited = state.bus.is_rate_limited(); + let model = state.bus.model(); + let context_window_tokens = state + .bus + .api_context_window() + .unwrap_or_else(|| crate::events::context_window_tokens(&model)); + // Full context-window size = input + cache-read + cache-creation. Using + // raw `input_tokens` here reported only the *uncached* sliver, which is + // ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match + // the agent page (which ships the whole `ctx_usage` and sums it) and the + // cache-TTL logic in turn.rs, both of which use `context_tokens()`. + let ctx_tokens = state.bus.last_ctx_usage().map(|u| u.context_tokens()); + axum::Json(DashboardState { + status_text, + status_set_at, + ctx_tokens, + context_window_tokens, + rate_limited, + links: agent_links(&state.label, state.gui_vnc_port.is_some()), + }) +} + +#[derive(Serialize)] +pub(super) struct StateSnapshot { + /// Bus seq at the moment this snapshot was assembled. Clients dedupe + /// their buffered SSE traffic against this value: events with + /// `seq <= snapshot.seq` are already reflected (or pre-date the + /// snapshot); `seq > snapshot.seq` is post-snapshot. Reset to 0 on + /// harness restart — clients treat reconnect as a fresh world. + seq: u64, + label: String, + /// Hive-qualified long name (`${label}@${hyperhive.domain}`) when + /// the host has been configured for a multi-hive swarm; falls back + /// to the short label when the hive domain env var is unset. + /// The frontend uses this for the page title / agent self-introduction; + /// when it equals `label`, the page renders the short form unchanged. + qualified_label: String, + dashboard_port: u16, + /// `"online"` | `"rate_limited"` | `"needs_login_idle"` | `"needs_login_in_progress"`. + status: &'static str, + /// Present when `status == "needs_login_in_progress"`. + session: Option<SessionView>, + /// Last N messages addressed to this agent, newest-first. Pulled + /// from the broker via the per-agent socket on each render. + /// Empty on transport failure. + inbox: Vec<hive_sh4re::InboxRow>, + /// Authoritative turn-loop state from the harness and the unix + /// timestamp the state was entered. The JS computes the age + /// client-side off this rather than tracking it from SSE events. + turn_state: crate::events::TurnState, + turn_state_since: i64, + /// Currently-active claude model name. Reflected on the page so + /// the operator can see what they just switched to (and what's + /// in flight). Mutable at runtime via `POST /api/model`. + model: String, + /// Effective context-window token budget for the current model. + /// Primary source: API-reported `modelUsage.*.contextWindow` from + /// the last result event (authoritative per-inference active window). + /// Falls back to `HIVE_CONTEXT_WINDOW_TOKENS_*` env vars, then 200 000. + /// Consumers (e.g. dashboard badge) use this to render ctx-usage %. + context_window_tokens: u64, + /// Last-inference token usage from the most recent completed + /// turn — represents the current context-window size at turn-end. + /// `null` until the first turn finishes. + ctx_usage: Option<hive_claude::TokenUsage>, + /// Cumulative token usage across the most recent turn's inferences + /// (cost signal). `null` until the first turn finishes. + cost_usage: Option<hive_claude::TokenUsage>, + /// Navigation links for this agent page. Also served via + /// `DashboardState.links` (`GET /api/dashboard-state`) for the + /// dashboard card's icon strip. Both are produced by `agent_links()` + /// — single source of truth. See [`docs/web-ui/dashboard.md::Container row`] + /// for the frontend resolver + which links appear in which conditions. + links: Vec<AgentLink>, + /// Public URL of the forge served by hive-gateway (e.g. + /// `"https://forge.pr1ma.darkest.space"`). Sourced from + /// `HIVE_FORGE_PUBLIC_URL`; `None` when `forge.behindGateway=false` + /// or the env var is absent. The frontend uses this to build forge + /// nav-strip links instead of hardcoding `<hostname>:3000`. + forge_public_url: Option<String>, + /// Human name of this hive instance (e.g. `"pr1ma"`). Sourced + /// from `HYPERHIVE_HIVE_NAME`; `None` when unset. The frontend + /// uses this for the page `<title>` and header label so browser + /// tabs disambiguate when multiple hives are open in parallel. + hive_name: Option<String>, + /// Human name of the swarm (e.g. `"constellat1on"`). Sourced from + /// `HYPERHIVE_SWARM_NAME`; `None` when unset. + swarm_name: Option<String>, + /// Ordered list of model short-names the operator has declared as + /// available on this hive. Sourced from `HIVE_AVAILABLE_MODELS` + /// (comma-separated, set by `services.hyperhive.availableModels`). + /// Falls back to `["haiku", "sonnet", "opus"]` when the env var is + /// absent or empty. The frontend model quick-picker renders one button + /// per entry in this list, so operators can add new models or drop + /// ones they don't want without touching the frontend code. + available_models: Vec<String>, + /// Currently-active claude effort level. Reflected on the page so the + /// operator's effort picker shows the live selection. Mutable at + /// runtime via `POST /api/effort`; applies on the next session. + effort: String, + /// Selectable effort levels for the picker, ascending. Fixed set + /// (`low`, `medium`, `high`, `xhigh`, `max`) — sourced from + /// [`crate::events::EFFORT_LEVELS`], not operator-configurable like + /// `available_models`. The frontend renders one button per entry. + available_efforts: Vec<String>, +} + +#[derive(Serialize)] +struct SessionView { + /// First `https://…` claude emitted on stdout, if any. + url: Option<String>, + /// Accumulated stdout + stderr. + output: String, + finished: bool, + exit_note: Option<String>, +} + +/// One navigation link in the agent page header row. The same JSON +/// shape appears in both `StateSnapshot.links` (`GET /api/state`, +/// per-agent page) and `DashboardState.links` (`GET /api/dashboard-state`, +/// dashboard card icon strip). `agent_links()` is the single source +/// of truth for what links an agent exposes. +#[derive(Serialize)] +struct AgentLink { + /// `kind = Container | Forge` → path; `kind = External` → full URL. + /// The frontend prepends the right base before rendering. + url: String, + icon: String, + label: String, + kind: AgentLinkKind, +} + +/// Resolution hint for `AgentLink.url`. The agent backend can't know +/// which hostname the browser sees (especially when the dashboard +/// proxies the call from a different origin), so it labels each link +/// and lets the frontend prepend the right base. +#[derive(Serialize, Clone, Copy)] +#[serde(rename_all = "snake_case")] +enum AgentLinkKind { + /// `url` is a path on the agent's container web UI (`/stats`, + /// `/screen`). Agent page: same-origin path. Dashboard: + /// `http://<host>:<container.port><url>`. + Container, + /// `url` is a path on the local Forgejo (`/<label>`, + /// `/agent-configs/<label>`). Both surfaces: + /// `http://<host>:3000<url>`. + Forge, + /// `url` is already a fully-qualified absolute URL — use as-is. + /// Agent-declared `hyperhive.dashboardLinks` extras arrive here. + External, +} + +#[derive(serde::Serialize)] +pub(super) struct DashboardState { + /// Free-text status set by `set_status`, read directly from the + /// `hyperhive-status` file the harness writes. `None` when unset. + #[serde(skip_serializing_if = "Option::is_none")] + status_text: Option<String>, + /// Unix timestamp (seconds) when the status file was last written. + /// `None` when no status is set. + #[serde(skip_serializing_if = "Option::is_none")] + status_set_at: Option<i64>, + /// Full context-window size from the most recent completed turn + /// (`ctx_usage.context_tokens()` = input + cache-read + cache-creation). + /// `None` until the first turn finishes. Drives the `ctx·Nk` card badge. + #[serde(skip_serializing_if = "Option::is_none")] + ctx_tokens: Option<u64>, + /// Effective context-window budget for the current model. Same + /// derivation as `StateSnapshot::context_window_tokens`. + context_window_tokens: u64, + /// True while the harness is parked after a rate-limit response. + rate_limited: bool, + /// Navigation links for the dashboard card's icon strip. This is + /// the authoritative source — includes the screen link (GUI agents) + /// which hive-c0re's disk-based fallback cannot determine. + links: Vec<AgentLink>, +} + +/// Read the agent's own free-text status and the timestamp when it was +/// set, directly from the `hyperhive-status` file in the state dir. +/// Mirrors `hive_c0re::container_view::read_agent_status` but runs +/// inside the agent container using its own state dir. +fn read_own_status() -> (Option<String>, Option<i64>) { + let path = crate::paths::state_dir().join("hyperhive-status"); + let meta = std::fs::metadata(&path).ok(); + let text = std::fs::read_to_string(&path) + .ok() + .as_deref() + .map(str::trim) + .filter(|t| !t.is_empty()) + .map(str::to_owned); + let mtime = meta.and_then(|m| { + m.modified().ok().and_then(|t| { + t.duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + }) + }); + if text.is_none() { + (None, None) + } else { + (text, mtime) + } +} + +/// Build the navigation link list for the agent page header. URLs +/// are paths (relative) for `Container`/`Forge` targets and absolute +/// for `External`; the frontend resolves each against its `kind` +/// against the right base so the backend never has to guess the +/// operator's browser host. See +/// [`docs/web-ui/dashboard.md::Container row`](../../../docs/web-ui/dashboard.md) for +/// the resolver + how `deployed:<sha>` ships alongside. +fn agent_links(label: &str, gui_enabled: bool) -> Vec<AgentLink> { + let mut links = Vec::new(); + + links.push(AgentLink { + url: "stats.html".to_owned(), + icon: "📊".to_owned(), + label: "stats".to_owned(), + kind: AgentLinkKind::Container, + }); + + if gui_enabled { + links.push(AgentLink { + url: "screen.html".to_owned(), + icon: "🖥".to_owned(), + label: "screen".to_owned(), + kind: AgentLinkKind::Container, + }); + } + + if crate::paths::state_dir().join("forge-token").is_file() { + links.push(AgentLink { + url: format!("/{label}"), + icon: "⬡".to_owned(), + label: "forge".to_owned(), + kind: AgentLinkKind::Forge, + }); + links.push(AgentLink { + url: format!("/agent-configs/{label}"), + icon: "↳".to_owned(), + label: "config".to_owned(), + kind: AgentLinkKind::Forge, + }); + } + + // Agent-declared extras (`hyperhive.dashboardLinks` → the + // `hive-dashboard-links` NixOS oneshot writes them to + // `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk + // is `{label, icon, url}` with absolute URLs — those become + // `kind = External` links, passed through verbatim. + let extras_path = crate::paths::state_dir().join("hyperhive-dashboard-links.json"); + if let Ok(text) = std::fs::read_to_string(&extras_path) + && !text.trim().is_empty() + && let Ok(extras) = serde_json::from_str::<Vec<ExtraLink>>(&text) + { + for e in extras { + links.push(AgentLink { + url: e.url, + icon: e.icon, + label: e.label, + kind: AgentLinkKind::External, + }); + } + } + + links +} + +/// On-disk shape of `hyperhive-dashboard-links.json` (the +/// `hive-dashboard-links` NixOS oneshot's output). Mapped to +/// `AgentLink { kind: External }` inside `agent_links`. +#[derive(serde::Deserialize)] +struct ExtraLink { + label: String, + #[serde(default)] + icon: String, + url: String, +} + +/// Best-effort: pull the last 30 messages addressed to us via the +/// per-agent / manager socket. Empty list on any transport / decode +/// failure — the inbox section is decorative, not authoritative. +async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> { + const LIMIT: u64 = 30; + // Deadline-bounded: `/api/state` must render even when hive-c0re is + // busy — an empty inbox section beats a hung snapshot. + match tokio::time::timeout( + SOCKET_FETCH_TIMEOUT, + client::request::<_, hive_sh4re::Response>( + socket, + &hive_sh4re::Request::Recent { limit: LIMIT }, + ), + ) + .await + { + Ok(Ok(hive_sh4re::Response::Recent { rows })) => rows, + _ => Vec::new(), + } +} + +/// Fetch reminder activity stats from the broker via the per-agent / +/// manager socket. Returns None on any transport / decode failure — the +/// stats are decorative, not authoritative. +pub(super) async fn fetch_reminder_stats( + socket: &std::path::Path, + window_secs: u64, +) -> Option<hive_sh4re::ReminderStats> { + match tokio::time::timeout( + SOCKET_FETCH_TIMEOUT, + client::request::<_, hive_sh4re::Response>( + socket, + &hive_sh4re::Request::ReminderRollup { + since_secs: window_secs, + agent: None, + }, + ), + ) + .await + { + Ok(Ok(hive_sh4re::Response::ReminderRollup(stats))) => Some(stats), + _ => None, + } +} + +/// Read `HIVE_AVAILABLE_MODELS` (comma-separated short names injected by +/// `services.hyperhive.availableModels`) and return the parsed list. +/// Falls back to `["haiku", "sonnet", "opus"]` when the env var is absent +/// or resolves to an empty list after trimming. +fn available_models() -> Vec<String> { + const DEFAULT: &[&str] = &["haiku", "sonnet", "opus"]; + let raw = match std::env::var("HIVE_AVAILABLE_MODELS") { + Ok(v) if !v.trim().is_empty() => v, + _ => return DEFAULT.iter().map(ToString::to_string).collect(), + }; + let models: Vec<String> = raw + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + if models.is_empty() { + DEFAULT.iter().map(ToString::to_string).collect() + } else { + models + } +} diff --git a/hive-ag3nt/src/web_ui/stats.rs b/hive-ag3nt/src/web_ui/stats.rs new file mode 100644 index 00000000..9c0ace5e --- /dev/null +++ b/hive-ag3nt/src/web_ui/stats.rs @@ -0,0 +1,131 @@ +//! Stats + loose-ends + bash-tasks read endpoints. + +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde::Deserialize; + +use crate::client; + +use super::state::fetch_reminder_stats; +use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response}; + +#[derive(Deserialize)] +pub(super) struct StatsQuery { + window: Option<String>, +} + +pub(super) async fn api_stats( + State(state): State<AppState>, + axum::extract::Query(q): axum::extract::Query<StatsQuery>, +) -> axum::Json<crate::stats::Snapshot> { + let window = crate::stats::Window::parse(q.window.as_deref().unwrap_or("24h")); + let mut snapshot = crate::stats::snapshot_default(window); + // Pass the window span to the reminder-stats RPC so the broker + // filters its counts to the same time range as the chart data. + let window_secs = window.span_secs(); + let window_secs_u = u64::try_from(window_secs).unwrap_or(0); + snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await; + axum::Json(snapshot) +} + +/// Proxy this agent's loose-ends list via the per-agent socket. The +/// web UI surfaces the result as a collapsible section in the page +/// so the operator can see at a glance what's pending against the +/// agent (questions asked by it, peer questions targeting it, +/// reminders it scheduled, approvals for the manager). Same data +/// the `mcp__hyperhive__get_loose_ends` tool sees from inside the +/// container. +pub(super) async fn api_loose_ends(State(state): State<AppState>) -> Response { + let loose_ends: Vec<hive_sh4re::LooseEnd> = match tokio::time::timeout( + SOCKET_FETCH_TIMEOUT, + client::request::<_, hive_sh4re::Response>( + &state.socket, + &hive_sh4re::Request::GetLooseEnds { agent: None }, + ), + ) + .await + { + Ok(Ok(hive_sh4re::Response::LooseEnds { loose_ends })) => loose_ends, + Ok(Ok(hive_sh4re::Response::Err { message })) => { + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("get_loose_ends: {message}"), + ); + } + Ok(Ok(other)) => { + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("unexpected response: {other:?}"), + ); + } + Ok(Err(e)) => { + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("transport: {e:#}"), + ); + } + Err(_) => { + return error_response( + StatusCode::CONFLICT, + "get_loose_ends: timed out — hive-c0re busy, retry", + ); + } + }; + axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response() +} + +/// `GET /api/bash-tasks` — snapshot of this agent's in-flight bash tasks. +/// +/// The `hive-bash-mcp` daemon runs in this same container and writes one +/// `<id>.json` ([`hive_sh4re::TaskFile`]) per task under the harness +/// `bash-tasks/` dir. This reads that dir and returns the tasks still +/// `Pending` or `Running`, so the agent page can show what's running without +/// going through the broker. Snapshot only — the page polls/refreshes it like +/// `/api/loose-ends`; there's no live SSE push for task state yet. Unreadable +/// or malformed files (incl. the daemon's `.json.tmp` scratch writes, which +/// don't match the `.json` extension) are skipped so one stray file can't +/// fail the whole list. +pub(super) async fn api_bash_tasks() -> Response { + let dir = crate::paths::harness_dir().join("bash-tasks"); + // The dir scan + per-file reads are blocking fs I/O; run them off the + // async executor so a slow or large tasks dir can't stall other requests. + let tasks = tokio::task::spawn_blocking(move || { + let mut tasks: Vec<hive_sh4re::TaskFile> = Vec::new(); + let Ok(rd) = std::fs::read_dir(&dir) else { + return tasks; + }; + for entry in rd.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + let Ok(text) = std::fs::read_to_string(&path) else { + continue; + }; + let Ok(task) = serde_json::from_str::<hive_sh4re::TaskFile>(&text) else { + continue; + }; + if matches!( + task.status, + hive_sh4re::TaskStatus::Pending | hive_sh4re::TaskStatus::Running + ) { + tasks.push(task); + } + } + // Running before Pending, then oldest-first so a long-runner sits on top. + tasks.sort_by(|a, b| { + let rank = |s: &hive_sh4re::TaskStatus| match s { + hive_sh4re::TaskStatus::Running => 0, + _ => 1, + }; + rank(&a.status) + .cmp(&rank(&b.status)) + .then(a.created_at.cmp(&b.created_at)) + }); + tasks + }) + .await + .unwrap_or_default(); + axum::Json(serde_json::json!({ "tasks": tasks })).into_response() +} diff --git a/hive-ag3nt/src/web_ui/stream.rs b/hive-ag3nt/src/web_ui/stream.rs new file mode 100644 index 00000000..dabbcc45 --- /dev/null +++ b/hive-ag3nt/src/web_ui/stream.rs @@ -0,0 +1,70 @@ +//! Live SSE event stream + history endpoints. + +use std::convert::Infallible; + +use axum::Json; +use axum::extract::{Query, State}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use serde::Deserialize; +use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream}; + +use super::AppState; + +/// Query params for the paginated history endpoint. +#[derive(Debug, Deserialize)] +pub(super) struct HistoryParams { + /// Cursor: only return events with sqlite row id < `before`. + /// Omit for the initial (most-recent) page. + before: Option<i64>, + /// Page size (default 100, capped at `HISTORY_CAPACITY`). + limit: Option<usize>, +} + +pub(super) async fn events_history( + State(state): State<AppState>, + Query(params): Query<HistoryParams>, +) -> Json<serde_json::Value> { + use crate::events::HISTORY_CAPACITY; + let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY); + let before = params.before; + let is_initial = before.is_none(); + + // Capture seq *before* the read on initial loads so the SSE dedupe + // window is "drop buffered events you've already seen in history", + // never "lose an event that fired between the read and the seq." + // On paginated loads (`before` is set) seq is not needed. + let seq = if is_initial { + Some(state.bus.current_seq()) + } else { + None + }; + + let (events, min_id, has_more) = state.bus.history_page(before, limit); + let mut resp = serde_json::json!({ + "events": events, + "min_id": min_id, + "has_more": has_more, + }); + if let Some(s) = seq { + resp["seq"] = serde_json::json!(s); + } + Json(resp) +} + +pub(super) async fn events_stream( + State(state): State<AppState>, +) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { + tracing::info!("sse: client subscribed"); + let rx = state.bus.subscribe(); + // Drop a "hello" note into the bus so every new subscriber sees at + // least one event immediately and can clear the connecting placeholder. + state.bus.emit(crate::events::LiveEvent::Note { + text: "live stream attached".into(), + }); + let stream = BroadcastStream::new(rx).filter_map(|res| { + let ev = res.ok()?; + let json = serde_json::to_string(&ev).ok()?; + Some(Ok(Event::default().data(json))) + }); + Sse::new(stream).keep_alive(KeepAlive::default()) +}