zynk 1.0.0

use crate::db_dashboard::escape_html;
use crate::read_model::{DecisionView, FeedEvent};
use std::collections::BTreeMap;
use std::io::Write;

#[derive(Debug, PartialEq, Eq)]
pub enum FeedDelta {
    /// The new feed is `old ++ tail`; render events from this index onward.
    Append(usize),
    /// Not a pure suffix-extension — resend the full snapshot.
    Reset,
}

/// A CONTENT-sensitive per-event key for suffix-diffing (ADR 032 D4). The first
/// component is `event_key`, the table-unique id (`message:{id}` / `status:{id}`),
/// which gives a collision-free identity even for tied RFC3339-second timestamps
/// or a null `summary`. The remaining components are the mutable render fields:
/// because a message's RENDERED proof strip changes IN PLACE when its overlay
/// updates (delivery_status pending→sent→observed, `verified_by`, latest
/// `proof_audit_id`) or its body is redacted (`body` Some→None), those edits keep
/// `event_key` but alter the key here — so `diff_feed` sees a non-prefix and
/// returns `Reset`, pushing the updated proof transition (reset-on-uncertainty).
pub fn feed_key(event: &FeedEvent) -> String {
    format!(
        "{}|{}|{}|{}|{}|{}",
        event.event_key,
        event.delivery_status.as_deref().unwrap_or(""),
        event.verified_by.as_deref().unwrap_or(""),
        event.proof_audit_id.as_deref().unwrap_or(""),
        event.re.as_deref().unwrap_or(""),
        event.body.as_deref().unwrap_or(""),
    )
}

/// The COMBINED per-event diff key the SSE loop uses (M2b R1 P2). `feed_key` keys
/// only the feed ROW's mutable render fields, but a gate/conflict DECISION is an
/// OVERLAY — it lives in the separate `decision_overlay_for_session` map keyed by the
/// work_event id, NOT in the feed row. So a decision added while a client is open
/// leaves the work-event row (and `feed_key`) unchanged; only a heartbeat would fire
/// and the open dashboard would miss the verdict/resolution overlay until reload.
/// This folds the rendered decision state into the key so an overlay's appearance /
/// change flips the key for that feed item -> `diff_feed` sees a non-prefix ->
/// `Reset` (a full re-render that now includes the overlay):
///   (a) the base `feed_key(event)` (row identity + its own mutable render fields);
///   (b) for a gate/conflict WORK_EVENT, its overlay digest from `overlay_map`
///       (the overlay's audit_id + verdict / resolution / notification_status, or
///       empty when undecided), so a not-decided -> decided transition — AND any
///       later re-decision (a NEW audit_id, even with the same verdict but a changed
///       note/actor) or notify-status change on the same work-event — alters the key;
///   (c) for a STANDALONE decision feed item (`event.decision.is_some()`,
///       mode/interrupt/redirect), the decision's `notification_status`, so a
///       not-requested -> sent/failed transition on that card alters the key too.
/// Used for BOTH the `feed-reset` and `feed-append` key vectors so they agree.
pub fn feed_diff_key(event: &FeedEvent, overlay_map: &BTreeMap<i64, DecisionView>) -> String {
    let mut key = feed_key(event);
    // (b) gate/conflict overlay digest — only a `work_events` row whose id is in the
    // overlay map carries one (the map holds only target_work_event_id IS NOT NULL
    // gate/conflict decisions). Mirror the render's lookup: parse `source_id` as i64.
    // The digest leads with the overlay's `audit_id`: a RE-DECISION is latest-wins in
    // the map (same work_event id) but is a NEW operator_decisions row with a NEW
    // audit_id, and `render_decision_overlay` shows the overlay's actor/timestamp/NOTE
    // too — so keying on verdict/resolution/notification alone would miss a same-verdict
    // re-decision that only changed the note/actor/timestamp (P2 residual). Folding in
    // the audit_id makes ANY decision change flip the key -> diff_feed resets -> the
    // streamed feed re-renders with the current overlay. (verdict/resolution/notify are
    // kept in the digest for readability; the audit_id alone already covers them.)
    if event.source_table == "work_events" {
        if let Ok(work_event_id) = event.source_id.parse::<i64>() {
            if let Some(decision) = overlay_map.get(&work_event_id) {
                key.push_str(&format!(
                    "|ov:{}|{}|{}|{}",
                    decision.audit_id,
                    decision.verdict.as_deref().unwrap_or(""),
                    decision.resolution.as_deref().unwrap_or(""),
                    decision.notification_status,
                ));
            }
        }
    }
    // (c) standalone decision item notification status (mode/interrupt/redirect).
    if let Some(decision) = &event.decision {
        key.push_str(&format!("|dn:{}", decision.notification_status));
    }
    key
}

