421 lines
18 KiB
Rust
421 lines
18 KiB
Rust
//! 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 proxy;
|
|
mod screen;
|
|
mod state;
|
|
mod stats;
|
|
mod stream;
|
|
|
|
use std::net::SocketAddr;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::atomic::AtomicBool;
|
|
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>,
|
|
/// Set by `post_cancel_turn` on a successful SIGINT; read-and-cleared
|
|
/// by the serve loop's next `handle_turn` to prepend
|
|
/// `hive_sh4re::INTERRUPTED_HINT` to that turn's wake prompt. Shared
|
|
/// with the serve loop via the same `Arc` (see `serve_main`) — an
|
|
/// in-memory flag, not a marker file, since a `/cancel` from a prior
|
|
/// process lifetime isn't meaningful once the harness restarts.
|
|
interrupted: Arc<AtomicBool>,
|
|
}
|
|
|
|
/// 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,
|
|
interrupted: Arc<AtomicBool>,
|
|
) -> 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,
|
|
interrupted,
|
|
};
|
|
let app: Router<AppState> = 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/todos", get(stats::api_todos))
|
|
.route("/api/todos/mark-done", post(actions::post_mark_todos_done))
|
|
.route("/api/stats", get(stats::api_stats))
|
|
.route("/screen/ws", get(screen::screen_ws))
|
|
.route("/icon", get(screen::serve_icon));
|
|
// Mount any `hyperhive.extraWebProxies` under `/extra/<name>/` before the
|
|
// static fallback so declared proxies win over `ServeDir`.
|
|
let app = proxy::mount_extra_proxies(app)
|
|
// 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)
|
|
}
|
|
|
|
/// Maximum bind attempts before `bind_with_retry` gives up on `AddrInUse`.
|
|
const MAX_BIND_ATTEMPTS: u32 = 12;
|
|
|
|
/// Bind a TCP listener with `SO_REUSEADDR` set, retrying on `AddrInUse` with
|
|
/// exponential backoff capped at 2s, up to [`MAX_BIND_ATTEMPTS`] attempts. If
|
|
/// the port is still held after the final attempt, returns the `AddrInUse`
|
|
/// error rather than looping forever (a genuine collision needs the operator,
|
|
/// not an unbounded wait).
|
|
///
|
|
/// Retry rationale + dashboard-banner-on-real-collision:
|
|
/// 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 >= MAX_BIND_ATTEMPTS {
|
|
return Err(e).with_context(|| {
|
|
format!("bind {label} on {addr}: still AddrInUse after {attempt} attempts")
|
|
});
|
|
}
|
|
tracing::warn!(
|
|
%addr, attempt,
|
|
"{label}: AddrInUse, 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()
|
|
}
|
|
|
|
/// Outcome of [`sigint_claude`] — shared by `/api/cancel` and `/api/logout`
|
|
/// so both can render the same three cases.
|
|
enum SigintOutcome {
|
|
/// The tracked pid was signalled.
|
|
Signalled,
|
|
/// No claude child is currently tracked (turn not in flight, or it
|
|
/// finished/exited between the check and the signal — same "nothing to
|
|
/// interrupt" outcome either way).
|
|
NoProcess,
|
|
/// The `kill` syscall itself failed for a reason other than "no such
|
|
/// process" (permission issue, etc).
|
|
Failed(std::io::Error),
|
|
}
|
|
|
|
/// Find a `claude` process that is a direct OS child of this harness
|
|
/// process, if any. `hive-claude`'s driver spawns the turn's claude
|
|
/// directly (`Command::new(program).spawn()`, no shell in between), so the
|
|
/// harness is always the immediate parent of any claude turn it started —
|
|
/// scanning `/proc/*/status` for `PPid: <our own pid>` plus `/proc/*/cmdline`
|
|
/// for an argv[0] of `claude` finds *that* specific process without needing
|
|
/// the driver to surface its pid through any extra plumbing. Distinguishes
|
|
/// the harness's own tracked turn from an unrelated `claude` someone is
|
|
/// running interactively in the same container (a manually shelled-in
|
|
/// "choom" session) — that one's parent is a login shell, not us.
|
|
///
|
|
/// **Matches on `cmdline`, not `status`'s `Name:` field.** The nixpkgs
|
|
/// `claude-code` package wraps its real binary (`wrapProgram`-style: the
|
|
/// executable on `PATH` is a thin `exec -a claude .../.claude-wrapped ...`
|
|
/// shim) — `exec -a` only overrides argv[0] as the process itself/`cmdline`
|
|
/// see it, not the kernel's own `comm` (what `status`'s `Name:` line
|
|
/// reports, set from the executed binary's own basename at `execve` time).
|
|
/// So `Name:` shows `.claude-wrapped`, not `claude`, on a wrapped package —
|
|
/// `cmdline`'s first argument still carries the bare name
|
|
/// `Command::new("claude")` resolved on `PATH`, which is what actually
|
|
/// matters here. Best-effort: a process that exits mid-scan (its
|
|
/// `/proc/<pid>/{status,cmdline}` read fails, ESRCH) is just skipped.
|
|
fn find_claude_child() -> Option<u32> {
|
|
let own_pid = std::process::id();
|
|
for entry in std::fs::read_dir("/proc").ok()?.flatten() {
|
|
let Ok(pid) = entry.file_name().to_string_lossy().parse::<u32>() else {
|
|
continue; // not a pid dir (self, cwd, net, ...)
|
|
};
|
|
let Ok(status) = std::fs::read_to_string(entry.path().join("status")) else {
|
|
continue;
|
|
};
|
|
let parent_pid = status
|
|
.lines()
|
|
.find_map(|line| line.strip_prefix("PPid:"))
|
|
.and_then(|v| v.trim().parse::<u32>().ok());
|
|
if parent_pid != Some(own_pid) {
|
|
continue;
|
|
}
|
|
let Ok(cmdline) = std::fs::read(entry.path().join("cmdline")) else {
|
|
continue;
|
|
};
|
|
let argv0 = cmdline.split(|&b| b == 0).next().unwrap_or_default();
|
|
if argv0 == b"claude" {
|
|
return Some(pid);
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
/// SIGINT this container's harness-spawned claude child, if any
|
|
/// (best-effort). See [`find_claude_child`] for how it's identified —
|
|
/// **not** a name-based `pkill claude`, which would also hit an unrelated
|
|
/// `claude` process someone is running interactively in the container.
|
|
async fn sigint_claude() -> SigintOutcome {
|
|
let Some(pid) = find_claude_child() else {
|
|
return SigintOutcome::NoProcess;
|
|
};
|
|
match tokio::process::Command::new("kill")
|
|
.args(["-INT", &pid.to_string()])
|
|
.status()
|
|
.await
|
|
{
|
|
Ok(status) if status.success() => SigintOutcome::Signalled,
|
|
// Non-zero from `kill` means "no such process" (ESRCH) — it already
|
|
// exited between the check above and the signal. Same as never
|
|
// having found it.
|
|
Ok(_) => SigintOutcome::NoProcess,
|
|
Err(e) => SigintOutcome::Failed(e),
|
|
}
|
|
}
|
|
|
|
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()
|
|
}
|
|
|
|
/// Why a deadline-bounded broker request via the per-agent socket didn't
|
|
/// yield a response. Kept distinct so action handlers pick the right status
|
|
/// code (see [`broker_error_response`]) while decorative fetches `.ok()` both.
|
|
enum BrokerError {
|
|
/// Outran [`SOCKET_FETCH_TIMEOUT`] — hive-c0re is busy or stalled. A
|
|
/// retryable state conflict (→ 409), not a server fault.
|
|
Timeout,
|
|
/// The socket transport itself failed (connect / encode / decode).
|
|
Transport(anyhow::Error),
|
|
}
|
|
|
|
/// Issue a broker request over the per-agent socket, bounded by
|
|
/// [`SOCKET_FETCH_TIMEOUT`] so a busy or stalled hive-c0re degrades the
|
|
/// response instead of hanging it. Callers match the returned [`Response`]
|
|
/// variant themselves; the error side distinguishes a retryable timeout from
|
|
/// a transport failure. This is the one shared broker-call scaffold — every
|
|
/// web-UI handler that talks to the broker goes through it.
|
|
async fn broker_request(
|
|
socket: &Path,
|
|
req: &hive_core_agent_sock::Request,
|
|
) -> std::result::Result<hive_core_agent_sock::Response, BrokerError> {
|
|
match tokio::time::timeout(
|
|
SOCKET_FETCH_TIMEOUT,
|
|
hive_sock_client::request::<_, hive_core_agent_sock::Response>(
|
|
socket,
|
|
req,
|
|
crate::CONTROL_SOCKET_RETRY,
|
|
),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(resp)) => Ok(resp),
|
|
Ok(Err(e)) => Err(BrokerError::Transport(e)),
|
|
Err(_) => Err(BrokerError::Timeout),
|
|
}
|
|
}
|
|
|
|
/// Map a [`BrokerError`] to an operator-facing error response: a timeout is a
|
|
/// retryable "busy" conflict (409), a transport failure is a 500. `action`
|
|
/// prefixes the message (e.g. `"send"`, `"get_loose_ends"`).
|
|
fn broker_error_response(err: &BrokerError, action: &str) -> Response {
|
|
match err {
|
|
BrokerError::Timeout => error_response(
|
|
StatusCode::CONFLICT,
|
|
&format!("{action}: timed out — hive-c0re busy, retry"),
|
|
),
|
|
BrokerError::Transport(e) => error_response(
|
|
StatusCode::INTERNAL_SERVER_ERROR,
|
|
&format!("{action}: transport: {e:#}"),
|
|
),
|
|
}
|
|
}
|