vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#1456): extract dashboard build-log endpoints into dashboard/build_logs.rs

Browse source
This commit is contained in:
damocles 2026-06-08 23:16:05 +02:00 • committed by mara
commit 4e06a9682d
2 changed files with 248 additions and 211 deletions
Download patch file Download diff file Expand all files Collapse all files

230
hive-c0re/src/dashboard.rs
View file

@ -21,7 +21,7 @@ use axum::{
};
use hive_sh4re::Approval;
use serde::{Deserialize, Serialize};
use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::{Stream, StreamExt};
use tower_http::services::ServeDir;
@ -30,6 +30,7 @@ use crate::container_view::{ContainerView, claude_has_session};
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, MANAGER_NAME};
mod build_logs;
mod journal;
mod permissions;
mod questions;
@ -91,11 +92,23 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/api/operator-inbox", get(api_operator_inbox))
.route("/api/stats-hive", get(api_stats_hive))
.route("/api/container-resources", get(api_container_resources))
.route("/api/build-logs", get(get_build_logs_all))
.route("/api/build-logs/{agent}", get(get_build_logs_agent))
.route("/api/build-logs/id/{id}", get(get_build_log_full))
.route("/api/build-logs/id/{id}/stream", get(get_build_log_stream))
.route("/api/build-logs/id/{id}/raw", get(get_build_log_raw))
.route("/api/build-logs", get(build_logs::get_build_logs_all))
.route(
"/api/build-logs/{agent}",
get(build_logs::get_build_logs_agent),
)
.route(
"/api/build-logs/id/{id}",
get(build_logs::get_build_log_full),
)
.route(
"/api/build-logs/id/{id}/stream",
get(build_logs::get_build_log_stream),
)
.route(
"/api/build-logs/id/{id}/raw",
get(build_logs::get_build_log_raw),
)
.route("/api/agent/{name}/mark-all-read", post(post_mark_all_read))
.route(
"/cancel-reminder/{id}",
@ -1094,26 +1107,6 @@ struct RequestSpawnForm {
name: String,
}
#[derive(Deserialize)]
struct BuildLogsAllQuery {
/// Max rows to return. Capped at 100. Default 30.
#[serde(default)]
limit: Option<usize>,
}
/// `GET /api/build-logs?limit=N` — most-recent build log headers across
/// all agents, newest first. Same JSON shape as the per-agent endpoint.
async fn get_build_logs_all(
State(state): State<AppState>,
axum::extract::Query(q): axum::extract::Query<BuildLogsAllQuery>,
) -> Response {
let limit = q.limit.unwrap_or(30);
match state.coord.build_logs.list_recent_all(limit) {
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs all: {e:#}")),
}
}
#[derive(Deserialize)]
struct StateFileQuery {
path: String,
@ -1608,191 +1601,6 @@ async fn api_container_resources() -> Response {
axum::Json(crate::container_stats::gather().await).into_response()
}
#[derive(Deserialize)]
struct BuildLogsQuery {
/// Maximum number of rows to return. Capped server-side at 50
/// (see `build_logs::list_recent_for_agent`). Default 10.
#[serde(default)]
limit: Option<usize>,
}
/// `GET /api/build-logs/{agent}?limit=N` — most-recent build log
/// headers for one agent, newest first. Returns
/// `Vec<BuildLogHeader>` (JSON). Limit defaults to 10, server-side
/// cap at 50. Backs the per-agent log chip in the agent card and
/// the side-panel header list.
async fn get_build_logs_agent(
State(state): State<AppState>,
AxumPath(name): AxumPath<String>,
axum::extract::Query(q): axum::extract::Query<BuildLogsQuery>,
) -> Response {
if let Some(reason) = validate_agent_name(&name) {
return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response();
}
let limit = q.limit.unwrap_or(10);
match state.coord.build_logs.list_recent_for_agent(&name, limit) {
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs {name}: {e:#}")),
}
}
/// `GET /api/build-logs/id/{id}` — full build log row (stdout +
/// stderr concatenated) by id. Returns `BuildLogFull` (JSON), or
/// HTTP 404 when the id doesn't exist (vacuum-reaped, or the
/// operator passed a stale id from a refresh race).
async fn get_build_log_full(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => axum::Json(log).into_response(),
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}
/// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
/// `stdout_append` / `stderr_append` carry only the new bytes since the
/// last frame; `done = true` means the build finished and the stream
/// will close after this frame.
#[derive(Serialize)]
struct BuildLogFrame {
stdout_append: String,
stderr_append: String,
#[serde(skip_serializing_if = "Option::is_none")]
status: Option<String>,
done: bool,
}
/// `GET /api/build-logs/id/{id}/stream` — SSE stream that delivers
/// incremental stdout/stderr as a build runs. The client connects when
/// it opens a running-build panel; the stream closes automatically once
/// the build finishes (or the row disappears due to a vacuum).
///
/// Each frame is a JSON-serialised `BuildLogFrame`. The first frame
/// always carries the full accumulated log so far (cursors start at 0);
/// subsequent frames carry only new bytes. `done: true` on the final
/// frame signals the browser to close the `EventSource`.
async fn get_build_log_stream(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(32);
let logs = state.coord.build_logs.clone();
tokio::spawn(async move {
let mut notify_rx = logs.subscribe_notifications();
let mut stdout_cursor = 0usize;
let mut stderr_cursor = 0usize;
// ── initial snapshot ──────────────────────────────────────────
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) {
let _ = tx.send(Ok(Event::default().data(json))).await;
}
if done {
return;
}
}
Ok(None) => {
// Row missing — send a single error event and exit.
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id} not found"))))
.await;
return;
}
Err(e) => {
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id}: {e:#}"))))
.await;
return;
}
}
// ── live delta loop ───────────────────────────────────────────
loop {
match notify_rx.recv().await {
// Notification for a different build — ignore and wait
// for the next one.
Ok(notif_id) if notif_id != id => {}
Ok(_) => {
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) && tx.send(Ok(Event::default().data(json))).await.is_err()
{
return; // browser disconnected
}
if done {
return;
}
}
Ok(None) | Err(_) => return, // vacuum reaped row / channel closed
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
}
}
});
Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default())
}
/// `GET /api/build-logs/id/{id}/raw` — full log as `text/plain` for
/// download. Stdout and stderr are concatenated with a `--- stderr ---`
/// separator (same layout the JS side-panel renders). The
/// `Content-Disposition` header triggers a browser download with a
/// descriptive filename so the operator can save and share the log.
async fn get_build_log_raw(State(state): State<AppState>, AxumPath(id): AxumPath<i64>) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => {
let mut text = log.stdout;
if !log.stderr.is_empty() {
text.push_str("\n--- stderr ---\n");
text.push_str(&log.stderr);
}
(
StatusCode::OK,
[
("content-type", "text/plain; charset=utf-8".to_string()),
(
"content-disposition",
format!(
"attachment; filename=\"build-log-{}-{}.txt\"",
log.header.agent, id
),
),
],
text,
)
.into_response()
}
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}
/// Validate that a path-param agent name conforms to the hyperhive
/// naming whitelist: 1-63 chars of `[a-z0-9_-]`. Rejects empty,
/// uppercase, slashes, dots, and any non-ASCII (incl. unicode