/// Oldest-first feeds: a normal new row makes `old` a prefix of `new`.
pub fn diff_feed(old: &[String], new: &[String]) -> FeedDelta {
    if new.len() >= old.len() && new.starts_with(old) {
        FeedDelta::Append(old.len())
    } else {
        FeedDelta::Reset
    }
}

/// Frame one SSE event with an event name + an HTML payload. SSE `data:` lines are
/// newline-delimited, so multi-line HTML is split into one `data:` line per source
/// line (the browser rejoins them with `\n`). An empty payload still terminates.
///
/// The SSE wire format treats `\r`, `\n`, and `\r\n` as line breaks, so a bare
/// inline `\r` left inside a `data:` line would corrupt framing: `EventSource`
/// would split it and treat the tail as a `data:`-less line (silently dropped).
/// We normalize every break to a `data:` boundary — `\r\n` collapses to one break
/// (no spurious empty `data:` line), and a bare `\r` becomes its own break.
pub fn sse_event(buf: &mut Vec<u8>, event: &str, payload: &str) {
    let _ = writeln!(buf, "event: {event}");
    if payload.is_empty() {
        let _ = writeln!(buf, "data:");
    } else {
        let normalized = payload.replace("\r\n", "\n").replace('\r', "\n");
        for line in normalized.split('\n') {
            let _ = writeln!(buf, "data: {line}");
        }
    }
    let _ = writeln!(buf); // blank line terminates the event
}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum Provenance {
    LiveHerdr,
    DbFallback,
    Unknown,
}

impl Provenance {
    fn label(self) -> &'static str {
        match self {
            Provenance::LiveHerdr => "live-herdr",
            Provenance::DbFallback => "db-fallback",
            Provenance::Unknown => "unknown",
        }
    }
}

#[derive(Debug, Clone)]
pub struct RosterEntry {
    pub agent: String,
    pub address: String,
    pub status: String, // idle/working/blocked/done/unknown
    pub provenance: Provenance,
}

/// db-fallback roster: the session participants with status from the DB
/// projection (not transport-live-authoritative; ADR 032 D5). The `db_status` is
/// the DB roster state (`agents.current_agent_status` / `session_agents.agent_status`),
/// so a status-only session's lead agent shows its projected status, not a
/// hardcoded `unknown` (R1 P3).
pub fn roster_from_db(participants: Vec<(String, String, String)>) -> Vec<RosterEntry> {
    participants
        .into_iter()
        .map(|(agent, address, db_status)| RosterEntry {
            agent,
            address,
            status: db_status,
            provenance: Provenance::DbFallback,
        })
        .collect()
}

pub fn render_roster_html(roster: &[RosterEntry]) -> String {
    let mut html = String::from("<ul class=\"roster\">");
    for e in roster {
        html.push_str(&format!(
            "<li class=\"roster-entry status-{}\"><span class=\"who\">{}</span><span class=\"addr\">{}</span><span class=\"prov\">{}</span></li>",
            escape_html(&e.status),
            escape_html(&e.agent),
            escape_html(&e.address),
            e.provenance.label(),
        ));
    }
    html.push_str("</ul>");
    html
}

