zynk 0.8.0

use crate::read_model::{feed_oldest_first, permalink, verify_chain, FeedEvent};
use crate::{CliError, CliResult};
use clap::Args;
use rusqlite::Connection;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

#[derive(Debug, Args)]
pub struct DbServeArgs {
    #[arg(long, default_value = "127.0.0.1")]
    pub host: String,
    #[arg(long, default_value_t = 8787)]
    pub port: u16,
    #[arg(long, help = "serve one request, then exit")]
    pub once: bool,
    #[arg(
        long,
        help = "import outputs/ artifacts before each render (opt-in; OFF by default — does not change the ADR 025 DB-read-only default)"
    )]
    pub auto_import: bool,
    #[arg(
        long,
        default_value = "outputs",
        help = "runtime outputs root: used for --auto-import imports and, when --allow-writes is set, as the artifact root the browser composer send writes under (ADR 031 D3)"
    )]
    pub root: PathBuf,
    #[arg(
        long,
        help = "enable browser-originated writes (ADR 031); OFF by default — the dashboard is read-only unless set."
    )]
    pub allow_writes: bool,
    #[arg(
        long,
        default_value = "herdr",
        help = "herdr executable the composer send shells out to (ADR 031)."
    )]
    pub herdr_bin: String,
}

/// ADR 031: per-serve write configuration. Present only when `--allow-writes` is
/// set; carries the CSRF token, the bound authority (`127.0.0.1:<port>`), and the
/// herdr binary the composer send shells out to.
struct WriteConfig {
    token: String,
    authority: String,
    herdr_bin: String,
}

/// Per-serve, thread-shared connection config. Carries the served authority (for
/// the exact-Host read guard, ADR 032 P1) so read routes can validate Host even
/// when writes are disabled.
struct ServeContext {
    db_path: PathBuf,
    root: PathBuf,
    auto_import: bool,
    authority: String,
    writes: Option<WriteConfig>,
    /// ADR 032 D3: count of currently-live SSE streams, shared across connection
    /// threads. `serve_sse` increments on entry and an RAII guard decrements on
    /// every exit path, so the cap bounds concurrent SSE resource use.
    active_sse: Arc<std::sync::atomic::AtomicUsize>,
}

struct DashboardSession {
    session_id: String,
    title: String,
    phase: String,
    mode: String,
    workflow_status: String,
    lead_agent_id: String,
    artifact_ref: String,
    updated_at: String,
    next_action: String,
    blockers: String,
    asks_for_zevs: String,
    risk_or_residual_uncertainty: String,
    expected_wait: String,
}

pub fn serve(path: &Path, args: DbServeArgs) -> CliResult<()> {
    if args.host != "127.0.0.1" {
        return Err(CliError::usage(
            "db dashboard server binds only to 127.0.0.1 in v0.2",
        ));
    }
    crate::db::open_database(path)?;
    let listener = TcpListener::bind((args.host.as_str(), args.port)).map_err(|error| {
        CliError::failure(format!(
            "failed to bind dashboard server on {}:{}: {error}",
            args.host, args.port
        ))
    })?;
    let address = listener.local_addr().map_err(|error| {
        CliError::failure(format!(
            "failed to read dashboard listener address: {error}"
        ))
    })?;
    println!("listening on http://{address}/");
    std::io::stdout()
        .flush()
        .map_err(|error| CliError::failure(format!("failed to flush dashboard URL: {error}")))?;

    // ADR 031: writes are OFF unless --allow-writes. The CSRF token is minted once
    // per serve and embedded in the rendered composer; the authority pins the exact
    // Host/Origin the write authorization requires.
    let authority = address.to_string();
    let writes = args.allow_writes.then(|| WriteConfig {
        token: crate::dashboard_write::mint_csrf_token(),
        authority: authority.clone(),
        herdr_bin: args.herdr_bin.clone(),
    });
    let context = Arc::new(ServeContext {
        db_path: path.to_path_buf(),
        root: args.root.clone(),
        auto_import: args.auto_import,
        authority,
        writes,
        active_sse: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
    });

    if args.once {
        let (stream, _) = listener.accept().map_err(|error| {
            CliError::failure(format!("failed to accept dashboard request: {error}"))
        })?;
        return handle_connection(stream, &context);
    }

    for stream in listener.incoming() {
        let stream = stream.map_err(|error| {
            CliError::failure(format!("failed to accept dashboard request: {error}"))
        })?;
        let context = Arc::clone(&context);
        std::thread::spawn(move || {
            if let Err(error) = handle_connection(stream, &context) {
                eprintln!("dashboard connection error: {}", error.message);
            }
        });
    }
    Ok(())
}

