//! 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};
use crate::mcp;
use crate::turn::TurnFiles;

/// 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>>;

/// Shared turn lock. The serve loop acquires this (as an async mutex) for the
/// duration of every `drive_turn` call. The `/api/compact` handler tries
/// `try_lock()` and rejects immediately if a turn is in flight, preventing
/// concurrent access to the claude session.
pub type TurnLock = Arc<tokio::sync::Mutex<()>>;

#[derive(Clone)]
struct AppState {
    label: String,
    login: LoginStateCell,
    session: Arc<Mutex<Option<Arc<LoginSession>>>>,
    bus: Bus,
    socket: PathBuf,
    /// Same `TurnFiles` the harness's turn loop uses. Shared so
    /// `/api/compact` re-uses the exact MCP config / system prompt /
    /// settings claude saw on the last regular turn — keeps the
    /// session shape identical across compact + normal turns.
    files: TurnFiles,
    /// Prevents `/api/compact` from racing with an in-flight normal turn.
    turn_lock: TurnLock,
    /// VNC port read from `/etc/hyperhive/gui.json` at startup.
    /// `None` when the file is absent (gui not enabled for this agent).
    gui_vnc_port: Option<u16>,
}

/// Re-export so callers in `turn.rs` can name the type via `web_ui::Flavor`.
pub type Flavor = mcp::Flavor;

/// 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,
    files: TurnFiles,
    turn_lock: TurnLock,
) -> Result<()> {
    let gui_vnc_port = read_gui_json();
    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,
        files,
        turn_lock,
        gui_vnc_port,
    };
    let app = Router::new()
        .route("/api/state", get(api_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/new-session", post(post_new_session))
        .route("/api/logout", post(post_logout))
        .route("/api/loose-ends", get(api_loose_ends))
        .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/<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)
}

// ---------------------------------------------------------------------------
// 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<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)
}

/// 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)
}

/// Read `/etc/hyperhive/gui.json` and extract the `vnc_port` field.
/// Returns `None` if the file is absent or unparseable — GUI not enabled.
fn read_gui_json() -> Option<u16> {
    let text = std::fs::read_to_string("/etc/hyperhive/gui.json").ok()?;
    let val: serde_json::Value = serde_json::from_str(&text).ok()?;
    val["vnc_port"].as_u64().and_then(|p| u16::try_from(p).ok())
}

/// 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.
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 => {}
    }
}

#[derive(Deserialize)]
struct StatsQuery {
    window: Option<String>,
}

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)
}

#[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<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<crate::events::TokenUsage>,
    /// Cumulative token usage across the most recent turn's inferences
    /// (cost signal). `null` until the first turn finishes.
    cost_usage: Option<crate::events::TokenUsage>,
    /// Navigation links for this agent page. The same list feeds the
    /// dashboard card's icon-only nav strip via hive-c0re's
    /// `GET /api/agent/{name}/links` same-origin passthrough proxy
    /// — single source of truth, no CORS. Per-agent page also reads
    /// this directly. 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>,
}

/// One navigation link in the agent page header row. Same JSON
/// shape feeds the dashboard's icon-only nav strip via the host's
/// `GET /api/agent/{name}/links` passthrough proxy, so the agent
/// backend 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 client::request::<_, hive_sh4re::Response>(
        &state.socket,
        &hive_sh4re::Request::GetLooseEnds { agent: None },
    )
    .await
    {
        Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => loose_ends,
        Ok(hive_sh4re::Response::Err { message }) => {
            return error_response(&format!("get_loose_ends: {message}"));
        }
        Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
        Err(e) => return error_response(&format!("transport: {e:#}")),
    };
    axum::Json(serde_json::json!({ "loose_ends": loose_ends })).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();
    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(),
    })
}

