build-logs: SSE live streaming + raw download (#726 phase 4)

Backend (hive-c0re):
- build_logs.rs: add tokio::sync::broadcast::Sender<i64> to BuildLogs;
  append() and finish() notify subscribers after each sqlite write.
  Add BuildLogProgress struct and get_progress(id, stdout_cursor,
  stderr_cursor) for incremental delta reads.
- dashboard.rs: two new endpoints —
    GET /api/build-logs/id/{id}/stream  SSE; streams BuildLogFrame
      {stdout_append, stderr_append, status?, done} deltas until the
      build finishes or the browser disconnects. Backed by an mpsc
      channel task that watches the per-build broadcast notifications.
    GET /api/build-logs/id/{id}/raw    text/plain download with
      Content-Disposition: attachment; filename build-log-{agent}-{id}.txt

Frontend (dashboard):
- tabs.js: running builds (status === null) connect an EventSource to
  /stream and append lines live; "live" badge pulses amber while active,
  flips to ok/fail on done. Finished builds still use the JSON fetch path.
  Collapsing a running panel closes the EventSource; re-expanding
  reconnects. Adds a "⬇ raw" download link to every expanded row.
- dashboard.css: .build-logs-dl inline download link; .build-logs-live
  live pulse @keyframes animation.

Docs: web-ui.md updated for all three new endpoints + behaviour.
This commit is contained in:
iris 2026-05-31 21:23:01 +02:00 committed by mara
commit 050e130eba
5 changed files with 365 additions and 14 deletions

View file

@ -26,6 +26,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use tokio::sync::broadcast;
/// Process-singleton handle, set once at coordinator startup. Lets
/// the `lifecycle` module's `run` / `prebuild_toplevel` access the
@ -123,11 +124,38 @@ pub struct BuildLogFull {
pub stderr: String,
}
/// Incremental text returned by `get_progress`. Carries only the new
/// bytes since the caller's last cursor positions so the SSE stream
/// handler can send deltas without re-transmitting the full log.
#[derive(Debug, Clone)]
pub struct BuildLogProgress {
/// New stdout bytes beyond `stdout_cursor`.
pub stdout_append: String,
/// New stderr bytes beyond `stderr_cursor`.
pub stderr_append: String,
/// `Some(unix_ts)` once the build is finished.
pub finished_at: Option<i64>,
/// Terminal status string (`"ok"` / `"fail"`) once finished.
pub status: Option<String>,
}
/// Channel capacity for per-build append notifications. 64 slots is
/// plenty — the consumer reads fast relative to line-append rate and
/// any lag means "read now, you have new content" rather than a lost
/// data line.
const NOTIFY_CAP: usize = 64;
/// Sqlite-backed build-log store. `Arc<BuildLogs>`-friendly: all
/// methods take `&self`, internal `Mutex<Connection>` serializes
/// access.
pub struct BuildLogs {
conn: Mutex<Connection>,
/// Broadcast channel that fires with the `id` of the row that just
/// had a line appended or was finished. The SSE stream handler
/// subscribes once per open panel and drives delta reads from this.
/// `send()` is non-async and silently drops frames when there are
/// no subscribers — safe to call from sync append/finish paths.
notify_tx: broadcast::Sender<i64>,
}
impl BuildLogs {
@ -139,11 +167,20 @@ impl BuildLogs {
.with_context(|| format!("open build_logs db {}", path.display()))?;
conn.execute_batch(SCHEMA)
.context("apply build_logs schema")?;
let (notify_tx, _) = broadcast::channel(NOTIFY_CAP);
Ok(Self {
conn: Mutex::new(conn),
notify_tx,
})
}
/// Subscribe to per-build append/finish notifications. Each emitted
/// value is the `id` of the row that changed. The SSE stream handler
/// calls this once and filters for its target id.
pub fn subscribe_notifications(&self) -> broadcast::Receiver<i64> {
self.notify_tx.subscribe()
}
/// Open a row for a new build attempt. Returns the assigned id
/// — the caller threads it through `append_stdout` / `append_stderr`
/// while the child runs and into `finish` once it exits.
@ -185,6 +222,14 @@ impl BuildLogs {
"build_logs: append failed (dropping line)"
);
}
drop(conn);
// Notify SSE stream subscribers — non-blocking, no-op when no
// subscribers are watching (e.g. no panel is open). Lagged
// receivers (channel full) automatically drop frames; the SSE
// handler re-reads the full delta on the next notification it
// does receive, so no content is lost, only an intermediate
// wake-up is coalesced.
let _ = self.notify_tx.send(id);
}
/// Finalize a build attempt. Sets `finished_at` to now and
@ -202,6 +247,62 @@ impl BuildLogs {
"build_logs: finish failed"
);
}
drop(conn);
// Final notification so the SSE stream handler sees the
// finished_at and status, closes the connection cleanly.
let _ = self.notify_tx.send(id);
}
/// Return incremental log content beyond the given byte cursors.
/// Used by the SSE stream handler to compute deltas between polls.
///
/// `stdout_cursor` / `stderr_cursor` are byte offsets into the
/// stored `stdout` / `stderr` columns from the previous read.
/// Slicing is safe because cursors are always derived from prior
/// `String::len()` values (valid UTF-8 boundaries).
///
/// Returns `None` when the row no longer exists (vacuum reap during
/// a long-open panel).
pub fn get_progress(
&self,
id: i64,
stdout_cursor: usize,
stderr_cursor: usize,
) -> Result<Option<BuildLogProgress>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT stdout, stderr, finished_at, status \
FROM build_logs WHERE id = ?1",
)?;
let row = stmt
.query_row(params![id], |r| {
Ok((
r.get::<_, String>(0)?,
r.get::<_, String>(1)?,
r.get::<_, Option<i64>>(2)?,
r.get::<_, Option<String>>(3)?,
))
})
.optional()?;
match row {
None => Ok(None),
Some((stdout, stderr, finished_at, status)) => {
let stdout_append = stdout
.get(stdout_cursor..)
.unwrap_or("")
.to_string();
let stderr_append = stderr
.get(stderr_cursor..)
.unwrap_or("")
.to_string();
Ok(Some(BuildLogProgress {
stdout_append,
stderr_append,
finished_at,
status,
}))
}
}
}
/// Return the most recent `limit` rows for `agent`, newest first.

View file

@ -21,7 +21,7 @@ use axum::{
};
use hive_sh4re::Approval;
use serde::{Deserialize, Serialize};
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};
use tokio_stream::{Stream, StreamExt};
use tower_http::services::ServeDir;
@ -68,6 +68,8 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/api/reminders", get(api_reminders))
.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/agent/{name}/links", get(get_agent_links))
.route("/api/agent/{name}/mark-all-read", post(post_mark_all_read))
.route("/cancel-reminder/{id}", post(post_cancel_reminder))
@ -1686,6 +1688,154 @@ async fn get_build_log_full(
}
}
/// 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 => continue,
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,
}) {
if tx.send(Ok(Event::default().data(json))).await.is_err() {
return; // browser disconnected
}
}
if done {
return;
}
}
Ok(None) => return, // vacuum reaped the row
Err(_) => return,
}
}
Err(_) => return, // notification channel closed (shutdown)
}
}
});
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:#}")),
}
}
/// `GET /api/schedules` — snapshot of every schedule for the
/// scheduled-prompts tab. Returns the wire shape directly
/// so the frontend can render without an extra translation layer.