fn handle_connection(mut stream: TcpStream, ctx: &ServeContext) -> CliResult<()> {
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .map_err(|error| CliError::failure(format!("failed to set read timeout: {error}")))?;
    let (head, raw_body) = read_http_request(&mut stream)?;
    let request = crate::dashboard_write::parse_request(&head, raw_body);

    // ADR 032 P1: anti-DNS-rebind. Every request (reads, SSE, and the write POST)
    // must carry the exact served authority as Host; no-CORS alone does not stop a
    // same-origin read by a DNS-rebound page. (The write path also re-checks Host
    // in authorize_write; this guards the read surface too.)
    //
    // R1 P5: the header map keeps last-value-wins, so a request with TWO Host lines
    // is ambiguous; require EXACTLY one Host header that equals the served authority
    // (zero or 2+ Host lines fail closed).
    if request.host_count != 1 || request.header("host") != Some(ctx.authority.as_str()) {
        return write_response(
            &mut stream,
            "403 Forbidden",
            "text/plain; charset=utf-8",
            "host mismatch\n",
        );
    }

    // ADR 032 D3: GET /events is the SSE live stream. Its response is written
    // incrementally (no fixed Content-Length), so it cannot flow through the
    // `(status, content_type, body, location)` tuple below — handle it inline and
    // return. It runs AFTER the Task 2 Host guard above.
    if request.method == "GET" && request.route == "/events" {
        return serve_sse(
            &mut stream,
            ctx,
            query_param(&request.query, "session").as_deref(),
        );
    }

    // ADR 031 D6: POST /send is the only write surface. Validation + authorization
    // happen here, BEFORE any child spawns; GET/HEAD never dispatch a write.
    let (status, content_type, body, location) =
        if request.method == "POST" && request.route == "/send" {
            match ctx.writes.as_ref() {
                Some(config) => match crate::dashboard_write::authorize_write(
                    &request,
                    &config.authority,
                    &config.token,
                ) {
                    Ok(()) => {
                        match crate::dashboard_write::handle_send(
                            &request,
                            ctx.db_path.as_path(),
                            ctx.root.as_path(),
                            &config.herdr_bin,
                        ) {
                            crate::dashboard_write::WriteOutcome::Redirect(loc) => (
                                "303 See Other",
                                "text/plain; charset=utf-8",
                                String::new(),
                                Some(loc),
                            ),
                            crate::dashboard_write::WriteOutcome::Error { status, message } => (
                                status,
                                "text/html; charset=utf-8",
                                format!(
                                    "<!doctype html><meta charset=\"utf-8\"><pre>{}</pre>",
                                    escape_html(&message)
                                ),
                                None,
                            ),
                        }
                    }
                    Err(error) => (
                        "403 Forbidden",
                        "text/plain; charset=utf-8",
                        format!("{}\n", error.message),
                        None,
                    ),
                },
                None => (
                    "405 Method Not Allowed",
                    "text/plain; charset=utf-8",
                    "writes disabled\n".to_string(),
                    None,
                ),
            }
        } else if !matches!(request.method.as_str(), "GET" | "HEAD") {
            (
                "405 Method Not Allowed",
                "text/plain; charset=utf-8",
                "method not allowed\n".to_string(),
                None,
            )
        } else if matches!(request.route.as_str(), "/" | "/index.html") {
            // v0.2.2: when --auto-import is set, import file artifacts immediately
            // before rendering so the dashboard always reflects the latest writes
            // (no time-based staleness). Tied to the render path, so 404/405/asset
            // requests do not trigger imports. Import is idempotent (reused path).
            if ctx.auto_import {
                crate::db::import_outputs_root(ctx.db_path.as_path(), ctx.root.as_path())?;
            }
            let connection = crate::db::open_read_database(ctx.db_path.as_path())?;
            (
                "200 OK",
                "text/html; charset=utf-8",
                render_dashboard(
                    &connection,
                    query_param(&request.query, "session").as_deref(),
                    ctx.writes.as_ref().map(|config| config.token.as_str()),
                )?,
                None,
            )
        } else if matches!(request.route.as_str(), "/audit") {
            if ctx.auto_import {
                crate::db::import_outputs_root(ctx.db_path.as_path(), ctx.root.as_path())?;
            }
            let connection = crate::db::open_read_database(ctx.db_path.as_path())?;
            (
                "200 OK",
                "text/html; charset=utf-8",
                render_audit(
                    &connection,
                    query_param(&request.query, "session").as_deref(),
                )?,
                None,
            )
        } else {
            (
                "404 Not Found",
                "text/plain; charset=utf-8",
                "not found\n".to_string(),
                None,
            )
        };
    match location {
        Some(target) => redirect_response(&mut stream, &target),
        None => write_response(&mut stream, status, content_type, &body),
    }
}

/// ADR 032 D3: bound concurrent live SSE streams (loopback, single operator) so a
/// misbehaving client opening many streams cannot multiply the per-stream thread +
/// per-tick `herdr pane list` subprocess unboundedly.
const SSE_CONNECTION_CAP: usize = 8;

/// ADR 032 D4: the live feed is a "windowed (last-N), oldest-first" feed. Both the
/// SSE per-tick render and the static `#feed` render apply this bound so they agree
/// (a long design session does not stream/render thousands of rows every tick).
const FEED_WINDOW: usize = 200;

/// Keep the last `n` items of an oldest-first slice (the suffix); if the slice is
/// `<= n`, keep all. When the window slides (the feed grows past `n`), the front
/// changes, so `diff_feed` sees a non-prefix and returns `Reset` — the intended
/// reset-on-uncertainty (ADR 032 D4); we do not special-case it here.
fn windowed<T: Clone>(events: &[T], n: usize) -> Vec<T> {
    if events.len() > n {
        events[events.len() - n..].to_vec()
    } else {
        events.to_vec()
    }
}