229
hive-c0re/src/dashboard/build_logs.rs Normal file
View file

@ -0,0 +1,229 @@
//! Build-log endpoints for the dashboard.
//!
//! Header lists (all-agents + per-agent), the full row by id, a `text/plain`
//! download, and an SSE stream that delivers incremental stdout/stderr while
//! a build runs (closing once it finishes or the row is vacuum-reaped).
use std::convert::Infallible;
use axum::{
extract::{Path as AxumPath, State},
http::StatusCode,
response::{
IntoResponse, Response,
sse::{Event, KeepAlive, Sse},
},
};
use serde::{Deserialize, Serialize};
use tokio_stream::Stream;
use tokio_stream::wrappers::ReceiverStream;
use super::{AppState, error_response, validate_agent_name};
#[derive(Deserialize)]
pub(super) struct BuildLogsAllQuery {
/// Max rows to return. Capped at 100. Default 30.
#[serde(default)]
limit: Option<usize>,
}
/// `GET /api/build-logs?limit=N` — most-recent build log headers across
/// all agents, newest first. Same JSON shape as the per-agent endpoint.
pub(super) async fn get_build_logs_all(
State(state): State<AppState>,
axum::extract::Query(q): axum::extract::Query<BuildLogsAllQuery>,
) -> Response {
let limit = q.limit.unwrap_or(30);
match state.coord.build_logs.list_recent_all(limit) {
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs all: {e:#}")),
}
}
#[derive(Deserialize)]
pub(super) struct BuildLogsQuery {
/// Maximum number of rows to return. Capped server-side at 50
/// (see `build_logs::list_recent_for_agent`). Default 10.
#[serde(default)]
limit: Option<usize>,
}
/// `GET /api/build-logs/{agent}?limit=N` — most-recent build log
/// headers for one agent, newest first. Returns
/// `Vec<BuildLogHeader>` (JSON). Limit defaults to 10, server-side
/// cap at 50. Backs the per-agent log chip in the agent card and
/// the side-panel header list.
pub(super) async fn get_build_logs_agent(
State(state): State<AppState>,
AxumPath(name): AxumPath<String>,
axum::extract::Query(q): axum::extract::Query<BuildLogsQuery>,
) -> Response {
if let Some(reason) = validate_agent_name(&name) {
return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response();
}
let limit = q.limit.unwrap_or(10);
match state.coord.build_logs.list_recent_for_agent(&name, limit) {
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs {name}: {e:#}")),
}
}
/// `GET /api/build-logs/id/{id}` — full build log row (stdout +
/// stderr concatenated) by id. Returns `BuildLogFull` (JSON), or
/// HTTP 404 when the id doesn't exist (vacuum-reaped, or the
/// operator passed a stale id from a refresh race).
pub(super) async fn get_build_log_full(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => axum::Json(log).into_response(),
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}
/// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
/// `stdout_append` / `stderr_append` carry only the new bytes since the
/// last frame; `done = true` means the build finished and the stream
/// will close after this frame.
#[derive(Serialize)]
struct BuildLogFrame {
stdout_append: String,
stderr_append: String,
#[serde(skip_serializing_if = "Option::is_none")]
status: Option<String>,
done: bool,
}
/// `GET /api/build-logs/id/{id}/stream` — SSE stream that delivers
/// incremental stdout/stderr as a build runs. The client connects when
/// it opens a running-build panel; the stream closes automatically once
/// the build finishes (or the row disappears due to a vacuum).
///
/// Each frame is a JSON-serialised `BuildLogFrame`. The first frame
/// always carries the full accumulated log so far (cursors start at 0);
/// subsequent frames carry only new bytes. `done: true` on the final
/// frame signals the browser to close the `EventSource`.
pub(super) async fn get_build_log_stream(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(32);
let logs = state.coord.build_logs.clone();
tokio::spawn(async move {
let mut notify_rx = logs.subscribe_notifications();
let mut stdout_cursor = 0usize;
let mut stderr_cursor = 0usize;
// ── initial snapshot ──────────────────────────────────────────
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) {
let _ = tx.send(Ok(Event::default().data(json))).await;
}
if done {
return;
}
}
Ok(None) => {
// Row missing — send a single error event and exit.
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id} not found"))))
.await;
return;
}
Err(e) => {
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id}: {e:#}"))))
.await;
return;
}
}
// ── live delta loop ───────────────────────────────────────────
loop {
match notify_rx.recv().await {
// Notification for a different build — ignore and wait
// for the next one.
Ok(notif_id) if notif_id != id => {}
Ok(_) => {
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) && tx.send(Ok(Event::default().data(json))).await.is_err()
{
return; // browser disconnected
}
if done {
return;
}
}
Ok(None) | Err(_) => return, // vacuum reaped row / channel closed
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
}
}
});
Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default())
}
/// `GET /api/build-logs/id/{id}/raw` — full log as `text/plain` for
/// download. Stdout and stderr are concatenated with a `--- stderr ---`
/// separator (same layout the JS side-panel renders). The
/// `Content-Disposition` header triggers a browser download with a
/// descriptive filename so the operator can save and share the log.
pub(super) async fn get_build_log_raw(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => {
let mut text = log.stdout;
if !log.stderr.is_empty() {
text.push_str("\n--- stderr ---\n");
text.push_str(&log.stderr);
}
(
StatusCode::OK,
[
("content-type", "text/plain; charset=utf-8".to_string()),
(
"content-disposition",
format!(
"attachment; filename=\"build-log-{}-{}.txt\"",
log.header.agent, id
),
),
],
text,
)
.into_response()
}
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}