/// M4a (Codex C3 + M1): render the usage aggregate as a REUSABLE fragment. It emits
/// `class="usage"` ONLY — NO hardcoded `id="usage"` — so it can mount in BOTH the
/// TopBar and the right-rail Context budget without a duplicate id (invalid HTML;
/// `getElementById` would only hit the first). Each placement wraps it in a stable
/// `<div class="usage-mount" data-usage>` and the `usage` SSE handler sets that
/// wrapper's `innerHTML` to this fragment via `querySelectorAll('[data-usage]')`, so a
/// single `usage` event live-updates every mount. Cost renders ONLY from real
/// cost_cents; None -> `—`, NEVER $0.00.
pub fn render_usage_html(agg: &crate::read_model::UsageAggregate) -> String {
    fn cents(c: Option<u64>) -> String {
        match c {
            Some(v) => format!("${}.{:02}", v / 100, v % 100),
            None => "—".to_string(),
        }
    }
    let mut html = format!(
        "<div class=\"usage\"><span class=\"usage-tokens\">{} tokens</span><span class=\"usage-cost\">{}</span>",
        agg.total_tokens,
        cents(agg.total_cost_cents)
    );
    for row in &agg.per_agent {
        html.push_str(&format!(
            "<span class=\"usage-agent\">{}: {} ({})</span>",
            crate::db_dashboard::escape_html(&row.agent),
            row.tokens,
            cents(row.cost_cents)
        ));
    }
    html.push_str("</div>");
    html
}

/// Content-sensitive diff key INCLUDING the aggregate values, so a usage change streams
/// to the top-bar/budget even without a new feed row.
pub fn usage_diff_key(agg: &crate::read_model::UsageAggregate) -> String {
    let mut key = format!("{}|{:?}", agg.total_tokens, agg.total_cost_cents);
    for row in &agg.per_agent {
        key.push_str(&format!(
            "|{}:{}:{:?}",
            row.agent, row.tokens, row.cost_cents
        ));
    }
    key
}

/// ADR 032 D5: prefer a live herdr source (here: `herdr pane list`, the CLI wrapper
/// — automation starts with the CLI per ADR 031 D7) when HERDR_ENV=1; otherwise
/// fall back to the DB participants. Never fails the stream — on any error it
/// degrades to db-fallback. `known_targets`-style participants are passed in by the
/// caller so this stays free of a DB connection.
pub fn load_roster(
    herdr_bin: &str,
    db_participants: Vec<(String, String, String)>,
) -> Vec<RosterEntry> {
    if std::env::var("HERDR_ENV").as_deref() == Ok("1") {
        if let Some(live) = live_herdr_roster(herdr_bin, &db_participants) {
            return live;
        }
    }
    roster_from_db(db_participants)
}

fn live_herdr_roster(
    herdr_bin: &str,
    db_participants: &[(String, String, String)],
) -> Option<Vec<RosterEntry>> {
    let output = std::process::Command::new(herdr_bin)
        .args(["pane", "list"])
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    let text = String::from_utf8_lossy(&output.stdout);
    // `herdr pane list` emits a single-line JSON blob whose `result.panes` array
    // holds one object per pane (`pane_id` + `agent_status`). We scan that text for
    // the per-pane object that carries each known participant's address (pane id)
    // and read its `agent_status`. This is a tolerant, dependency-free scan: any
    // structural surprise (missing field, no match) yields `None` for that pane —
    // when herdr is up but the pane isn't found, we keep the participant's DB
    // status with provenance `Unknown` (R1 P3) instead of erasing it to a literal
    // "unknown". The whole call never panics — an exec/parse failure returns `None`
    // so `load_roster` uses db-fallback.
    let mut roster = Vec::new();
    for (agent, address, db_status) in db_participants {
        let status = herdr_status_for_pane(&text, address);
        roster.push(RosterEntry {
            agent: agent.clone(),
            address: address.clone(),
            status: status.clone().unwrap_or_else(|| db_status.clone()),
            provenance: if status.is_some() {
                Provenance::LiveHerdr
            } else {
                Provenance::Unknown
            },
        });
    }
    Some(roster)
}

/// Read `agent_status` for the pane whose `pane_id` equals `pane`, scanning the
/// single-line `herdr pane list` JSON. We isolate the per-pane object by finding
/// the exact `"pane_id":"<pane>"` token, then read the `agent_status` value within
/// that object's `{ ... }` bounds. Returns `None` on any miss (no such pane, no
/// status, or an empty/unparseable value) so the caller marks the pane `unknown`.
///
/// Assumes herdr's documented FLAT pane object (`pane_id` and `agent_status` as
/// sibling string fields). A nested `{...}` between them would close the object
/// scope early and read as `unknown` — degrades safely, never panics.
fn herdr_status_for_pane(list_text: &str, pane: &str) -> Option<String> {
    let needle = format!("\"pane_id\":\"{pane}\"");
    let id_at = list_text.find(&needle)?;
    // Bound the search to this pane's JSON object: back to the opening `{`, forward
    // to the closing `}`. Both fields live in the same object.
    let obj_start = list_text[..id_at].rfind('{').unwrap_or(0);
    let obj_end = list_text[id_at..]
        .find('}')
        .map(|rel| id_at + rel)
        .unwrap_or(list_text.len());
    let object = &list_text[obj_start..obj_end];
    let status = json_string_field(object, "agent_status")?;
    if status.is_empty() {
        None
    } else {
        Some(status)
    }
}