fn serve_sse(stream: &mut TcpStream, ctx: &ServeContext, session: Option<&str>) -> CliResult<()> {
    // ADR 032 D3: count this stream as active; the RAII guard decrements on EVERY
    // exit path (the 503 below, a normal disconnect `return Ok(())`, or an error
    // `?`), so the counter never leaks. Placed before the SSE headers so the cap
    // is enforced before any work.
    struct SseGuard(std::sync::Arc<std::sync::atomic::AtomicUsize>);
    impl Drop for SseGuard {
        fn drop(&mut self) {
            self.0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
        }
    }
    let active = ctx
        .active_sse
        .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
        + 1;
    let _guard = SseGuard(ctx.active_sse.clone());
    if active > SSE_CONNECTION_CAP {
        return write_response(
            stream,
            "503 Service Unavailable",
            "text/plain; charset=utf-8",
            "too many live connections\n",
        );
    }

    // SSE headers — note: NO CORS headers (ADR 032 D7).
    let headers = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n";
    stream
        .write_all(headers.as_bytes())
        .map_err(|e| CliError::failure(format!("failed to write SSE headers: {e}")))?;

    use std::time::Duration;
    let mut last_keys: Vec<String> = Vec::new();
    let mut first = true;
    loop {
        // Short-lived read connection per tick; dropped before the sleep (no DB
        // handle held across the stream — ADR 032 D3).
        let (keys, reset_html, append_html_from, participants, session_state_html) = {
            let connection = crate::db::open_read_database(&ctx.db_path)?;
            // Resolve the selected session from the SAME session list the static page
            // renders, so the live `session-state` fragments match the page's chrome.
            let sessions = load_sessions(&connection)?;
            let selected = session
                .and_then(|id| sessions.iter().find(|s| s.session_id == id))
                .or_else(|| sessions.first());
            let selected_id = selected.map(|s| s.session_id.clone());
            // ADR 032 D4: window to the last-N oldest-first events. When the window
            // slides (feed grows past N), the front changes and diff_feed resets.
            let feed = match &selected_id {
                Some(id) => windowed(
                    &crate::read_model::feed_oldest_first(&connection, id)?,
                    FEED_WINDOW,
                ),
                None => Vec::new(),
            };
            // Roster participants computed in the SAME per-tick connection scope (no
            // DB handle held across the stream — ADR 032 D3); `load_roster` runs
            // AFTER the scope on the by-value tuples. R1 P3: source from the DB
            // roster state (lead_agent + agents/session_agents), NOT just the audit
            // participants, so a status-only session still has a non-empty roster.
            // (`known_targets`/`known_targets_pairs` stay the composer allow-list.)
            let participants = match &selected_id {
                Some(id) => roster_db_participants(&connection, id)?,
                None => Vec::new(),
            };
            // ADR 032 D4 (R1 P1): the current-state chrome (sidebar nav + timeline
            // header + detail panel), rendered with the SAME inner renderers the
            // static page uses. Joined by ASCII record separator (U+001E) — not a
            // CR/LF, so `sse_event`'s line handling preserves the delimiter; each
            // fragment is server-rendered + `escape_html`-escaped (trusted HTML).
            // R1 hardening: strip any literal U+001E from each fragment first so a
            // free-text field containing U+001E can't mis-split the payload client-side.
            let session_state_html = format!(
                "{}\u{1e}{}\u{1e}{}",
                render_session_nav_inner(&sessions, selected_id.as_deref()).replace('\u{1e}', ""),
                render_timeline_header_inner(selected).replace('\u{1e}', ""),
                render_detail_inner(selected).replace('\u{1e}', ""),
            );
            let keys: Vec<String> = feed.iter().map(crate::dashboard_live::feed_key).collect();
            let delta = if first {
                crate::dashboard_live::FeedDelta::Reset
            } else {
                crate::dashboard_live::diff_feed(&last_keys, &keys)
            };
            match delta {
                crate::dashboard_live::FeedDelta::Reset => (
                    keys,
                    Some(render_feed_html(&feed)),
                    None,
                    participants,
                    session_state_html,
                ),
                crate::dashboard_live::FeedDelta::Append(from) if from < feed.len() => (
                    keys,
                    None,
                    Some(render_feed_html(&feed[from..])),
                    participants,
                    session_state_html,
                ),
                crate::dashboard_live::FeedDelta::Append(_) => {
                    (keys, None, None, participants, session_state_html)
                } // no feed change
            }
        };
        let mut buf = Vec::new();
        if let Some(html) = reset_html {
            crate::dashboard_live::sse_event(&mut buf, "feed-reset", &html);
        } else if let Some(html) = append_html_from {
            crate::dashboard_live::sse_event(&mut buf, "feed-append", &html);
        } else {
            crate::dashboard_live::sse_event(&mut buf, "heartbeat", "");
        }
        // ADR 032 D4 (R1 P1): emit the current-state chrome each tick so a live
        // status change updates the sidebar badges / header / detail rail without a
        // reload (the feed alone would leave those panels stale).
        crate::dashboard_live::sse_event(&mut buf, "session-state", &session_state_html);
        // Roster (ADR 032 D5): live-herdr when HERDR_ENV=1, else db-fallback. The
        // DB connection is already dropped; `load_roster` works on the by-value
        // participants and never fails the stream (errors degrade to db-fallback).
        let herdr_bin = ctx
            .writes
            .as_ref()
            .map(|w| w.herdr_bin.as_str())
            .unwrap_or("herdr");
        let roster = crate::dashboard_live::load_roster(herdr_bin, participants);
        crate::dashboard_live::sse_event(
            &mut buf,
            "roster",
            &crate::dashboard_live::render_roster_html(&roster),
        );
        if stream.write_all(&buf).is_err() {
            return Ok(()); // client disconnected — end the stream thread cleanly
        }
        last_keys = keys;
        first = false;
        std::thread::sleep(Duration::from_millis(750));
    }
}

/// ADR 031 D6: Post/Redirect/Get — after a successful write, send a 303 so a
/// refresh re-renders (GET) instead of re-submitting the POST.
fn redirect_response(stream: &mut TcpStream, location: &str) -> CliResult<()> {
    let response = format!(
        "HTTP/1.1 303 See Other\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
    );
    stream
        .write_all(response.as_bytes())
        .map_err(|error| CliError::failure(format!("failed to write redirect: {error}")))
}

fn query_param(query: &str, name: &str) -> Option<String> {
    query.split('&').find_map(|part| {
        let (key, value) = part.split_once('=')?;
        (key == name).then(|| percent_decode(value))
    })
}

fn read_http_request(stream: &mut TcpStream) -> CliResult<(String, Vec<u8>)> {
    let mut buf = Vec::new();
    let mut chunk = [0_u8; 1024];
    // Read until the end-of-headers marker (or a sane cap).
    let header_end = loop {
        if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
            break pos + 4;
        }
        if buf.len() > 16_384 {
            break buf.len();
        }
        let count = stream.read(&mut chunk).map_err(|error| {
            CliError::failure(format!("failed to read dashboard request: {error}"))
        })?;
        if count == 0 {
            break buf.len();
        }
        buf.extend_from_slice(&chunk[..count]);
    };
    let head = String::from_utf8_lossy(&buf[..header_end.min(buf.len())]).to_string();
    let mut body = buf.get(header_end..).unwrap_or(&[]).to_vec();
    // Read the remaining Content-Length body bytes (POST), capped at 1 MiB.
    let content_length = head
        .split("\r\n")
        .filter_map(|line| line.split_once(':'))
        .find(|(key, _)| key.trim().eq_ignore_ascii_case("content-length"))
        .and_then(|(_, value)| value.trim().parse::<usize>().ok())
        .unwrap_or(0)
        .min(1_048_576);
    while body.len() < content_length {
        let count = stream.read(&mut chunk).map_err(|error| {
            CliError::failure(format!("failed to read dashboard body: {error}"))
        })?;
        if count == 0 {
            break;
        }
        body.extend_from_slice(&chunk[..count]);
    }
    body.truncate(content_length);
    Ok((head, body))
}

