From ae3f011eb3ba37bbbbf46b187a882c7fe4d99daa Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 19 Jul 2026 17:32:35 +0200 Subject: [PATCH] =?UTF-8?q?hive-agent:=20add=20stream=5Fenrich=20module=20?= =?UTF-8?q?=E2=80=94=20stamp=20=5Ficon/=5Fsummary/=5Fcategory=20on=20SSE?= =?UTF-8?q?=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move per-message rendering logic from frontend JS to backend Rust. A new stream_enrich::enrich() function stamps display fields onto LiveEvent::Stream payloads at SSE-emit time (both live tail and history replay), so the frontend can consume pre-computed fields instead of re-implementing the dispatch logic in JavaScript. Phase 1: backend stamps _icon/_summary/_category; client falls back to its own JS tables when absent. Zero user-visible change. - system events: _category (drop/thinking_tok/note/details) + _summary and optional _body (commands_changed expands to a slash-cmd list) - assistant tool_use entries: _icon + _summary per tool; rich tools (Write, Edit, send, ask, answer) get _category: 'rich' - enrichment applied in web_ui/stream.rs at emit time; DB stores raw events so no migration is needed when enrichment logic changes - idempotent: existing _-prefixed fields are left unchanged --- hive-agent/src/main.rs | 1 + hive-agent/src/stream_enrich.rs | 712 ++++++++++++++++++++++++++++++++ hive-agent/src/web_ui/stream.rs | 21 +- 3 files changed, 733 insertions(+), 1 deletion(-) create mode 100644 hive-agent/src/stream_enrich.rs diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 7e0d98c4..23ee0e7d 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -22,6 +22,7 @@ mod plugins; mod prompt; mod serve_common; mod stats; +mod stream_enrich; mod turn; mod turn_stats; mod vacuum; diff --git a/hive-agent/src/stream_enrich.rs b/hive-agent/src/stream_enrich.rs new file mode 100644 index 00000000..241e16dd --- /dev/null +++ b/hive-agent/src/stream_enrich.rs @@ -0,0 +1,712 @@ +//! Enrich raw claude stream-json values before SSE delivery. +//! +//! A single [`enrich`] function stamps `_icon`, `_summary`, `_category`, +//! and optionally `_body` onto [`crate::events::LiveEvent::Stream`] payloads +//! so the frontend can read pre-computed fields instead of duplicating the +//! dispatch logic in JavaScript. +//! +//! The sqlite event log stores raw (un-enriched) events — the DB never needs +//! migration when the enrichment logic changes. Enrichment is applied at +//! SSE-emit time in [`crate::web_ui::stream`] so both the live tail +//! (`events/stream`) and the history replay (`events/history`) endpoints +//! deliver the same enriched shape. +//! +//! # Migration (two-phase) +//! +//! **Phase 1** (this change): backend stamps `_icon`/`_summary`/`_category` +//! fields; the client reads them when present and falls back to its own JS +//! tables when absent. Zero user-visible change — a no-op for clients that +//! haven't yet been updated. +//! +//! **Phase 2** (follow-up): the client drops the JS tables once phase 1 is +//! deployed everywhere. + +use serde_json::{Value, json}; + +/// Stamp enrichment fields onto a raw claude stream-json [`Value`]. +/// +/// - `type="system"` events get `_category` + `_summary` (and `_body` for +/// expandable detail, e.g. `commands_changed`). +/// - `type="assistant"` events get `_icon`, `_summary`, and optionally +/// `_category: "rich"` stamped onto each `message.content[]` entry that +/// has `type="tool_use"`. +/// +/// No-ops for unknown/unhandled top-level types. Existing `_`-prefixed fields +/// are left unchanged so the call is idempotent (history replay may hit +/// already-enriched values if the DB is ever pre-populated by a future phase). +pub fn enrich(v: &mut Value) { + match v.get("type").and_then(Value::as_str).unwrap_or("") { + "system" => enrich_system(v), + "assistant" => enrich_assistant(v), + _ => {} + } +} + +// --------------------------------------------------------------------------- +// system events +// --------------------------------------------------------------------------- + +fn enrich_system(v: &mut Value) { + if v.get("_category").is_some() { + return; // idempotent + } + let subtype = v + .get("subtype") + .and_then(Value::as_str) + .unwrap_or("") + .to_owned(); + let (category, summary, body) = system_fields(v, &subtype); + let Some(obj) = v.as_object_mut() else { return }; + obj.insert("_category".to_owned(), json!(category)); + if let Some(s) = summary { + obj.insert("_summary".to_owned(), json!(s)); + } + if let Some(b) = body { + obj.insert("_body".to_owned(), json!(b)); + } +} + +/// Compute `(category, summary, body)` for a `type="system"` event. +/// +/// Categories understood by the frontend: +/// - `"drop"` — client discards without rendering (noise) +/// - `"thinking_tok"` — client updates a single in-place counter row +/// - `"note"` — client renders `_summary` as a terminal note line +/// - `"details"` — client renders a collapsible `
` with `_summary` +/// as the header and `_body` as the expanded text +fn system_fields(v: &Value, subtype: &str) -> (&'static str, Option, Option) { + match subtype { + // Silent startup / stream-end noise — drop. + "init" | "result" | "rate_limit_event" => ("drop", None, None), + + // Live thinking-token counter. + "thinking_tokens" => { + let n = v.get("estimated_tokens").and_then(Value::as_u64); + let s = n.map_or_else( + || "thinking…".to_owned(), + |n| format!("thinking… ~{n} tokens"), + ); + ("thinking_tok", Some(s), None) + } + + // API retry back-off. + "api_retry" => { + let mut parts = vec!["⚠ api retry".to_owned()]; + if let (Some(a), Some(m)) = ( + v.get("attempt").and_then(Value::as_u64), + v.get("max_retries").and_then(Value::as_u64), + ) { + parts.push(format!("{a}/{m}")); + } + if let Some(e) = v.get("error").and_then(Value::as_str) { + parts.push(e.to_owned()); + } else if let Some(s) = v.get("error_status").and_then(Value::as_u64) { + parts.push(format!("HTTP {s}")); + } + if let Some(ms) = v.get("retry_delay_ms").and_then(Value::as_f64) { + parts.push(format!("{:.0}ms", ms.round())); + } + ("note", Some(parts.join(" · ")), None) + } + + // Hard API error. + "api_error" => { + let msg = v + .get("error") + .or_else(|| v.get("message")) + .and_then(Value::as_str) + .map_or_else( + || { + v.get("error_status") + .and_then(Value::as_u64) + .map_or("unknown".to_owned(), |s| format!("HTTP {s}")) + }, + str::to_owned, + ); + ("note", Some(format!("✗ api error · {msg}")), None) + } + + // Plugin (MCP server / slash-command provider) load progress. + "plugin_install" => { + let status = v.get("status").and_then(Value::as_str).unwrap_or("?"); + let label = match status { + "completed" => "✓ done", + "started" => "loading…", + other => other, + }; + ("note", Some(format!("⚙ plugin install · {label}")), None) + } + + // Available slash-command set changed — expandable list. + "commands_changed" => { + let cmds = v.get("commands").and_then(Value::as_array); + let count = cmds.map_or(0, Vec::len); + let summary = format!("⚙ commands changed · {count} available"); + let body = cmds.map(|cmds| { + cmds.iter() + .filter_map(|c| c.get("name").and_then(Value::as_str)) + .map(|n| format!("/{n}")) + .collect::>() + .join("\n") + }); + ("details", Some(summary), body) + } + + // Compaction complete — pre/post token counts + duration. + "compact_boundary" => { + let m = v + .get("compact_metadata") + .cloned() + .unwrap_or_else(|| json!({})); + let mut parts = vec!["⚙ compact".to_owned()]; + if let Some(t) = m.get("trigger").and_then(Value::as_str) { + parts.push(t.to_owned()); + } + if let (Some(pre), Some(post)) = ( + m.get("pre_tokens").and_then(Value::as_u64), + m.get("post_tokens").and_then(Value::as_u64), + ) { + parts.push(format!("{}→{} tokens", fmt_tok(pre), fmt_tok(post))); + } + if let Some(ms) = m.get("duration_ms").and_then(Value::as_u64) { + let dur = if ms < 1000 { + format!("{ms}ms") + } else { + #[allow(clippy::cast_precision_loss)] + let s_f = ms as f64 / 1_000.0; + format!("{s_f:.1}s") + }; + parts.push(dur); + } + ("note", Some(parts.join(" · ")), None) + } + + // Generic "still working" heartbeat tick. + "status" => ("note", Some("⚙ status".to_owned()), None), + + // Unknown subtype — render with the subtype label as a muted note. + other => ("note", Some(format!("⚙ {other}")), None), + } +} + +// --------------------------------------------------------------------------- +// assistant events +// --------------------------------------------------------------------------- + +fn enrich_assistant(v: &mut Value) { + // Navigate message.content[] — absent on text-only turns. + let Some(content) = v + .get_mut("message") + .and_then(|m| m.get_mut("content")) + .and_then(Value::as_array_mut) + else { + return; + }; + for entry in content.iter_mut() { + enrich_tool_use_entry(entry); + } +} + +fn enrich_tool_use_entry(entry: &mut Value) { + if entry.get("type").and_then(Value::as_str) != Some("tool_use") { + return; + } + if entry.get("_icon").is_some() { + return; // idempotent + } + let name = entry + .get("name") + .and_then(Value::as_str) + .unwrap_or("") + .to_owned(); + let input = entry.get("input").cloned().unwrap_or_else(|| json!({})); + let icon = tool_icon(&name); + let summary = fmt_tool_use(&name, &input); + let rich = is_rich_tool(&name); + let Some(obj) = entry.as_object_mut() else { + return; + }; + obj.insert("_icon".to_owned(), json!(icon)); + obj.insert("_summary".to_owned(), json!(summary)); + if rich { + obj.insert("_category".to_owned(), json!("rich")); + } +} + +// --------------------------------------------------------------------------- +// tool helpers +// --------------------------------------------------------------------------- + +/// Whether this tool is rendered by the frontend's *rich* renderer (diff +/// view, body expansion) rather than the flat `_summary` row. Flagged with +/// `_category: "rich"` so the client can distinguish without re-implementing +/// the tool name list. +fn is_rich_tool(name: &str) -> bool { + matches!( + name, + "Write" + | "Edit" + | "mcp__hyperhive__send" + | "mcp__hyperhive__ask" + | "mcp__hyperhive__answer" + ) +} + +fn tool_icon(name: &str) -> &'static str { + // Exact-match table first, then prefix/contains fallbacks. + match name { + "mcp__hyperhive__send" => "📤", + "mcp__hyperhive__recv" => "📥", + "mcp__hyperhive__ask" => "❓", + "mcp__hyperhive__answer" => "✍️", + "mcp__hyperhive__remind" => "⏰", + "mcp__hyperhive__set_status" => "🏷️", + "mcp__hyperhive__get_loose_ends" => "🪢", + "mcp__hyperhive__cancel_loose_end" => "✂️", + "mcp__hyperhive__ack_until" => "✅", + "mcp__hyperhive__get_agent_meta" => "ℹ️", + "mcp__hyperhive__request_next_turn" => "⏩", + "mcp__hyperhive__restart" => "↻", + "mcp__hyperhive__kill" => "⏹️", + "mcp__hyperhive__start" => "▶️", + "mcp__hyperhive__update" => "🔄", + "mcp__hyperhive__list_containers" + | "mcp__matrix__list_rooms" + | "mcp__matrix__list_room_members" + | "mcp__matrix__list_invites" => "📋", + "mcp__hyperhive__get_logs" | "mcp__hyperhive__get_host_journal" => "📜", + "mcp__matrix__read_room" | "Read" => "📖", + "mcp__matrix__mark_read" => "👁️", + "mcp__bash__kill" => "🛑", + "Write" => "💾", + "Edit" => "✏️", + "Glob" | "Grep" => "🔍", + _ => tool_icon_fallback(name), + } +} + +fn tool_icon_fallback(name: &str) -> &'static str { + if name.starts_with("mcp__matrix__") { + "💬" + } else if name.starts_with("mcp__bash__") { + "🖥️" + } else if name.contains("schedule") { + "⏱️" + } else if name.starts_with("mcp__hyperhive__request_") { + "📦" + } else { + "🔧" + } +} + +// --------------------------------------------------------------------------- +// tool_use summary formatter +// --------------------------------------------------------------------------- + +#[allow(clippy::too_many_lines)] +fn fmt_tool_use(name: &str, input: &Value) -> String { + // Short name: strip the MCP server prefix for display. + let short = if let Some(rest) = name.strip_prefix("mcp__hyperhive__") { + format!("{rest}*") + } else if let Some(rest) = name.strip_prefix("mcp__bash__") { + format!("{rest}*") + } else if let Some(rest) = name.strip_prefix("mcp__matrix__") { + format!("{rest}*") + } else { + name.to_owned() + }; + + match name { + "Read" | "Write" | "Edit" => format!("{short} {}", sv(input, "file_path")), + "Glob" | "Grep" => format!("{short} {}", sv(input, "pattern")), + "Bash" => { + let bg = if input + .get("run_in_background") + .and_then(Value::as_bool) + .unwrap_or(false) + { + " [bg]" + } else { + "" + }; + format!("{short}{bg} $ {}", sv(input, "command")) + } + "TodoWrite" => { + let n = input + .get("todos") + .and_then(Value::as_array) + .map_or(0, Vec::len); + format!("{short} ({n} items)") + } + "mcp__hyperhive__send" => { + let body = trim_str(&sv(input, "body"), 80); + format!("{short} → {}: {}", sv(input, "to"), json_str(&body)) + } + "mcp__hyperhive__recv" => { + let mut parts = Vec::new(); + if let Some(w) = input.get("wait_seconds").and_then(Value::as_u64) { + parts.push(format!("wait {w}s")); + } + if let Some(m) = input.get("max").and_then(Value::as_u64) { + parts.push(format!("max {m}")); + } + if parts.is_empty() { + format!("{short}()") + } else { + format!("{short} {}", parts.join(" · ")) + } + } + "mcp__hyperhive__kill" + | "mcp__hyperhive__restart" + | "mcp__hyperhive__start" + | "mcp__hyperhive__update" + | "mcp__hyperhive__request_init_config" => format!("{short} {}", sv(input, "name")), + "mcp__hyperhive__ack_until" => { + let up_to = input + .get("up_to") + .and_then(Value::as_u64) + .map_or_else(|| "?".to_owned(), |n| n.to_string()); + format!("{short} ≤{up_to}") + } + "mcp__hyperhive__get_logs" => { + let lines = input + .get("lines") + .and_then(Value::as_u64) + .map(|n| format!(" · {n}L")) + .unwrap_or_default(); + format!("{short} {}{lines}", sv(input, "agent")) + } + "mcp__hyperhive__get_host_journal" => { + let mut parts = Vec::new(); + if let Some(c) = input.get("container").and_then(Value::as_str) { + parts.push(c.to_owned()); + } else if let Some(u) = input.get("unit").and_then(Value::as_str) { + parts.push(u.to_owned()); + } + if let Some(g) = input.get("grep").and_then(Value::as_str) { + parts.push(format!("/{g}/")); + } + if let Some(l) = input.get("lines").and_then(Value::as_u64) { + parts.push(format!("{l}L")); + } + if parts.is_empty() { + format!("{short}()") + } else { + format!("{short} {}", parts.join(" · ")) + } + } + "mcp__hyperhive__remind" => { + let when = if let Some(s) = input.get("delay_seconds").and_then(Value::as_u64) { + if s < 60 { + format!("+{s}s") + } else if s < 3_600 { + format!("+{}m", s / 60) + } else { + #[allow(clippy::cast_precision_loss)] + let h_f = s as f64 / 3_600.0; + format!("+{h_f:.1}h") + } + } else if let Some(ts) = input.get("at_unix_timestamp").and_then(Value::as_u64) { + let h = (ts % 86_400) / 3_600; + let m = (ts % 3_600) / 60; + format!("at {h:02}:{m:02}Z") + } else { + String::new() + }; + let msg_raw = input + .get("message") + .or_else(|| input.get("file_path")) + .and_then(Value::as_str) + .unwrap_or(""); + let msg = trim_str(&msg_raw.replace(char::is_whitespace, " "), 60); + let when_part = if when.is_empty() { + String::new() + } else { + format!(" {when}") + }; + let msg_part = if msg.is_empty() { + String::new() + } else { + format!(" \"{msg}\"") + }; + format!("{short}{when_part}{msg_part}") + } + "mcp__hyperhive__request_update_meta_inputs" => { + let ins = match input.get("inputs").and_then(Value::as_array) { + Some(arr) if !arr.is_empty() => { + let names: Vec<&str> = arr.iter().filter_map(Value::as_str).take(4).collect(); + let tail = if arr.len() > 4 { ", …" } else { "" }; + format!("[{}{}]", names.join(", "), tail) + } + _ => "all".to_owned(), + }; + format!("{short} {ins}") + } + "mcp__hyperhive__list_schedules" => format!("{short}()"), + "mcp__hyperhive__cancel_schedule" => { + let id = input + .get("id") + .and_then(Value::as_u64) + .map_or_else(|| "?".to_owned(), |n| n.to_string()); + let tgts = match input.get("targets").and_then(Value::as_array) { + Some(arr) if !arr.is_empty() => { + let names: Vec<&str> = arr.iter().filter_map(Value::as_str).collect(); + format!(" [{}]", names.join(", ")) + } + _ => " all".to_owned(), + }; + format!("{short} #{id}{tgts}") + } + "mcp__hyperhive__fire_schedule_now" => { + let id = input + .get("id") + .and_then(Value::as_u64) + .map_or_else(|| "?".to_owned(), |n| n.to_string()); + format!("{short} #{id}") + } + "mcp__hyperhive__edit_schedule" => { + let id = input + .get("id") + .and_then(Value::as_u64) + .map_or_else(|| "?".to_owned(), |n| n.to_string()); + let mut parts = vec![format!("#{id}")]; + if input.get("body").is_some() { + parts.push("body".to_owned()); + } + if input.get("interval_seconds").is_some() { + parts.push("interval".to_owned()); + } + if input.get("next_fire_at_unix").is_some() { + parts.push("next".to_owned()); + } + if let Some(a) = input.get("targets_add").and_then(Value::as_array) + && !a.is_empty() + { + parts.push(format!("+{} tgt", a.len())); + } + if let Some(r) = input.get("targets_remove").and_then(Value::as_array) + && !r.is_empty() + { + parts.push(format!("-{} tgt", r.len())); + } + format!("{short} {}", parts.join(" · ")) + } + "mcp__hyperhive__request_schedule_prompt" => { + let tgts = match input.get("targets").and_then(Value::as_array) { + Some(arr) => arr + .iter() + .filter_map(Value::as_str) + .collect::>() + .join(", "), + None => "?".to_owned(), + }; + let when = input + .get("first_fire_at_unix") + .and_then(Value::as_u64) + .map_or_else( + || "?".to_owned(), + |ts| { + let h = (ts % 86_400) / 3_600; + let m = (ts % 3_600) / 60; + format!("{h:02}:{m:02}Z") + }, + ); + let recur = input + .get("interval_seconds") + .and_then(Value::as_u64) + .map(|s| format!(" +{s}s")) + .unwrap_or_default(); + format!("{short} → {tgts} at {when}{recur}") + } + "mcp__bash__run" => { + let cmd = sv(input, "cmd"); + let first = cmd.lines().next().unwrap_or("").trim().to_owned(); + format!("{short} $ {}", trim_str(&first, 72)) + } + "mcp__bash__status" => { + let wait = input + .get("wait_seconds") + .and_then(Value::as_u64) + .map(|w| format!(" · wait {w}s")) + .unwrap_or_default(); + format!("{short} id:{}{wait}", sv(input, "id")) + } + "mcp__bash__kill" => { + let force = if input.get("force").and_then(Value::as_bool).unwrap_or(false) { + " [force]" + } else { + "" + }; + format!("{short} {}{force}", sv(input, "id")) + } + "mcp__hyperhive__set_status" => { + format!("{short} \"{}\"", trim_str(&sv(input, "text"), 60)) + } + "mcp__hyperhive__get_loose_ends" => { + let agent = input + .get("agent") + .and_then(Value::as_str) + .map_or_else(|| "()".to_owned(), |a| format!(" [{a}]")); + format!("{short}{agent}") + } + "mcp__hyperhive__get_agent_meta" => { + let name_part = input + .get("name") + .and_then(Value::as_str) + .map_or_else(|| "()".to_owned(), |n| format!(" {n}")); + format!("{short}{name_part}") + } + "mcp__hyperhive__cancel_loose_end" => { + let id = input + .get("id") + .and_then(Value::as_u64) + .map_or_else(|| "?".to_owned(), |n| n.to_string()); + format!("{short} {} #{id}", sv(input, "kind")) + } + "mcp__matrix__read_room" => { + let limit = input + .get("limit") + .and_then(Value::as_u64) + .map(|l| format!(" [{l}]")) + .unwrap_or_default(); + format!("{short} {}{limit}", fmt_room(&sv(input, "room"))) + } + "mcp__matrix__mark_read" | "mcp__matrix__join_room" | "mcp__matrix__download_file" => { + format!("{short} {}", fmt_room(&sv(input, "room"))) + } + "mcp__matrix__send_message" | "mcp__matrix__send_reply" => format!( + "{short} → {}: {}", + fmt_room(&sv(input, "room")), + json_str(&trim_str(&sv(input, "body"), 50)) + ), + "mcp__matrix__send_dm" => format!( + "{short} → {}: {}", + fmt_user(&sv(input, "user_id")), + json_str(&trim_str(&sv(input, "body"), 50)) + ), + "mcp__matrix__send_reaction" => { + format!( + "{short} {} {}", + fmt_room(&sv(input, "room")), + sv(input, "key") + ) + } + "mcp__matrix__open_dm" => format!("{short} {}", fmt_user(&sv(input, "user_id"))), + "mcp__matrix__invite_user" => format!( + "{short} {} → {}", + fmt_user(&sv(input, "user_id")), + fmt_room(&sv(input, "room")) + ), + _ => fmt_args_generic(&short, input), + } +} + +// --------------------------------------------------------------------------- +// generic fallback formatter (port of JS fmtArgsGeneric) +// --------------------------------------------------------------------------- + +fn fmt_args_generic(name: &str, input: &Value) -> String { + let Some(obj) = input.as_object() else { + return format!("{name}()"); + }; + if obj.is_empty() { + return format!("{name}()"); + } + let keys: Vec<&String> = obj.keys().collect(); + if keys.len() == 1 { + let k = keys[0]; + let v = &obj[k]; + if let Some(s) = v.as_str() { + return format!("{name} {k}: {}", json_str(&trim_str(s, 100))); + } + if v.is_number() || v.is_boolean() { + return format!("{name} {k}: {v}"); + } + } + let pretty: Vec = keys + .iter() + .take(4) + .map(|k| { + let v = &obj[*k]; + if v.is_null() { + format!("{k}: null") + } else if let Some(s) = v.as_str() { + format!("{k}: {}", json_str(&trim_str(s, 40))) + } else if v.is_number() || v.is_boolean() { + format!("{k}: {v}") + } else if let Some(arr) = v.as_array() { + format!("{k}: [{}]", arr.len()) + } else { + format!("{k}: {{…}}") + } + }) + .collect(); + let tail = if keys.len() > 4 { + format!(" …+{}", keys.len() - 4) + } else { + String::new() + }; + format!("{name} {}{tail}", pretty.join(" · ")) +} + +// --------------------------------------------------------------------------- +// string utilities +// --------------------------------------------------------------------------- + +/// Format a token count as "1k", "1.2M", etc. (mirrors JS `fmtTok`). +fn fmt_tok(n: u64) -> String { + if n >= 1_000_000 { + // Divide before casting to keep the value in a range where f64 + // has enough precision for one decimal place (≤ ~9×10^12 tokens). + #[allow(clippy::cast_precision_loss)] + let m = (n / 1_000) as f64 / 1_000.0; + format!("{m:.1}M") + } else if n >= 1_000 { + format!("{}k", n / 1_000) + } else { + n.to_string() + } +} + +/// Shorten a matrix room id/alias for display. +fn fmt_room(r: &str) -> String { + if r.starts_with('!') { + // Room id: keep only the local part before the colon (up to 8 chars). + let end = r.find(':').unwrap_or(r.len()).min(9); + r[..end].to_owned() + } else if r.starts_with('#') { + // Alias: keep `#name` part before the server. + r.split(':').next().unwrap_or(r).to_owned() + } else { + trim_str(r, 20) + } +} + +/// Shorten `@user:server` → `@user`. +fn fmt_user(u: &str) -> String { + let end = u.find(':').unwrap_or(u.len()); + u[..end].to_owned() +} + +/// Extract a string field from a JSON object, defaulting to `""`. +fn sv(v: &Value, key: &str) -> String { + v.get(key).and_then(Value::as_str).unwrap_or("").to_owned() +} + +/// Trim to `max` *characters* (not bytes), collapsing whitespace first. +fn trim_str(s: &str, max: usize) -> String { + let collapsed: String = s.split_whitespace().collect::>().join(" "); + let mut chars = collapsed.chars(); + let trimmed: String = chars.by_ref().take(max).collect(); + if chars.next().is_some() { + format!("{trimmed}…") + } else { + trimmed + } +} + +/// Serialize a string as a JSON string literal (e.g. `"hello"`) for display +/// in summary lines. Falls back to the raw string if serialization fails. +fn json_str(s: &str) -> String { + serde_json::to_string(s).unwrap_or_else(|_| s.to_owned()) +} diff --git a/hive-agent/src/web_ui/stream.rs b/hive-agent/src/web_ui/stream.rs index e94a520a..24f855a9 100644 --- a/hive-agent/src/web_ui/stream.rs +++ b/hive-agent/src/web_ui/stream.rs @@ -40,6 +40,17 @@ pub(super) async fn events_history( }; let (events, min_id, has_more) = state.bus.history_page(before, limit); + // Apply the same enrichment as the live SSE path so history replay + // and live tail deliver identical shapes. The DB stores raw events. + let events: Vec<_> = events + .into_iter() + .map(|mut se| { + if let crate::events::LiveEvent::Stream(ref mut v) = se.event { + crate::stream_enrich::enrich(v); + } + se + }) + .collect(); let mut resp = serde_json::json!({ "events": events, "min_id": min_id, @@ -68,7 +79,15 @@ pub(super) async fn events_stream( .unwrap_or_default(), ); let live = BroadcastStream::new(rx).filter_map(|res| { - let ev = res.ok()?; + let mut ev = res.ok()?; + // Enrich stream-json values with pre-computed display fields + // (`_icon`, `_summary`, `_category`) so the frontend doesn't need to + // duplicate the dispatch logic. The DB stores raw events; enrichment + // is applied here so both the live tail and the history endpoint + // deliver the same shape (see `events_history` below). + if let crate::events::LiveEvent::Stream(ref mut v) = ev.event { + crate::stream_enrich::enrich(v); + } let json = serde_json::to_string(&ev).ok()?; Some(Ok(Event::default().data(json))) });