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

Simplify terminal message shape to a uniform TermMsg

Browse source 
Move terminal-row classification server-side into a new
hive-agent/src/term_msg.rs, replacing the old JSON-mutation
enrich()/stamped-field approach in stream_enrich.rs with one
uniform wire shape: {icon?, level: debug|info|warn|error, summary,
body?, body_format?: markdown|diff, coalesce_key?}. No more per-row
`kind` tag or raw claude-JSON passthrough — every row is the same
shape, with structural identity carried by icon + summary text
instead of a CSS class per row kind.

hive-agent/src/web_ui/stream.rs's history + SSE endpoints now both
call term_msg::classify() and serve TermEnvelope{ts, seq?, msgs}
frames; events that classify to zero rows (agent-state changes,
drop-noise) never reach the wire.

Frontend: classifyEvent.ts collapses from a large per-tool dispatch
tree to a thin TermMsg -> StreamRow adapter. streamRow.ts/Row.tsx
drop the now-dead meta/childText fields. terminal.css switches from
a dozen-odd per-row-kind classes to four level-based color rules.
Expand/collapse of a bodied row is now a uniform client-side
decision (the operator's preference), no server-side per-tool
override.

docs/terminal-rendering.md rewritten to match.
This commit is contained in:
iris 2026-08-30 21:11:22 +02:00
commit 5eefaa951d
13 changed files with 886 additions and 628 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -28,6 +28,7 @@ mod serve_common;
mod state_entry_watch;
mod stats;
mod stream_enrich;
mod term_msg;
mod todo_server;
mod todos;
mod turn;

363
hive-agent/src/stream_enrich.rs
View file

@ -1,75 +1,97 @@
//! Enrich raw claude stream-json values before SSE delivery.
//! Classify raw claude stream-json values into [`crate::term_msg::TermMsg`]
//! rows 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.
//! [`classify_stream_value`] is the entry point — it walks one raw claude
//! `stream-json` line (the payload of a [`crate::events::LiveEvent::Stream`])
//! and returns zero or more terminal rows. 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 classified
//! shape — the sqlite event log stores the raw, unclassified event, so the
//! DB never needs migration when classification logic changes.
//!
//! 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.
//! The per-tool icon/summary formatting below (`tool_icon`, `fmt_tool_use`
//! and its per-family helpers, `rich_tool_body`) is reused as-is from
//! before mara's terminal-message redesign — that logic (what does
//! `mcp__hyperhive__send`'s row say, which tools get an expandable body)
//! didn't change; only the shape it gets packed into did.
use crate::term_msg::{BodyFormat, ClassifyCtx, Level, TermMsg};
use serde_json::{Value, json};
/// Stamp enrichment fields onto a raw claude stream-json [`Value`].
/// Classify one raw claude stream-json line into zero or more terminal rows.
///
/// - `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("") {
// Top-level result/rate_limit_event are drop-category noise — stamp
// the same category the frontend uses to silently discard them.
"result" | "rate_limit_event" => {
if let Some(obj) = v.as_object_mut() {
obj.entry("_category").or_insert_with(|| json!("drop"));
}
}
"system" => enrich_system(v),
"assistant" => enrich_assistant(v),
_ => {}
/// Dispatch order mirrors the old client-side `classifyEvent.ts` (before
/// this classification moved server-side):
/// top-level drop-noise types first, then `type="system"`, then task events
/// (matched on `subtype` regardless of `type`), then `assistant`/`user`
/// content, with an unrecognised shape falling through to a loud
/// warn-level catch-all so a silently-dropped event type stays visible.
pub fn classify_stream_value(v: &Value, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
let vtype = v.get("type").and_then(Value::as_str).unwrap_or("");
if matches!(vtype, "result" | "rate_limit_event") {
return vec![];
}
if vtype == "system" {
return classify_system(v);
}
let subtype = v.get("subtype").and_then(Value::as_str);
if matches!(subtype, Some("task_started" | "task_notification")) {
return classify_task_event(v).into_iter().collect();
}
if vtype == "assistant" {
return v
.get("message")
.and_then(|m| m.get("content"))
.and_then(Value::as_array)
.map_or_else(Vec::new, |content| classify_assistant_content(content, ctx));
}
if vtype == "user" {
return v
.get("message")
.and_then(|m| m.get("content"))
.and_then(Value::as_array)
.map_or_else(Vec::new, |content| classify_user_content(content, ctx));
}
vec![TermMsg::new(Level::Warn, trim_str(&v.to_string(), 200)).icon("!")]
}
// ---------------------------------------------------------------------------
// 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));
fn classify_system(v: &Value) -> Vec<TermMsg> {
let subtype = v.get("subtype").and_then(Value::as_str).unwrap_or("");
let (category, summary, body) = system_fields(v, subtype);
match category {
"drop" => vec![],
"thinking_tok" => vec![
TermMsg::new(
Level::Debug,
summary.unwrap_or_else(|| "thinking…".to_owned()),
)
.icon("🧠")
.coalesce("thinking-tok"),
],
"details" => {
let mut m = TermMsg::new(Level::Debug, summary.unwrap_or_default());
if let Some(b) = body {
m = m.body(b, None);
}
vec![m]
}
// "note" category — ambient harness/system chatter. Level + coalesce
// key vary by subtype; everything else defaults to a plain debug note.
_ => {
let s = summary.unwrap_or_default();
let m = match subtype {
"api_error" => TermMsg::new(Level::Error, s),
"api_retry" => TermMsg::new(Level::Warn, s),
"plugin_install" => TermMsg::new(Level::Debug, s).coalesce("plugin-install"),
"status" => TermMsg::new(Level::Debug, s).coalesce("status-tick"),
_ => TermMsg::new(Level::Debug, s),
};
vec![m]
}
}
}
@ -197,51 +219,196 @@ fn system_fields(v: &Value, subtype: &str) -> (&'static str, Option<String>, Opt
}
// ---------------------------------------------------------------------------
// assistant events
// assistant events (claude's own output: text / thinking / tool calls)
// ---------------------------------------------------------------------------
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;
fn classify_assistant_content(content: &[Value], ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
let mut rows = Vec::new();
for c in content {
match c.get("type").and_then(Value::as_str) {
Some("text") => {
let text = c.get("text").and_then(Value::as_str).unwrap_or("");
if !text.trim().is_empty() {
// No separate summary line — the markdown body is the
// whole row, matching the old `.text` row shape.
rows.push(
TermMsg::new(Level::Info, String::new())
.body(text.to_owned(), Some(BodyFormat::Markdown)),
);
}
}
Some("thinking") => {
let txt = c
.get("thinking")
.or_else(|| c.get("text"))
.and_then(Value::as_str)
.unwrap_or("")
.trim()
.to_owned();
let summary = if txt.is_empty() {
"thinking …".to_owned()
} else {
txt
};
rows.push(TermMsg::new(Level::Debug, summary).icon("💭"));
}
Some("tool_use") => {
if let (Some(id), Some(name)) = (
c.get("id").and_then(Value::as_str),
c.get("name").and_then(Value::as_str),
) {
ctx.record_tool_use(id, name);
}
rows.push(classify_tool_use(c));
}
_ => {}
}
}
rows
}
/// `_category === 'rich'` tools used to get an expandable row (diff body
/// for Edit, default-open markdown for send, plain for everything else with
/// a body). That's now just "does this row have a body" — `body.is_some()`
/// on the returned [`TermMsg`] *is* the expandable signal, no separate flag.
fn classify_tool_use(c: &Value) -> TermMsg {
let name = c.get("name").and_then(Value::as_str).unwrap_or("");
let input = c.get("input").cloned().unwrap_or_else(|| json!({}));
let mut m = TermMsg::new(Level::Info, fmt_tool_use(name, &input)).icon(tool_icon(name));
if let Some((body, body_type)) = rich_tool_body(name, &input) {
let format = match body_type {
"diff" => Some(BodyFormat::Diff),
"markdown" => Some(BodyFormat::Markdown),
_ => None, // "plain"
};
m = m.body(body, format);
}
m
}
// ---------------------------------------------------------------------------
// user events (tool_result — claude's own tool calls answered)
// ---------------------------------------------------------------------------
fn classify_user_content(content: &[Value], ctx: &ClassifyCtx) -> Vec<TermMsg> {
content
.iter()
.filter(|c| c.get("type").and_then(Value::as_str) == Some("tool_result"))
.map(|c| classify_tool_result(c, ctx))
.collect()
}
/// `<tool_use_error>…</tool_use_error>` is claude's own wrapper on failed
/// tool calls — implementation detail, adds nothing for the operator.
fn strip_tool_use_error_wrapper(s: &str) -> String {
let trimmed = s.trim();
trimmed
.strip_prefix("<tool_use_error>")
.and_then(|rest| rest.strip_suffix("</tool_use_error>"))
.map_or_else(|| trimmed.to_owned(), |inner| inner.trim().to_owned())
}
fn classify_tool_result(c: &Value, ctx: &ClassifyCtx) -> TermMsg {
let raw_txt = match c.get("content") {
Some(Value::Array(parts)) => parts
.iter()
.filter_map(|p| p.get("text").and_then(Value::as_str))
.collect::<String>(),
Some(Value::String(s)) => s.clone(),
_ => String::new(),
};
for entry in content.iter_mut() {
enrich_tool_use_entry(entry);
let is_error = c.get("is_error").and_then(Value::as_bool).unwrap_or(false);
let txt = if is_error {
strip_tool_use_error_wrapper(&raw_txt)
} else {
raw_txt
};
let tool_use_id = c.get("tool_use_id").and_then(Value::as_str);
let source_name = tool_use_id.and_then(|id| ctx.tool_name(id));
let is_message_bearing = source_name == Some("mcp__hyperhive__recv");
let trimmed: String = txt.split_whitespace().collect::<Vec<_>>().join(" ");
let summary = summarize_tool_result(&txt, &trimmed);
let short = txt.trim().is_empty() || txt.chars().count() <= 120;
if is_error {
let m = TermMsg::new(Level::Error, summary);
return if short {
m.icon("✗")
} else {
m.body(txt, None)
};
}
if is_message_bearing && !txt.trim().is_empty() {
return TermMsg::new(Level::Info, format!("recv ← {summary}"))
.body(txt, Some(BodyFormat::Markdown));
}
let m = TermMsg::new(Level::Info, summary);
if short {
m.icon("←")
} else {
m.body(txt, None)
}
}
fn enrich_tool_use_entry(entry: &mut Value) {
if entry.get("type").and_then(Value::as_str) != Some("tool_use") {
return;
/// `(empty)` / a short trimmed line / `"NL · headline…"` for a long one —
/// matches the old client-side `summaryBody` computation exactly.
fn summarize_tool_result(txt: &str, trimmed: &str) -> String {
if trimmed.is_empty() {
return "(empty)".to_owned();
}
if entry.get("_icon").is_some() {
return; // idempotent
if trimmed.chars().count() <= 120 {
return trimmed.to_owned();
}
let name = entry
.get("name")
let lines = txt.lines().filter(|l| !l.is_empty()).count();
let headline: String = trimmed.chars().take(90).collect();
format!("{lines}L · {headline}…")
}
// ---------------------------------------------------------------------------
// subagent (claude Task-tool) activity — dead path for agents today (`Task`
// is omitted from the allow-list) but kept for parity with the old
// client-side `classifyTaskEvent`.
// ---------------------------------------------------------------------------
fn classify_task_event(v: &Value) -> Option<TermMsg> {
let id: String = v
.get("task_id")
.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 body = rich_tool_body(&name, &input);
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"));
}
if let Some((b, bt)) = body {
obj.insert("_body".to_owned(), json!(b));
obj.insert("_body_type".to_owned(), json!(bt));
.chars()
.take(8)
.collect();
let kind = v
.get("task_type")
.and_then(Value::as_str)
.map(|t| format!(" [{t}]"))
.unwrap_or_default();
let desc = v
.get("description")
.or_else(|| v.get("summary"))
.and_then(Value::as_str)
.unwrap_or("(no description)");
match v.get("subtype").and_then(Value::as_str) {
Some("task_started") => {
Some(TermMsg::new(Level::Info, format!("task {id} started · {desc}{kind}")).icon("⌁"))
}
Some("task_notification") => {
let status = v.get("status").and_then(Value::as_str).unwrap_or("unknown");
let (glyph, level) = match status {
"completed" => ("✓", Level::Info),
"failed" => ("✗", Level::Error),
_ => ("◌", Level::Info),
};
let out = v
.get("output_file")
.and_then(Value::as_str)
.map(|f| format!(" · → {f}"))
.unwrap_or_default();
Some(TermMsg::new(level, format!("task {id} {glyph} {status} · {desc}{out}")).icon("⌁"))
}
_ => None,
}
}
@ -249,14 +416,6 @@ fn enrich_tool_use_entry(entry: &mut Value) {
// 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, "Edit" | "mcp__bash__run" | "mcp__hyperhive__send")
}
/// Pre-compute the expandable body for rich tool entries.
///
/// Returns `Some((body, body_type))` where `body_type` tells the frontend

285
hive-agent/src/term_msg.rs Normal file
View file

@ -0,0 +1,285 @@
//! Terminal-message wire shape: what the per-agent web UI's live/history
//! endpoints actually serve for the "terminal" event stream, as opposed to
//! agent-state changes (`StatusChanged`/`ModelChanged`/`EffortChanged`/
//! `TokenUsageChanged`/`TurnStateChanged`), which have never rendered as
//! terminal rows (the header/badges poll `/api/state`, not this stream) and
//! produce zero [`TermMsg`]s here.
//!
//! One [`crate::events::LiveEvent`] maps to zero or more `TermMsg`s — most
//! map to exactly one, but `LiveEvent::Stream` (one raw claude
//! `stream-json` line) can expand to several: an `assistant` message with
//! both a text block and a `tool_use` block produces two rows.
//!
//! Design history: mara's terminal-message redesign (six rounds of
//! negotiation on the forge) collapsed what was an 11-field frontend-side
//! row shape (`StreamRow`, `frontend/packages/agent/src/lib/streamRow.ts`)
//! plus raw claude-JSON passthrough into this 6-field shape, with
//! classification moved server-side so the client-side `classifyEvent.ts` —
//! a large per-tool dispatch table — mostly goes away. `kind`, `unread`,
//! `from`, and `expanded_default` were all considered and dropped along the
//! way; `level` replaces free-text CSS-class styling.
use serde::Serialize;
use std::collections::HashMap;
use crate::events::LiveEvent;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Level {
/// Low-signal / ambient chatter — thinking, progress ticks, harness
/// housekeeping. The client's default rendering can dim/de-emphasize
/// these without hiding them outright.
Debug,
/// Routine substantive content — turn boundaries, assistant text, tool
/// calls/results, message bodies.
Info,
/// Heads-up, not necessarily broken — stderr lines, an unclassified
/// event shape landing (the old `.sys` catch-all), API retries.
Warn,
/// Something actually failed — a turn ending non-ok, a tool result with
/// `is_error: true`.
Error,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum BodyFormat {
Markdown,
Diff,
}
/// One terminal row. `body_format: None` with `body: Some(_)` means plain
/// text (the common case — no explicit tag on the wire for it, same logic
/// as `body` itself being absent meaning "nothing to expand").
#[derive(Debug, Clone, Serialize)]
pub struct TermMsg {
#[serde(skip_serializing_if = "Option::is_none")]
pub icon: Option<String>,
pub level: Level,
pub summary: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub body: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub body_format: Option<BodyFormat>,
#[serde(skip_serializing_if = "Option::is_none")]
pub coalesce_key: Option<String>,
}
impl TermMsg {
pub fn new(level: Level, summary: impl Into<String>) -> Self {
Self {
icon: None,
level,
summary: summary.into(),
body: None,
body_format: None,
coalesce_key: None,
}
}
#[must_use]
pub fn icon(mut self, icon: impl Into<String>) -> Self {
self.icon = Some(icon.into());
self
}
#[must_use]
pub fn body(mut self, body: impl Into<String>, format: Option<BodyFormat>) -> Self {
self.body = Some(body.into());
self.body_format = format;
self
}
#[must_use]
pub fn coalesce(mut self, key: impl Into<String>) -> Self {
self.coalesce_key = Some(key.into());
self
}
}
/// Per-connection/per-request classification state. A live SSE stream keeps
/// one of these alive for the connection's lifetime — `tool_use` id → name
/// correlation, so a `tool_result` can tell it's answering a `recv` call and
/// render as a default-open markdown message body. The history endpoint
/// uses a fresh one per page: correlation only works within the page
/// actually returned, not across the live/history boundary. Accepted
/// degradation (same shape as the turn-timestamp fallback documented in
/// `docs/terminal-rendering.md`) — the only user-visible effect is a `recv`
/// result whose `tool_use` fell on the other side of a page/reconnect
/// boundary rendering as a plain block instead of default-open markdown,
/// not a functional loss.
#[derive(Default)]
pub struct ClassifyCtx {
tool_name_by_id: HashMap<String, String>,
}
impl ClassifyCtx {
pub fn record_tool_use(&mut self, id: &str, name: &str) {
self.tool_name_by_id.insert(id.to_owned(), name.to_owned());
}
#[must_use]
pub fn tool_name(&self, id: &str) -> Option<&str> {
self.tool_name_by_id.get(id).map(String::as_str)
}
}
/// Classify one [`LiveEvent`] into zero or more terminal rows.
pub fn classify(ev: &LiveEvent, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
match ev {
LiveEvent::TurnStart { from, body, .. } => {
// `unread` (the third field) was dropped — mara: "i can see
// the todo/inbox count in toolbar and dont need the N unread".
let mut m = TermMsg::new(Level::Info, format!("TURN ← {from}")).icon("◆");
if !body.trim().is_empty() {
m = m.body(body.clone(), None);
}
vec![m]
}
LiveEvent::TurnEnd { ok, note } => {
let msg = if *ok {
TermMsg::new(Level::Info, "turn ok").icon("✅")
} else {
let summary = note
.as_deref()
.filter(|n| !n.is_empty())
.map_or_else(|| "turn fail".to_owned(), |n| format!("turn fail — {n}"));
TermMsg::new(Level::Error, summary).icon("❌")
};
vec![msg]
}
LiveEvent::Note { text } => vec![classify_note(text)],
LiveEvent::Stream(v) => crate::stream_enrich::classify_stream_value(v, ctx),
// Agent-state transitions never render as terminal rows — the
// header/badges read `/api/state`, not this stream (see module doc).
LiveEvent::StatusChanged { .. }
| LiveEvent::ModelChanged { .. }
| LiveEvent::EffortChanged { .. }
| LiveEvent::TokenUsageChanged { .. }
| LiveEvent::TurnStateChanged { .. } => vec![],
}
}
fn classify_note(text: &str) -> TermMsg {
if let Some(rest) = text.strip_prefix("stderr:") {
TermMsg::new(Level::Warn, format!("stderr:{rest}"))
} else if let Some(rest) = text.strip_prefix("operator:") {
TermMsg::new(Level::Info, format!("operator:{rest}"))
} else {
// Ambient harness chatter (session archived, plugin loaded, etc.) —
// routine, not worth the same visual weight as a tool call or
// assistant text.
TermMsg::new(Level::Debug, text.to_owned())
}
}
#[cfg(test)]
mod tests {
use super::{ClassifyCtx, Level, classify};
use crate::events::LiveEvent;
#[test]
fn turn_start_carries_from_and_body_no_unread() {
let ev = LiveEvent::TurnStart {
from: "operator".into(),
body: "sweep the backlog".into(),
unread: 3,
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].summary, "TURN ← operator");
assert_eq!(msgs[0].body.as_deref(), Some("sweep the backlog"));
assert_eq!(msgs[0].level, Level::Info);
}
#[test]
fn turn_start_empty_body_has_no_body() {
let ev = LiveEvent::TurnStart {
from: "reminder".into(),
body: String::new(),
unread: 0,
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert!(msgs[0].body.is_none());
}
#[test]
fn turn_end_ok_is_info() {
let ev = LiveEvent::TurnEnd {
ok: true,
note: None,
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs[0].level, Level::Info);
assert_eq!(msgs[0].summary, "turn ok");
}
#[test]
fn turn_end_fail_is_error_with_note() {
let ev = LiveEvent::TurnEnd {
ok: false,
note: Some("rate limited".into()),
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs[0].level, Level::Error);
assert_eq!(msgs[0].summary, "turn fail — rate limited");
}
#[test]
fn note_stderr_is_warn() {
let ev = LiveEvent::Note {
text: "stderr: warning: deprecated flag".into(),
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs[0].level, Level::Warn);
}
#[test]
fn note_operator_is_info() {
let ev = LiveEvent::Note {
text: "operator: /compact requested".into(),
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs[0].level, Level::Info);
}
#[test]
fn note_plain_is_debug() {
let ev = LiveEvent::Note {
text: "created fresh session".into(),
};
let mut ctx = ClassifyCtx::default();
let msgs = classify(&ev, &mut ctx);
assert_eq!(msgs[0].level, Level::Debug);
}
#[test]
fn agent_state_events_produce_no_rows() {
let mut ctx = ClassifyCtx::default();
assert!(
classify(
&LiveEvent::StatusChanged {
status: "online".into()
},
&mut ctx
)
.is_empty()
);
assert!(
classify(
&LiveEvent::ModelChanged {
model: "opus".into()
},
&mut ctx
)
.is_empty()
);
}
}

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

@ -9,13 +9,36 @@ use serde::{Deserialize, Serialize};
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
use super::AppState;
use crate::term_msg::{ClassifyCtx, Level, TermMsg, classify};
/// One classified envelope on the wire: transport-level metadata (a sibling
/// of the terminal-row payload, not part of it) plus the zero-or-more rows
/// the raw event classified into. An event that classifies to zero rows (an
/// agent-state change — `StatusChanged`/`ModelChanged`/etc. — or
/// drop-category noise) never reaches the wire at all; see
/// `crate::term_msg` for why.
///
/// `seq` is the live per-event dedup counter (`BusEvent::seq`) — `Some` on
/// the SSE path, `None` on history replay (a stored row has no live seq).
/// Same category of plumbing as `ts`: the client already used it to drop
/// buffered live traffic it's about to see again in the initial history
/// page, and that need didn't go away just because rows lost their `kind`
/// tag — dropping it here would silently reintroduce duplicate rows across
/// the live/history boundary.
#[derive(Serialize)]
pub(super) struct TermEnvelope {
ts: i64,
#[serde(skip_serializing_if = "Option::is_none")]
seq: Option<u64>,
msgs: Vec<TermMsg>,
}
/// Response body for `GET /api/events/history`. `seq` is omitted from the
/// wire entirely on a paginated (non-initial) load — matches the old
/// `json!` shape, which only ever set the `"seq"` key when `Some`.
#[derive(Serialize)]
pub(super) struct EventsHistoryBody {
events: Vec<crate::events::StoredEvent>,
events: Vec<TermEnvelope>,
min_id: Option<i64>,
has_more: bool,
#[serde(skip_serializing_if = "Option::is_none")]
@ -52,15 +75,27 @@ 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
// Classify with the same function the live SSE path uses so history
// replay and live tail deliver identical shapes. The DB stores raw
// events; classification is applied at read time here (see
// `crate::term_msg`). One `ClassifyCtx` for the whole page — tool_use→
// name correlation (for default-open `recv` results) only works within
// a single page/connection, not across the live/history boundary; see
// that module's doc for why that's an accepted degradation.
let mut ctx = ClassifyCtx::default();
let events: Vec<TermEnvelope> = events
.into_iter()
.map(|mut se| {
if let crate::events::LiveEvent::Stream(ref mut v) = se.event {
crate::stream_enrich::enrich(v);
.filter_map(|se| {
let msgs = classify(&se.event, &mut ctx);
if msgs.is_empty() {
None
} else {
Some(TermEnvelope {
ts: se.ts,
seq: None,
msgs,
})
}
se
})
.collect();
Json(EventsHistoryBody {
@ -81,23 +116,29 @@ pub(super) async fn events_stream(
// stream rather than emitted to the bus — a bus emit would spam every
// already-connected client with a spurious note each time anyone opens
// the stream.
let hello = Event::default().data(
serde_json::to_string(&crate::events::LiveEvent::Note {
text: "live stream attached".into(),
})
.unwrap_or_default(),
);
let live = BroadcastStream::new(rx).filter_map(|res| {
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` above).
if let crate::events::LiveEvent::Stream(ref mut v) = ev.event {
crate::stream_enrich::enrich(v);
let hello_envelope = TermEnvelope {
ts: chrono::Utc::now().timestamp(),
seq: None,
msgs: vec![TermMsg::new(Level::Debug, "live stream attached")],
};
let hello = Event::default().data(serde_json::to_string(&hello_envelope).unwrap_or_default());
// One `ClassifyCtx` per connection, moved into the closure — tool_use→
// name correlation persists for the connection's lifetime (see
// `crate::term_msg::ClassifyCtx`'s doc for the history-page boundary
// this doesn't cross).
let mut ctx = ClassifyCtx::default();
let live = BroadcastStream::new(rx).filter_map(move |res| {
let ev = res.ok()?;
let msgs = classify(&ev.event, &mut ctx);
if msgs.is_empty() {
return None;
}
let json = serde_json::to_string(&ev).ok()?;
let envelope = TermEnvelope {
ts: ev.ts,
seq: Some(ev.seq),
msgs,
};
let json = serde_json::to_string(&envelope).ok()?;
Some(Ok(Event::default().data(json)))
});
let stream = tokio_stream::once(Ok(hello)).chain(live);