fn write_response(
    stream: &mut TcpStream,
    status: &str,
    content_type: &str,
    body: &str,
) -> CliResult<()> {
    let response = format!(
        "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    );
    stream
        .write_all(response.as_bytes())
        .map_err(|error| CliError::failure(format!("failed to write dashboard response: {error}")))
}

/// ADR 031 D1: the known `agent:address` targets for a session — the distinct
/// participants (source + target) in its `audit_records`. The composer offers
/// only these, and the write path enforces the same allow-list before spawning
/// (so a browser write can never free-type an arbitrary herdr target).
pub(crate) fn known_targets(connection: &Connection, session_id: &str) -> CliResult<Vec<String>> {
    let mut statement = connection
        .prepare(
            "SELECT DISTINCT agent_id || ':' || address AS target FROM (
                 SELECT target_agent_id AS agent_id, target_address AS address
                   FROM audit_records WHERE session_id = ?1
                 UNION
                 SELECT source_agent_id AS agent_id, source_address AS address
                   FROM audit_records WHERE session_id = ?1
             )
             WHERE agent_id IS NOT NULL AND address IS NOT NULL
             ORDER BY target",
        )
        .map_err(|error| CliError::failure(format!("failed to prepare known_targets: {error}")))?;
    let rows = statement
        .query_map([session_id], |row| row.get::<_, String>(0))
        .map_err(|error| CliError::failure(format!("failed to query known_targets: {error}")))?;
    let mut targets = Vec::new();
    for row in rows {
        targets.push(
            row.map_err(|error| CliError::failure(format!("failed to read target: {error}")))?,
        );
    }
    Ok(targets)
}

/// The `(agent, address)` form of `known_targets` for a session — used to seed the
/// live roster (ADR 032 D5) with the session participants without holding a DB
/// handle across the SSE stream.
pub(crate) fn known_targets_pairs(
    connection: &Connection,
    session_id: &str,
) -> CliResult<Vec<(String, String)>> {
    Ok(known_targets(connection, session_id)?
        .into_iter()
        .filter_map(|t| {
            t.split_once(':')
                .map(|(a, b)| (a.to_string(), b.to_string()))
        })
        .collect())
}

/// ADR 032 D5 (R1 P3): the db-fallback roster participants — `(agent, address,
/// db_status)` gathered from the DB roster state, NOT only the audit participants.
/// A STATUS-ONLY session (a projected `lead_agent_id` + an `agents` row, no audit
/// rows) would otherwise yield an empty roster. We UNION:
///   (a) the session's `lead_agent_id` (address from `agents.current_address`,
///       db_status from `agents.current_agent_status`), skipped only when the lead
///       agent is NULL/`unknown`;
///   (b) the audit participants (`known_targets` agent:address; db_status joined
///       from `agents.current_agent_status`);
///   (c) `session_agents JOIN agents` (forward-compat: `session_agents` currently
///       has NO producer, so this contributes nothing today; queried for the day
///       it does, with db_status from `session_agents.agent_status`).
/// Deduped by `agent`, preferring an entry that carries a non-empty address.
pub(crate) fn roster_db_participants(
    connection: &Connection,
    session_id: &str,
) -> CliResult<Vec<(String, String, String)>> {
    let mut rows: Vec<(String, String, String)> = Vec::new();

    // (a) lead agent of the session.
    let mut lead_stmt = connection
        .prepare(
            "SELECT s.lead_agent_id,
                    COALESCE(a.current_address, ''),
                    COALESCE(a.current_agent_status, 'unknown')
             FROM sessions AS s
             LEFT JOIN agents AS a ON a.agent_id = s.lead_agent_id
             WHERE s.session_id = ?1
               AND s.lead_agent_id IS NOT NULL
               AND s.lead_agent_id <> 'unknown'",
        )
        .map_err(|e| CliError::failure(format!("failed to prepare roster lead query: {e}")))?;
    let lead_rows = lead_stmt
        .query_map([session_id], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
            ))
        })
        .map_err(|e| CliError::failure(format!("failed to query roster lead: {e}")))?;
    for row in lead_rows {
        rows.push(row.map_err(|e| CliError::failure(format!("failed to read roster lead: {e}")))?);
    }

    // (b) audit participants (same source as the composer allow-list), with
    // db_status overlaid from the agents row when present.
    for (agent, address) in known_targets_pairs(connection, session_id)? {
        let db_status: String = connection
            .query_row(
                "SELECT COALESCE(current_agent_status, 'unknown') FROM agents WHERE agent_id = ?1",
                [agent.as_str()],
                |row| row.get(0),
            )
            .unwrap_or_else(|_| "unknown".to_string());
        rows.push((agent, address, db_status));
    }

    // (c) session_agents rows (forward-compat; no producer today).
    let mut sa_stmt = connection
        .prepare(
            "SELECT sa.agent_id,
                    COALESCE(a.current_address, ''),
                    sa.agent_status
             FROM session_agents AS sa
             LEFT JOIN agents AS a ON a.agent_id = sa.agent_id
             WHERE sa.session_id = ?1",
        )
        .map_err(|e| {
            CliError::failure(format!(
                "failed to prepare roster session_agents query: {e}"
            ))
        })?;
    let sa_rows = sa_stmt
        .query_map([session_id], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
            ))
        })
        .map_err(|e| CliError::failure(format!("failed to query roster session_agents: {e}")))?;
    for row in sa_rows {
        rows.push(row.map_err(|e| {
            CliError::failure(format!("failed to read roster session_agents: {e}"))
        })?);
    }

    // Dedup by agent, preferring an entry with a non-empty address (so the audit
    // participant's transport address survives over a lead-only blank).
    let mut deduped: Vec<(String, String, String)> = Vec::new();
    for (agent, address, db_status) in rows {
        if let Some(existing) = deduped.iter_mut().find(|(a, _, _)| *a == agent) {
            if existing.1.is_empty() && !address.is_empty() {
                existing.1 = address;
                existing.2 = db_status;
            }
        } else {
            deduped.push((agent, address, db_status));
        }
    }
    deduped.sort_by(|left, right| left.0.cmp(&right.0));
    Ok(deduped)
}

