zynk 0.8.0

use crate::db_dashboard::escape_html;
use crate::read_model::FeedEvent;
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(""),
    )
}

/// 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
}

/// 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,
        }
    }

    #[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));
    }

    #[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));
    }
}