/// 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;
    match client::request::<_, hive_sh4re::Response>(
        socket,
        &hive_sh4re::Request::Recent { limit: LIMIT },
    )
    .await
    {
        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 client::request::<_, hive_sh4re::Response>(
        socket,
        &hive_sh4re::Request::ReminderRollup {
            since_secs: window_secs,
            agent: None,
        },
    )
    .await
    {
        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("send: `body` required");
    }
    let result = match client::request::<_, hive_sh4re::Response>(
        &state.socket,
        &hive_sh4re::Request::OperatorMsg { body },
    )
    .await
    {
        Ok(hive_sh4re::Response::Ok) => Ok(()),
        Ok(hive_sh4re::Response::Err { message }) => Err(message),
        Ok(other) => Err(format!("unexpected response: {other:?}")),
        Err(e) => Err(format!("transport: {e:#}")),
    };
    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(&format!("send failed: {e}")),
    }
}

async fn events_history(State(state): State<AppState>) -> axum::Json<serde_json::Value> {
    // Capture seq *before* the read so dedupe is "drop buffered events
    // you've already seen in history", never "lose an event that fired
    // between the read and the timestamp." Historical rows have no
    // per-row seq; only the high-water mark matters for the dedupe
    // window.
    let seq = state.bus.current_seq();
    let events = state.bus.history();
    axum::Json(serde_json::json!({ "seq": seq, "events": events }))
}

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(&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("no login session running");
    };
    if let Err(e) = session.submit_code(&form.code).await {
        return error_response(&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()
}

/// Operator-initiated session compaction. Spawns `turn::compact_session`
/// in the background — the HTTP handler returns immediately so the
/// async-form spinner can clear. Output (claude's compaction stream,
/// the "/compact done" note) lands in the live event panel like any
/// other turn. If a regular turn is in flight, claude's own session
/// lock will reject this one and we surface the error as a Note.
#[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("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()
}

async fn post_compact(State(state): State<AppState>) -> Response {
    // Clone the Arc before locking so the guard's lifetime is tied to the
    // clone (which we can move into the spawn) rather than to `state`.
    let lock = state.turn_lock.clone();
    // Reject immediately if a normal turn is in flight — concurrent access
    // to the claude session is unsafe and produces garbled output.
    let Ok(guard) = lock.try_lock_owned() else {
        return error_response("turn in flight — wait for it to finish before compacting");
    };
    let bus = state.bus.clone();
    let files = state.files.clone();
    tokio::spawn(async move {
        let _guard = guard; // keep lock alive for the duration of compaction
        bus.emit(crate::events::LiveEvent::Note {
            text: "operator: /compact — running on persistent session".into(),
        });
        bus.set_state(crate::events::TurnState::Compacting);
        let outcome = crate::turn::compact_session(&files, &bus).await;
        bus.set_state(crate::events::TurnState::Idle);
        // Best-effort manual /compact from the operator: compact_session
        // already emits a Note per outcome, so we don't need to re-emit
        // here — just record any underlying error to the harness log.
        if let crate::turn::TurnOutcome::Failed(e) = outcome {
            tracing::warn!(error = %format!("{e:#}"), "operator /compact failed");
        }
    });
    (axum::http::StatusCode::OK, "ok").into_response()
}

/// Cancel the in-flight claude turn. Coarse-grained: shells out
/// `pkill -INT claude` since there's at most one claude per container.
/// SIGINT (not SIGTERM) so claude flushes anything in-flight and emits a
/// final result row. Emits a Note so the operator sees the cancel
/// landed; the actual state transition back to `idle` happens when
/// `run_claude` wakes up and the harness emits `TurnEnd`.
/// Arm a one-shot: the next claude turn drops `--continue`, starting a
/// fresh session. Subsequent turns resume normal `--continue`
/// behavior. Idempotent before the next turn fires — calling twice
/// still results in a single fresh start. 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_new_session();
    state.bus.emit(crate::events::LiveEvent::Note {
        text: "operator: new session armed — next turn runs without --continue".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(message: &str) -> Response {
    // Plain text — JS app surfaces in `alert()`, HTML wrapping would just
    // be noise.
    (StatusCode::INTERNAL_SERVER_ERROR, message.to_owned()).into_response()
}