/// ADR 031 D1: whether a session row exists in the served DB. Browser writes
/// target an existing selected session; the server must not trust the posted id
/// (a trusted id would let the audited-send projection create a new session).
pub(crate) fn session_exists(connection: &Connection, session_id: &str) -> CliResult<bool> {
    let count: i64 = connection
        .query_row(
            "SELECT COUNT(*) FROM sessions WHERE session_id = ?1",
            [session_id],
            |row| row.get(0),
        )
        .map_err(|error| CliError::failure(format!("failed to check session: {error}")))?;
    Ok(count > 0)
}

fn render_dashboard(
    connection: &Connection,
    selected_session_id: Option<&str>,
    csrf_token: Option<&str>,
) -> CliResult<String> {
    let sessions = load_sessions(connection)?;
    let selected = selected_session_id
        .and_then(|session_id| {
            sessions
                .iter()
                .find(|session| session.session_id == session_id)
        })
        .or_else(|| sessions.first());
    let selected_id = selected.map(|session| session.session_id.as_str());
    // v0.8 T6: render the initial timeline OLDEST-first to match the SSE
    // `feed-reset` payload (`feed_oldest_first`); otherwise the chat feed visibly
    // flips order ~750ms after load when the first SSE tick lands. ADR 032 D4:
    // window to the last-N so the initial page and the stream agree.
    let feed = match selected_id {
        Some(id) => windowed(&feed_oldest_first(connection, id)?, FEED_WINDOW),
        None => Vec::new(),
    };
    let mut html = String::new();
    html.push_str("<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\">");
    html.push_str("<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\">");
    html.push_str("<title>zynk dashboard</title><style>");
    html.push_str(STYLES);
    html.push_str("</style></head><body>");
    html.push_str("<div class=\"app-shell\">");
    html.push_str(
        "<aside class=\"sidebar\"><div class=\"brand\">zynk</div><nav id=\"session-nav\">",
    );
    html.push_str(&render_session_nav_inner(&sessions, selected_id));
    html.push_str("</nav>");
    html.push_str("<section id=\"roster\" class=\"roster-panel\"><h2>Participants</h2><div class=\"roster-mount\"></div></section>");
    html.push_str("</aside>");

    html.push_str(
        "<main class=\"timeline\"><header id=\"timeline-header\" class=\"timeline-header\">",
    );
    html.push_str(&render_timeline_header_inner(selected));
    html.push_str("</header>");
    // ADR 031 D1: the operator composer — rendered only when writes are enabled
    // (a per-serve token is present) and a session is selected. The token rides in
    // a hidden field for the JS to lift into the required X-Zynk-CSRF header; the
    // server authorizes on the header, never the body field.
    if let (Some(token), Some(session)) = (csrf_token, selected) {
        // ADR 031 D1: the target is chosen from known agent/address rows, never
        // free-typed. The same allow-list is re-enforced server-side before spawn.
        let targets = known_targets(connection, &session.session_id)?;
        let mut options = String::new();
        for target in &targets {
            options.push_str(&format!(
                "<option value=\"{}\">{}</option>",
                escape_html(target),
                escape_html(target),
            ));
        }
        html.push_str(&format!(
            "<form class=\"composer\" method=\"post\" action=\"/send\">\
             <input type=\"hidden\" name=\"csrf\" value=\"{}\">\
             <input type=\"hidden\" name=\"session\" value=\"{}\">\
             <select name=\"to\" required>{}</select>\
             <input name=\"type\" value=\"status-update\">\
             <input name=\"body\" placeholder=\"message\u{2026}\" required>\
             <button>Send</button></form>",
            escape_html(token),
            escape_html(&session.session_id),
            options,
        ));
        if targets.is_empty() {
            html.push_str("<p class=\"composer-note\">No known targets for this session yet.</p>");
        }
        html.push_str(
            "<script>document.querySelector('.composer').addEventListener('submit',function(e){e.preventDefault();var f=e.target;fetch('/send',{method:'POST',headers:{'X-Zynk-CSRF':f.csrf.value,'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams(new FormData(f))}).then(function(r){if(r.redirected){location=r.url}else{r.text().then(function(t){document.body.innerHTML=t})}});});</script>",
        );
    }
    // v0.8 T6 (concern #1 fix): `id="feed"` wraps ONLY the feed articles, not the
    // whole `<main>`. The SSE `feed-reset`/`feed-append` payloads carry only the
    // article fragments, so `feed.innerHTML = …` must not destroy the timeline
    // `<header>` or the ADR 031 composer — they stay in `<main>`, OUTSIDE `#feed`.
    html.push_str("<div id=\"feed\">");
    if feed.is_empty() {
        html.push_str("<section class=\"empty-state\">No feed entries yet.</section>");
    } else {
        for event in &feed {
            render_feed_event(&mut html, event);
        }
    }
    html.push_str("</div></main>");

    html.push_str("<aside id=\"detail\" class=\"detail-panel\">");
    html.push_str(&render_detail_inner(selected));
    html.push_str("</aside></div>");
    // ADR 032 D7 / ADR 031 D4: a SEPARATE vanilla-JS SSE client (no build pipeline)
    // that live-updates the feed + roster. The feed/roster payloads are server-
    // rendered AND HTML-escaped (`render_feed_event` / `render_roster_html` via
    // `escape_html`), so `innerHTML`/`insertAdjacentHTML` here render TRUSTED server
    // output — the same trust model as the ADR 031 composer error branch above. No
    // client-side sanitization: the server escape is the load-bearing XSS control.
    //
    // Session ids are agent-created free-form strings, so they must NEVER be
    // interpolated into the <script> body: `{:?}` makes a valid JS string literal
    // but does NOT escape `</script>`, so a hostile id (e.g.
    // `</script><script>…</script>`) would break out and execute. Instead we emit
    // the id ONCE into an `escape_html`-escaped HTML data attribute (safe in
    // attribute context — `<` `>` `"` `'` `&` are escaped, no breakout) and the
    // STATIC script reads it via `dataset.sid`. The <script> body has zero dynamic
    // interpolation.
    if let Some(session) = selected {
        html.push_str(&format!(
            "<div id=\"sse-cfg\" data-sid=\"{}\" hidden></div>",
            escape_html(&session.session_id),
        ));
        html.push_str(
            "<script>(function(){var sid=document.getElementById('sse-cfg').dataset.sid;\
             var es=new EventSource('/events?session='+encodeURIComponent(sid));\
             var feed=document.getElementById('feed');var roster=document.querySelector('.roster-mount');\
             es.addEventListener('feed-reset',function(e){if(feed)feed.innerHTML=e.data;});\
             es.addEventListener('feed-append',function(e){if(feed)feed.insertAdjacentHTML('beforeend',e.data);});\
             es.addEventListener('roster',function(e){if(roster)roster.innerHTML=e.data;});\
             es.addEventListener('session-state',function(e){var p=e.data.split(String.fromCharCode(30));\
             var nav=document.getElementById('session-nav');if(nav&&p[0]!==undefined)nav.innerHTML=p[0];\
             var hdr=document.getElementById('timeline-header');if(hdr&&p[1]!==undefined)hdr.innerHTML=p[1];\
             var det=document.getElementById('detail');if(det&&p[2]!==undefined)det.innerHTML=p[2];});\
             })();</script>",
        );
    }
    html.push_str("</body></html>");
    Ok(html)
}

