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

hive-agent: add stream_enrich module — stamp _icon/_summary/_category on SSE events

Browse source 
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
This commit is contained in:
iris 2026-07-19 17:32:35 +02:00
commit ae3f011eb3
3 changed files with 733 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
hive-agent/src/main.rs
View file

@ -22,6 +22,7 @@ mod plugins;
mod prompt;
mod serve_common;
mod stats;
mod stream_enrich;
mod turn;
mod turn_stats;
mod vacuum;

712
hive-agent/src/stream_enrich.rs Normal file
View file

@ -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 `<details>` with `_summary`
/// as the header and `_body` as the expanded text
fn system_fields(v: &Value, subtype: &str) -> (&'static str, Option<String>, Option<String>) {
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::<Vec<_>>()
.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::<Vec<_>>()
.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<String> = 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::<Vec<_>>().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())
}

21
hive-agent/src/web_ui/stream.rs
View file

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