/// Minimal tolerant extractor for `"<field>":"<value>"` from a JSON fragment.
/// No escape handling beyond stopping at the next `"`; herdr status values are
/// bare ASCII tokens (idle/working/blocked/done/unknown), so this is sufficient
/// and degrades to `None` on anything unexpected.
fn json_string_field(fragment: &str, field: &str) -> Option<String> {
    let key = format!("\"{field}\":\"");
    let start = fragment.find(&key)? + key.len();
    let rest = &fragment[start..];
    let end = rest.find('"')?;
    Some(rest[..end].to_string())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A baseline message-kind `FeedEvent` for keying tests. Proof overlay starts
    /// at `delivery_status=sent`, self-verified by the helper tool.
    fn sample_event() -> FeedEvent {
        FeedEvent {
            event_key: "message:m1".to_string(),
            source_table: "messages".to_string(),
            source_id: "m1".to_string(),
            session_id: "s1".to_string(),
            timestamp: "2026-05-30T00:00:00Z".to_string(),
            kind: "message".to_string(),
            subtype: Some("status-update".to_string()),
            mid: Some("m1".to_string()),
            actor_agent_id: Some("claude".to_string()),
            target_agent_id: Some("codex".to_string()),
            source_address: Some("w1-2".to_string()),
            target_address: Some("w1-1".to_string()),
            transport: Some("herdr".to_string()),
            workspace_id: Some("w1".to_string()),
            mode: None,
            r#ref: None,
            re: None,
            summary: None,
            body: Some("live ping".to_string()),
            redaction_policy: Some("full".to_string()),
            proof_audit_id: Some("a1".to_string()),
            delivery_status: Some("sent".to_string()),
            verified_by: Some("helper-tool".to_string()),
            payload_hash: Some("deadbeef".to_string()),
            artifact_path: None,
            severity: None,
            is_derived: false,
            work: None,
            decision: None,
            revealable: false,
        }
    }

    #[test]
    fn feed_key_changes_on_proof_overlay_edit() {
        // An in-place proof transition (delivery_status sent -> observed) keeps the
        // same identity (event_key/mid) but MUST change feed_key, so diff_feed sees
        // a non-prefix and resets — pushing the updated proof strip (ADR 032 D4).
        let before = sample_event();
        let mut after = before.clone();
        after.delivery_status = Some("observed".to_string());
        assert_ne!(
            feed_key(&before),
            feed_key(&after),
            "an in-place proof overlay edit must alter feed_key (else the live feed goes stale)"
        );
        // And a same-second sibling with a distinct event_key is still distinguished.
        let mut sibling = before.clone();
        sibling.event_key = "message:m2".to_string();
        sibling.mid = Some("m2".to_string());
        assert_ne!(feed_key(&before), feed_key(&sibling));
    }

    /// A gate `work_events` FeedEvent (source_id "1") for overlay-keying tests — no
    /// decision overlay of its own; the overlay rides in the separate map.
    fn sample_gate_work_event() -> FeedEvent {
        FeedEvent {
            event_key: "work:1".to_string(),
            source_table: "work_events".to_string(),
            source_id: "1".to_string(),
            kind: "gate".to_string(),
            is_derived: true,
            ..FeedEvent::empty()
        }
    }

    /// A `DecisionView` overlay (gate verdict by default) for keying tests.
    fn sample_overlay() -> DecisionView {
        DecisionView {
            audit_id: "od1".to_string(),
            decision_type: "gate-decision".to_string(),
            target_work_event_id: Some(1),
            verdict: Some("approve".to_string()),
            resolution: None,
            mode_to: None,
            target_agent: None,
            reason: None,
            note: None,
            actor_agent_id: Some("operator".to_string()),
            timestamp: "2026-05-31T00:00:00Z".to_string(),
            notification_status: "not-requested".to_string(),
            revealable: false,
        }
    }

    #[test]
    fn feed_diff_key_flips_when_gate_overlay_appears_or_changes() {
        // M2b R1 P2: a gate/conflict DECISION is an overlay (separate map), not a feed
        // row. The combined diff key MUST fold it in, so a not-decided -> decided
        // transition (and a later notify-status change on the same overlay) flips the
        // key -> diff_feed resets -> the open dashboard gets the verdict overlay live.
        let event = sample_gate_work_event();
        let empty: BTreeMap<i64, DecisionView> = BTreeMap::new();
        let undecided = feed_diff_key(&event, &empty);
        // An undecided gate keys exactly like its bare feed_key (no overlay fold).
        assert_eq!(undecided, feed_key(&event));

        // Decided: the verdict overlay appears -> the key MUST change.
        let mut decided_map = BTreeMap::new();
        decided_map.insert(1_i64, sample_overlay());
        let decided = feed_diff_key(&event, &decided_map);
        assert_ne!(
            undecided, decided,
            "a gate's not-decided -> decided overlay transition must flip the diff key"
        );

        // A later notify-status change on the SAME overlay also flips the key (so a
        // not-requested -> sent/failed notify transition pushes a fresh reset).
        let mut notified_map = BTreeMap::new();
        let mut notified = sample_overlay();
        notified.notification_status = "sent".to_string();
        notified_map.insert(1_i64, notified);
        let notified_key = feed_diff_key(&event, &notified_map);
        assert_ne!(
            decided, notified_key,
            "an overlay notify-status change (not-requested -> sent) must flip the diff key"
        );

        // P2 residual: a RE-DECISION keeps the same work_event id (latest-wins in the
        // map) but is a NEW operator_decisions row (NEW audit_id). The render shows the
        // overlay's actor/timestamp/NOTE too, so a re-decision with the SAME verdict
        // but a different note (and a new audit_id) MUST flip the key — else only a
        // heartbeat fires and the new note never reaches the open dashboard. Folding the
        // overlay's audit_id into the digest makes ANY decision change flip the key.
        let mut redecided_map = BTreeMap::new();
        let mut redecided = sample_overlay();
        redecided.audit_id = "od2".to_string(); // a re-decision is a fresh row
        redecided.verdict = Some("approve".to_string()); // SAME verdict as `decided`
        redecided.note = Some("now with a note".to_string()); // only the note changed
        redecided_map.insert(1_i64, redecided);
        assert_ne!(
            decided,
            feed_diff_key(&event, &redecided_map),
            "a same-verdict re-decision (new audit_id / new note) must flip the diff key"
        );

        // An overlay for a DIFFERENT work-event id does not touch this row's key.
        let mut other_map = BTreeMap::new();
        other_map.insert(2_i64, sample_overlay());
        assert_eq!(
            undecided,
            feed_diff_key(&event, &other_map),
            "an overlay keyed to another work_event id must not change this row's key"
        );
    }

    #[test]
    fn feed_diff_key_flips_on_standalone_decision_notification_change() {
        // A STANDALONE decision feed item (mode/interrupt/redirect) carries its own
        // `decision`. Its notification_status is in-place mutable; a not-requested ->
        // sent/failed transition must flip the combined key so the live card updates.
        let empty: BTreeMap<i64, DecisionView> = BTreeMap::new();
        let mut decision = sample_overlay();
        decision.decision_type = "redirect".to_string();
        decision.target_work_event_id = None;
        decision.verdict = None;
        decision.target_agent = Some("codex:w1-1".to_string());
        let mut event = FeedEvent {
            event_key: "decision:od1".to_string(),
            source_table: "operator_decisions".to_string(),
            source_id: "od1".to_string(),
            kind: "redirect".to_string(),
            is_derived: true,
            decision: Some(decision),
            ..FeedEvent::empty()
        };
        let before = feed_diff_key(&event, &empty);
        // Transition the standalone decision's notify status in place.
        if let Some(d) = event.decision.as_mut() {
            d.notification_status = "sent".to_string();
        }
        let after = feed_diff_key(&event, &empty);
        assert_ne!(
            before, after,
            "a standalone decision's notify-status transition must flip the diff key"
        );
    }

    #[test]
    fn sse_event_frames_event_and_multiline_data() {
        let mut buf = Vec::new();
        sse_event(
            &mut buf,
            "feed-reset",
            "<article>a</article>\n<article>b</article>",
        );
        let text = String::from_utf8(buf).unwrap();
        assert_eq!(
            text,
            "event: feed-reset\ndata: <article>a</article>\ndata: <article>b</article>\n\n"
        );
    }

    #[test]
    fn sse_event_splits_bare_cr() {
        // A bare inline `\r` (which escape_html does not touch) must become its own
        // `data:` line, never survive inside one — else EventSource splits it and
        // drops the tail. CRLF collapses to a single break (no empty `data:` line).
        let mut buf = Vec::new();
        sse_event(&mut buf, "feed-reset", "a\rb");
        let text = String::from_utf8(buf).unwrap();
        assert_eq!(text, "event: feed-reset\ndata: a\ndata: b\n\n");
        assert!(!text.contains('\r'), "no bare CR survives in any data line");

        let mut crlf = Vec::new();
        sse_event(&mut crlf, "feed-reset", "a\r\nb");
        assert_eq!(
            String::from_utf8(crlf).unwrap(),
            "event: feed-reset\ndata: a\ndata: b\n\n",
            "CRLF is one break, not an empty data line",
        );
    }

    #[test]
    fn roster_falls_back_to_db_when_no_herdr() {
        // db-fallback roster from a list of (agent, address, db_status) participants
        // (R1 P3: the third field carries the DB roster status, so a status-only
        // session's lead agent keeps its projected status instead of a hardcoded
        // "unknown").
        let participants = vec![
            (
                "claude".to_string(),
                "w1-2".to_string(),
                "working".to_string(),
            ),
            (
                "codex".to_string(),
                "w1-1".to_string(),
                "unknown".to_string(),
            ),
        ];
        let roster = roster_from_db(participants);
        assert_eq!(roster.len(), 2);
        assert!(roster
            .iter()
            .all(|e| e.provenance == Provenance::DbFallback));
        // The db_status is honored, not overwritten with a literal "unknown".
        assert_eq!(
            roster.iter().find(|e| e.agent == "claude").unwrap().status,
            "working"
        );
        let html = render_roster_html(&roster);
        assert!(html.contains("codex") && html.contains("db-fallback"));
        assert!(!html.contains("<script"), "escaped");
    }

    #[test]
    fn herdr_status_attributes_each_panes_own_status() {
        // The whole point of the JSON deviation: a single-line blob must yield each
        // pane's OWN agent_status, not the FIRST status in the line.
        let blob = r#"{"result":{"panes":[{"pane_id":"w1-1","agent_status":"working"},{"pane_id":"w1-2","agent_status":"idle"}]}}"#;
        assert_eq!(
            herdr_status_for_pane(blob, "w1-1").as_deref(),
            Some("working"),
            "first pane keeps its own status",
        );
        assert_eq!(
            herdr_status_for_pane(blob, "w1-2").as_deref(),
            Some("idle"),
            "second pane is NOT mis-attributed the first pane's status",
        );
    }

    #[test]
    fn herdr_status_does_not_collide_on_pane_id_prefix() {
        // `1-1` must not match `1-11`. The needle carries the closing quote, so the
        // exact-id lookup skips past the prefix-sharing sibling. Order `1-11` first
        // to prove we don't stop at the prefix.
        let blob = r#"{"panes":[{"pane_id":"1-11","agent_status":"blocked"},{"pane_id":"1-1","agent_status":"done"}]}"#;
        assert_eq!(
            herdr_status_for_pane(blob, "1-1").as_deref(),
            Some("done"),
            "1-1 resolves to its own status, not 1-11's",
        );
        assert_eq!(
            herdr_status_for_pane(blob, "1-11").as_deref(),
            Some("blocked")
        );
    }

    #[test]
    fn herdr_status_missing_agent_status_is_none() {
        // A pane object with a pane_id but no agent_status -> None (caller: unknown).
        let blob = r#"{"panes":[{"pane_id":"w1-1","cwd":"/x","focused":false}]}"#;
        assert_eq!(herdr_status_for_pane(blob, "w1-1"), None);
    }

    #[test]
    fn herdr_status_non_json_or_empty_is_none() {
        // Tolerant: any non-matching / non-JSON / empty input degrades to None so
        // load_roster falls back to db-fallback (never fails the stream).
        assert_eq!(herdr_status_for_pane("not json at all", "w1-1"), None);
        assert_eq!(herdr_status_for_pane("", "w1-1"), None);
        // pane id absent from an otherwise valid blob.
        let blob = r#"{"panes":[{"pane_id":"w1-2","agent_status":"idle"}]}"#;
        assert_eq!(herdr_status_for_pane(blob, "w1-1"), None);
    }

    #[test]
    fn diff_feed_appends_pure_suffix_else_resets() {
        let a = vec!["k1".to_string(), "k2".to_string()];
        let ab = vec!["k1".to_string(), "k2".to_string(), "k3".to_string()];
        // pure suffix-extension -> append the tail
        assert_eq!(diff_feed(&a, &ab), FeedDelta::Append(2));
        // reorder / not a prefix -> reset
        let reordered = vec!["k2".to_string(), "k1".to_string(), "k3".to_string()];
        assert_eq!(diff_feed(&a, &reordered), FeedDelta::Reset);
        // shrink -> reset
        assert_eq!(diff_feed(&ab, &a), FeedDelta::Reset);
        // unchanged -> append zero (no-op)
        assert_eq!(diff_feed(&a, &a), FeedDelta::Append(2));
    }

    // M4a U2 (Codex C3): the usage strip renders cost ONLY from real cost_cents.
    // A `None` aggregate cost shows tokens-only + cost `—`, NEVER a fabricated
    // `$0.00`; a present cost renders as dollars from the integer cents.
    #[test]
    fn usage_html_never_renders_fake_zero_cost() {
        use crate::read_model::{UsageAgentRow, UsageAggregate};
        let none_cost = UsageAggregate {
            total_tokens: 350,
            total_cost_cents: None,
            per_agent: vec![UsageAgentRow {
                agent: "claude".into(),
                tokens: 350,
                cost_cents: None,
            }],
        };
        let html = render_usage_html(&none_cost);
        assert!(html.contains("350"), "tokens are shown: {html}");
        assert!(
            !html.contains("$0.00"),
            "an absent cost must NOT fabricate $0.00: {html}"
        );
        assert!(
            html.contains('\u{2014}'),
            "an absent cost renders as the em dash placeholder: {html}"
        );
        let with_cost = UsageAggregate {
            total_tokens: 350,
            total_cost_cents: Some(13),
            per_agent: vec![UsageAgentRow {
                agent: "claude".into(),
                tokens: 350,
                cost_cents: Some(13),
            }],
        };
        let html = render_usage_html(&with_cost);
        assert!(
            html.contains("$0.13"),
            "a present cost renders dollars from integer cents: {html}"
        );
    }

    // M4a U2 (Codex C3): the usage diff key INCLUDES the aggregate values so a usage
    // change streams to the top-bar/budget even without a new feed row. The key must
    // differ when ONLY total_tokens changes, when ONLY total_cost_cents changes, and
    // when ONLY a per_agent token count changes.
    #[test]
    fn usage_diff_key_changes_on_aggregate_change() {
        use crate::read_model::{UsageAgentRow, UsageAggregate};
        let base = UsageAggregate {
            total_tokens: 100,
            total_cost_cents: Some(5),
            per_agent: vec![UsageAgentRow {
                agent: "claude".into(),
                tokens: 100,
                cost_cents: Some(5),
            }],
        };
        let mut more_tokens = base.clone();
        more_tokens.total_tokens = 200;
        assert_ne!(
            usage_diff_key(&base),
            usage_diff_key(&more_tokens),
            "a total_tokens change must flip the key"
        );
        let mut more_cost = base.clone();
        more_cost.total_cost_cents = Some(8);
        assert_ne!(
            usage_diff_key(&base),
            usage_diff_key(&more_cost),
            "a total_cost_cents change must flip the key"
        );
        let mut more_agent_tokens = base.clone();
        more_agent_tokens.per_agent[0].tokens = 150;
        assert_ne!(
            usage_diff_key(&base),
            usage_diff_key(&more_agent_tokens),
            "a per_agent token change must flip the key"
        );
    }
}