/// ADR 032 D4 (R1 P1): the inner HTML of the sidebar `<nav id="session-nav">`. The
/// static page and the live `session-state` SSE event share this one renderer so
/// the sidebar badges update on a live status change without a reload. `selected_id`
/// is accepted for parity with the other inner renderers (the current markup does
/// not highlight the selected link).
fn render_session_nav_inner(sessions: &[DashboardSession], _selected_id: Option<&str>) -> String {
    let mut html = String::new();
    if sessions.is_empty() {
        html.push_str("<p class=\"empty\">No sessions in this database.</p>");
    } else {
        for session in sessions {
            html.push_str(&format!(
                "<a class=\"session-link\" href=\"/?session={}\"><span>{}</span><span class=\"badge status-{}\">{}</span></a>",
                escape_url_component(&session.session_id),
                escape_html(&session.session_id),
                escape_class(&session.workflow_status),
                escape_html(&session.workflow_status),
            ));
        }
    }
    html
}

/// ADR 032 D4 (R1 P1): the inner HTML of `<header id="timeline-header">`. Shared by
/// the static page and the `session-state` SSE event so the header (session / phase
/// / mode) updates live.
fn render_timeline_header_inner(selected: Option<&DashboardSession>) -> String {
    let mut html = String::from("<div><h1>Timeline</h1>");
    if let Some(session) = selected {
        html.push_str(&format!(
            "<p>{} / {} / {}</p>",
            escape_html(&session.session_id),
            escape_html(&session.phase),
            escape_html(&session.mode)
        ));
    }
    html.push_str("</div>");
    html
}

/// ADR 032 D4 (R1 P1): the inner HTML of the right `<aside id="detail">` status
/// rail. Shared by the static page and the `session-state` SSE event so the detail
/// panel updates live on a status change.
fn render_detail_inner(selected: Option<&DashboardSession>) -> String {
    let mut html = String::from("<h2>Status</h2>");
    if let Some(session) = selected {
        html.push_str(&format!(
            "<dl><dt>Session</dt><dd>{}</dd><dt>State</dt><dd>{}</dd><dt>Next</dt><dd>{}</dd><dt>Ask</dt><dd>{}</dd><dt>Blockers</dt><dd>{}</dd><dt>Risk</dt><dd>{}</dd><dt>Expected wait</dt><dd>{}</dd><dt>Artifact</dt><dd>{}</dd><dt>Lead</dt><dd>{}</dd><dt>Updated</dt><dd>{}</dd></dl>",
            escape_html(&session.title),
            escape_html(&session.workflow_status),
            escape_html(&session.next_action),
            escape_html(&session.asks_for_zevs),
            escape_html(&session.blockers),
            escape_html(&session.risk_or_residual_uncertainty),
            escape_html(&session.expected_wait),
            escape_html(&session.artifact_ref),
            escape_html(&session.lead_agent_id),
            escape_html(&session.updated_at),
        ));
    }
    html
}

/// Render a feed (already in display order) to concatenated, HTML-escaped article
/// fragments — the SSE payload AND the static-page body share this.
fn render_feed_html(events: &[FeedEvent]) -> String {
    let mut html = String::new();
    for event in events {
        render_feed_event(&mut html, event);
    }
    html
}

/// ADR 030 D2/D6: render one read-model feed entry — a message shows its body
/// (or a redacted marker for hash-only), with the proof strip carrying the latest
/// delivery state, the transport addresses, and the stable permalink.
fn render_feed_event(html: &mut String, event: &FeedEvent) {
    html.push_str(&format!(
        "<article class=\"feed-item kind-{}\"><div class=\"timestamp\">{}</div>",
        escape_class(&event.kind),
        escape_html(&event.timestamp),
    ));
    let who = event.actor_agent_id.as_deref().unwrap_or("system");
    let label = event.subtype.as_deref().unwrap_or(event.kind.as_str());
    html.push_str(&format!(
        "<h2>{} <span class=\"kind\">{}</span>",
        escape_html(who),
        escape_html(label),
    ));
    if let Some(mid) = &event.mid {
        html.push_str(&format!(" <span class=\"mid\">{}</span>", escape_html(mid)));
    }
    html.push_str("</h2>");
    match &event.body {
        Some(body) => html.push_str(&format!("<p class=\"body\">{}</p>", escape_html(body))),
        None if event.kind == "message" => {
            html.push_str("<p class=\"redacted\">\u{2298} redacted \u{00b7} hash-only</p>")
        }
        None => {
            if let Some(summary) = &event.summary {
                html.push_str(&format!("<p>{}</p>", escape_html(summary)));
            }
        }
    }
    if event.proof_audit_id.is_some() {
        html.push_str("<div class=\"proof-strip\">");
        html.push_str(&format!(
            "<span class=\"proof proof-{}\">{} / {}</span>",
            escape_class(event.delivery_status.as_deref().unwrap_or("unknown")),
            escape_html(event.delivery_status.as_deref().unwrap_or("unknown")),
            escape_html(event.verified_by.as_deref().unwrap_or("unknown")),
        ));
        if let (Some(source), Some(target)) = (&event.source_address, &event.target_address) {
            html.push_str(&format!(
                "<span class=\"addr\">{} \u{2192} {} \u{00b7} {}</span>",
                escape_html(source),
                escape_html(target),
                escape_html(event.transport.as_deref().unwrap_or("?")),
            ));
        }
        if let Some(link) = permalink(event) {
            html.push_str(&format!(
                "<span class=\"permalink\">{}</span>",
                escape_html(&link)
            ));
        }
        html.push_str("</div>");
    }
    html.push_str("</article>");
}

/// ADR 030 D5/D7: the read-only audit-trail view — a chain-shape verification
/// summary plus the full `audit_records` chain for the session (the feed shows
/// only the latest proof per message; the whole chain lives here).
fn render_audit(connection: &Connection, selected_session_id: Option<&str>) -> CliResult<String> {
    let sessions = load_sessions(connection)?;
    let selected = selected_session_id
        .and_then(|id| sessions.iter().find(|session| session.session_id == id))
        .or_else(|| sessions.first());
    let mut html = String::new();
    html.push_str("<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\">");
    html.push_str("<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\">");
    html.push_str("<title>zynk audit</title><style>");
    html.push_str(STYLES);
    html.push_str("</style></head><body><div class=\"app-shell\"><main class=\"timeline\">");
    html.push_str("<header class=\"timeline-header\"><div><h1>Audit Trail</h1></div></header>");
    if let Some(session) = selected {
        let verification = verify_chain(connection, &session.session_id)?;
        let label = if verification.ok {
            format!(
                "chain intact \u{00b7} {} verified",
                verification.verified_count
            )
        } else {
            format!(
                "chain anomaly at {}",
                verification.broken_at.unwrap_or_default()
            )
        };
        html.push_str(&format!("<p class=\"verify\">{}</p>", escape_html(&label)));
        let mut statement = connection
            .prepare(
                "SELECT audit_id, COALESCE(previous_audit_id, 'genesis'), record_type,
                        delivery_status, verified_by, payload_hash, timestamp
                 FROM audit_records WHERE session_id = ?1 ORDER BY timestamp, audit_id",
            )
            .map_err(|error| CliError::failure(format!("failed to query audit view: {error}")))?;
        let rows = statement
            .query_map([session.session_id.as_str()], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })
            .map_err(|error| CliError::failure(format!("failed to read audit view: {error}")))?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|error| CliError::failure(format!("failed to read audit view: {error}")))?;
        for (audit_id, previous, record_type, delivery, verified, hash, timestamp) in rows {
            html.push_str(&format!(
                "<article class=\"feed-item\"><div class=\"timestamp\">{}</div><h2>{} <span class=\"kind\">{}</span></h2><p>\u{2190} {} \u{00b7} {} / {} \u{00b7} {}</p></article>",
                escape_html(&timestamp),
                escape_html(&audit_id),
                escape_html(&record_type),
                escape_html(&previous),
                escape_html(&delivery),
                escape_html(&verified),
                escape_html(&hash),
            ));
        }
    } else {
        html.push_str("<section class=\"empty-state\">No session.</section>");
    }
    html.push_str("</main></div></body></html>");
    Ok(html)
}

fn load_sessions(connection: &Connection) -> CliResult<Vec<DashboardSession>> {
    let mut statement = connection
        .prepare(
            // ADR 027 / v0.3.1: derive displayed current-state from the latest
            // status_event (already joined as se), falling back to the sessions
            // row only when no status_event exists. Import is append-only and does
            // not advance the sessions row, so an import-only session would
            // otherwise render a stale phase/mode/workflow_status/updated_at.
            "SELECT
                s.session_id,
                s.title,
                COALESCE(se.phase, s.phase),
                COALESCE(se.mode, s.mode),
                COALESCE(se.workflow_status, s.workflow_status),
                COALESCE(s.lead_agent_id, 'unknown'),
                COALESCE(s.artifact_ref, 'unknown'),
                COALESCE(se.timestamp, s.updated_at),
                COALESCE(se.next_action, 'unknown'),
                COALESCE(se.blockers, 'unknown'),
                COALESCE(se.asks_for_zevs, 'unknown'),
                COALESCE(se.risk_or_residual_uncertainty, 'unknown'),
                COALESCE(se.expected_wait, 'unknown')
             FROM sessions AS s
             LEFT JOIN status_events AS se
               ON se.status_event_id = (
                 SELECT status_event_id
                 FROM status_events
                 WHERE session_id = s.session_id
                 ORDER BY timestamp DESC, status_event_id DESC
                 LIMIT 1
               )
             ORDER BY COALESCE(se.timestamp, s.updated_at) DESC, s.session_id",
        )
        .map_err(|error| {
            CliError::failure(format!("failed to query dashboard sessions: {error}"))
        })?;
    let sessions = statement
        .query_map([], |row| {
            Ok(DashboardSession {
                session_id: row.get(0)?,
                title: row.get(1)?,
                phase: row.get(2)?,
                mode: row.get(3)?,
                workflow_status: row.get(4)?,
                lead_agent_id: row.get(5)?,
                artifact_ref: row.get(6)?,
                updated_at: row.get(7)?,
                next_action: row.get(8)?,
                blockers: row.get(9)?,
                asks_for_zevs: row.get(10)?,
                risk_or_residual_uncertainty: row.get(11)?,
                expected_wait: row.get(12)?,
            })
        })
        .map_err(|error| CliError::failure(format!("failed to read dashboard sessions: {error}")))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| {
            CliError::failure(format!("failed to read dashboard sessions: {error}"))
        })?;
    Ok(sessions)
}

pub(crate) fn escape_html(value: &str) -> String {
    value
        .replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
        .replace('"', "&quot;")
        .replace('\'', "&#39;")
}

fn escape_class(value: &str) -> String {
    escape_html(value)
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') {
                ch
            } else {
                '-'
            }
        })
        .collect()
}

pub(crate) fn escape_url_component(value: &str) -> String {
    let mut escaped = String::new();
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
            escaped.push(byte as char);
        } else {
            escaped.push_str(&format!("%{byte:02X}"));
        }
    }
    escaped
}

pub(crate) fn percent_decode(value: &str) -> String {
    let mut decoded = Vec::new();
    let bytes = value.as_bytes();
    let mut index = 0;
    while index < bytes.len() {
        if bytes[index] == b'%' && index + 2 < bytes.len() {
            if let Ok(hex) = std::str::from_utf8(&bytes[index + 1..index + 3]) {
                if let Ok(byte) = u8::from_str_radix(hex, 16) {
                    decoded.push(byte);
                    index += 3;
                    continue;
                }
            }
        }
        decoded.push(bytes[index]);
        index += 1;
    }
    String::from_utf8_lossy(&decoded).to_string()
}

const STYLES: &str = r#"
:root { color-scheme: light; --ink: #1c2024; --muted: #667085; --line: #d6dbe1; --panel: #f7f8fa; --accent: #0f766e; --warn: #9a3412; --ok: #166534; }
* { box-sizing: border-box; }
body { margin: 0; font: 14px/1.45 system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; color: var(--ink); background: #ffffff; letter-spacing: 0; }
.app-shell { min-height: 100vh; display: grid; grid-template-columns: minmax(220px, 18vw) minmax(0, 1fr) minmax(260px, 22vw); }
.sidebar, .detail-panel { background: var(--panel); border-color: var(--line); padding: 18px; overflow: auto; }
.sidebar { border-right: 1px solid var(--line); }
.detail-panel { border-left: 1px solid var(--line); }
.brand { font-weight: 700; font-size: 18px; margin-bottom: 18px; }
.session-link { display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: 8px; align-items: center; color: inherit; text-decoration: none; padding: 9px 0; border-bottom: 1px solid var(--line); }
.badge, .proof { display: inline-flex; align-items: center; min-height: 24px; padding: 3px 8px; border: 1px solid var(--line); border-radius: 6px; background: #fff; font-size: 12px; white-space: nowrap; }
.status-working, .proof-observed { border-color: #86efac; color: var(--ok); }
.status-blocked, .status-waiting-for-operator, .proof-failed { border-color: #fdba74; color: var(--warn); }
.proof-sent { border-color: #5eead4; color: var(--accent); }
.status-idle, .status-done, .proof-drafted, .proof-unknown { border-color: #d0d5dd; color: var(--muted); }
.timeline { padding: 20px clamp(18px, 3vw, 42px); overflow: auto; }
.timeline-header { display: flex; justify-content: space-between; align-items: end; border-bottom: 1px solid var(--line); margin-bottom: 18px; padding-bottom: 12px; }
h1 { font-size: 24px; margin: 0; }
h2 { font-size: 15px; margin: 4px 0; }
p { color: var(--muted); margin: 4px 0; }
.timeline-item { max-width: 860px; border: 1px solid var(--line); border-radius: 8px; padding: 14px 16px; margin: 0 0 12px; background: #fff; }
.timestamp { color: var(--muted); font-size: 12px; }
.proof-strip { display: flex; flex-wrap: wrap; gap: 6px; margin-top: 10px; }
dl { display: grid; grid-template-columns: 96px minmax(0, 1fr); gap: 9px 12px; margin: 0; }
dt { color: var(--muted); }
dd { margin: 0; overflow-wrap: anywhere; }
.empty, .empty-state { color: var(--muted); }
.roster-panel { margin-top: 18px; padding-top: 12px; border-top: 1px solid var(--line); }
.roster { list-style: none; margin: 0; padding: 0; }
.roster li { display: flex; align-items: center; gap: 8px; padding: 5px 0; }
.feed-item { max-width: 860px; border: 1px solid var(--line); border-radius: 8px; padding: 14px 16px; margin: 0 0 12px; background: #fff; }
.feed-item .body { color: var(--ink); white-space: pre-wrap; overflow-wrap: anywhere; margin: 6px 0; }
.feed-item .redacted { color: var(--muted); font-style: italic; }
.kind { color: var(--muted); font-weight: 400; font-size: 12px; }
.mid { color: var(--muted); font-size: 12px; }
.addr, .permalink { color: var(--muted); font-size: 12px; }
.verify { color: var(--ok); font-weight: 600; }
@media (max-width: 900px) { .app-shell { grid-template-columns: 1fr; } .sidebar, .detail-panel { border: 0; border-bottom: 1px solid var(--line); } }
"#;

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

    #[test]
    fn windowed_keeps_last_n_else_all() {
        // ADR 032 D4: the SSE/initial feed is a windowed (last-N) oldest-first feed.
        // len > n -> keep the last n (the suffix); len <= n -> keep all.
        let five = [1, 2, 3, 4, 5];
        assert_eq!(windowed(&five, 3), vec![3, 4, 5]);
        let two = [1, 2];
        assert_eq!(windowed(&two, 3), vec![1, 2]);
        // exact boundary -> all
        assert_eq!(windowed(&five, 5), vec![1, 2, 3, 4, 5]);
        // empty -> empty
        assert_eq!(windowed::<i32>(&[], 3), Vec::<i32>::new());
    }
}