nornir 0.4.4

//! Thin-client timeline loader: fetch a [`Timeline`] from a running
//! `nornir-server` over the `Viz.Timeline` gRPC instead of opening a local
//! Iceberg warehouse. The server builds the timeline from the warehouse it
//! owns (it holds the redb lock) and returns it as JSON; we deserialize into
//! the same [`Timeline`] the embedded path produces, so the egui app is
//! source-agnostic.

use anyhow::{Context, Result};

use super::live::LiveEvent;
use super::model::Timeline;
use crate::warehouse::iceberg::TablePreview;

mod pb {
    tonic::include_proto!("nornir.v1");
}

/// Build an `http://…` endpoint url + a `Bearer <token>` metadata value — the
/// shared connect/auth shape for the viz gRPC clients.
fn endpoint_and_bearer(
    endpoint: &str,
    token: &str,
) -> Result<(String, tonic::metadata::MetadataValue<tonic::metadata::Ascii>)> {
    let endpoint = if endpoint.starts_with("http") {
        endpoint.to_string()
    } else {
        format!("http://{endpoint}")
    };
    let bearer = format!("Bearer {token}").parse().context("parse bearer token")?;
    Ok((endpoint, bearer))
}

/// The `nornir-workspace` metadata value for `workspace` (empty ⇒ none) — selects
/// which served workspace the gRPC calls target. Driven by the app's currently
/// selected workspace (the in-UI picker), not an env var, so the viz can switch
/// workspaces live.
fn ws_header(workspace: &str) -> Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> {
    (!workspace.is_empty()).then(|| workspace.parse().ok()).flatten()
}

/// List the workspaces the server has registered (`Workspaces.List` RPC) — the
/// names that populate the viz's workspace picker.
pub fn list_workspaces(endpoint: &str, token: &str) -> Result<Vec<String>> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz client")?;
    rt.block_on(async {
        let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let mut client = pb::workspaces_client::WorkspacesClient::with_interceptor(
            channel,
            move |mut req: tonic::Request<()>| {
                req.metadata_mut().insert("authorization", bearer.clone());
                Ok(req)
            },
        );
        let resp = client.list(pb::Empty {}).await.context("Workspaces.List RPC")?.into_inner();
        Ok(resp.workspaces.into_iter().map(|w| w.name).collect())
    })
}

/// Fetch the timeline for `workspace` from `endpoint` (e.g.
/// `http://127.0.0.1:7878`), authenticating with the bearer `token`. Runs the
/// async tonic call on a private current-thread runtime so it's safe to call
/// from the synchronous egui update loop.
pub fn fetch_timeline(endpoint: &str, token: &str, workspace: &str) -> Result<Timeline> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz client")?;
    rt.block_on(async {
        let endpoint = if endpoint.starts_with("http") {
            endpoint.to_string()
        } else {
            format!("http://{endpoint}")
        };
        let bearer: tonic::metadata::MetadataValue<tonic::metadata::Ascii> =
            format!("Bearer {token}").parse().context("parse bearer token")?;
        let ws_md = ws_header(workspace);
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let mut client = pb::viz_client::VizClient::with_interceptor(
            channel,
            move |mut req: tonic::Request<()>| {
                req.metadata_mut().insert("authorization", bearer.clone());
                if let Some(ws) = &ws_md {
                    req.metadata_mut().insert("nornir-workspace", ws.clone());
                }
                Ok(req)
            },
        );
        let resp = client
            .timeline(pb::VizTimelineRequest { workspace: workspace.to_string() })
            .await
            .context("Viz.Timeline RPC")?
            .into_inner();
        let timeline: Timeline =
            serde_json::from_str(&resp.json).context("decode timeline json from server")?;
        Ok(timeline)
    })
}

/// Open the server's `Release.Progress` server-stream and invoke `on_event`
/// for each converted [`LiveEvent`]. Blocks until the stream closes (the server
/// ends it after `RunEnd`) or errors — call it from a dedicated `std::thread`
/// (see [`super::live`]). Runs its own current-thread tokio runtime so it never
/// touches the egui loop. This is what makes a remote viz animate a release run
/// in real time over Tailscale: same events the local file tail would produce.
pub fn stream_progress(
    endpoint: &str,
    token: &str,
    mut on_event: impl FnMut(LiveEvent),
) -> Result<()> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz live client")?;
    rt.block_on(async {
        let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let mut client = pb::release_client::ReleaseClient::with_interceptor(
            channel,
            move |mut req: tonic::Request<()>| {
                req.metadata_mut().insert("authorization", bearer.clone());
                Ok(req)
            },
        );
        let mut stream = client
            .progress(pb::Empty {})
            .await
            .context("Release.Progress RPC")?
            .into_inner();
        while let Some(ev) = stream.message().await.context("progress stream")? {
            if let Some(live) = to_live(ev) {
                on_event(live);
            }
        }
        Ok(())
    })
}

/// List every warehouse table the server owns (the `Warehouse.Tables` RPC) —
/// the remote counterpart to `IcebergWarehouse::table_names`.
pub fn fetch_tables(endpoint: &str, token: &str, workspace: &str) -> Result<Vec<String>> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz client")?;
    rt.block_on(async {
        let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let ws_md = ws_header(workspace);
        let mut client = pb::warehouse_client::WarehouseClient::with_interceptor(
            channel,
            move |mut req: tonic::Request<()>| {
                req.metadata_mut().insert("authorization", bearer.clone());
                if let Some(ws) = &ws_md {
                    req.metadata_mut().insert("nornir-workspace", ws.clone());
                }
                Ok(req)
            },
        );
        let resp = client.tables(pb::Empty {}).await.context("Warehouse.Tables RPC")?.into_inner();
        Ok(resp.names)
    })
}

/// Scan one warehouse table for display (the `Warehouse.Scan` RPC) — the remote
/// counterpart to `IcebergWarehouse::scan_preview`. `limit` 0 = server default.
pub fn scan_table(endpoint: &str, token: &str, table: &str, limit: u32, workspace: &str) -> Result<TablePreview> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz client")?;
    rt.block_on(async {
        let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let ws_md = ws_header(workspace);
        let mut client = pb::warehouse_client::WarehouseClient::with_interceptor(
            channel,
            move |mut req: tonic::Request<()>| {
                req.metadata_mut().insert("authorization", bearer.clone());
                if let Some(ws) = &ws_md {
                    req.metadata_mut().insert("nornir-workspace", ws.clone());
                }
                Ok(req)
            },
        );
        let resp = client
            .scan(pb::WarehouseScanRequest { table: table.to_string(), limit })
            .await
            .context("Warehouse.Scan RPC")?
            .into_inner();
        Ok(TablePreview {
            columns: resp.columns,
            rows: resp.rows.into_iter().map(|r| r.cells).collect(),
        })
    })
}

// ── shared connect helper + reusable interceptor ────────────────────────────

/// Bearer + optional `nornir-workspace` header, reusable across every generated
/// client (so the new RPC wrappers below don't each re-inline the closure).
#[derive(Clone)]
pub(crate) struct Auth {
    bearer: tonic::metadata::MetadataValue<tonic::metadata::Ascii>,
    ws: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
}
impl tonic::service::Interceptor for Auth {
    fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
        req.metadata_mut().insert("authorization", self.bearer.clone());
        if let Some(ws) = &self.ws {
            req.metadata_mut().insert("nornir-workspace", ws.clone());
        }
        Ok(req)
    }
}

/// Run one blocking gRPC call on a private current-thread runtime: connect to
/// `endpoint`, build the [`Auth`] interceptor (bearer + `workspace` header), and
/// hand `(channel, auth)` to `f`. Keeps the egui loop synchronous.
fn call<T, F, Fut>(endpoint: &str, token: &str, workspace: &str, f: F) -> Result<T>
where
    F: FnOnce(tonic::transport::Channel, Auth) -> Fut,
    Fut: std::future::Future<Output = Result<T>>,
{
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime for viz client")?;
    rt.block_on(async {
        let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
        let channel = tonic::transport::Channel::from_shared(endpoint.clone())
            .with_context(|| format!("invalid server url `{endpoint}`"))?
            .connect()
            .await
            .with_context(|| format!("connect to nornir-server at {endpoint}"))?;
        let auth = Auth { bearer, ws: ws_header(workspace) };
        f(channel, auth).await
    })
}

// ── plain view structs (UI consumes these, not the pb types) ─────────────────

/// Workspace info for the picker's side panel (`Workspaces.Get`).
#[derive(Clone, Debug, Default)]
pub struct WorkspaceInfo {
    pub name: String,
    pub mode: String,
    pub poll: String,
    pub current_snapshot: String,
    pub updated_at: String,
    /// `(member, source/sha summary)` pairs.
    pub members: Vec<(String, String)>,
}

/// One search hit (`Search.Query`).
#[derive(Clone, Debug)]
pub struct Hit {
    pub corpus: String,
    pub repo: String,
    pub path: String,
    pub score: f32,
    pub title: String,
    pub snippet: String,
}

/// One symbol (`Knowledge.SymbolLookup`).
#[derive(Clone, Debug)]
pub struct KnownSym {
    pub crate_name: String,
    pub item_kind: String,
    pub item_name: String,
    pub visibility: String,
    pub file: String,
    pub line: u32,
    pub signature: String,
}

/// Release-gate outcome for a repo (`Release.GateAll`).
#[derive(Clone, Debug, Default)]
pub struct GateReport {
    pub repo: String,
    pub passed: Vec<String>,
    pub failed: Vec<(String, String)>,
}

/// One bench metric series point for charting (`Bench.History`).
#[derive(Clone, Debug)]
pub struct BenchPoint {
    pub date: String,
    pub version: String,
    pub metric: String,
    pub value: f64,
}

// ── new clickable-surface RPC wrappers ───────────────────────────────────────

/// `Workspaces.Get` — the info-panel backing call for the picker selection.
pub fn get_workspace(endpoint: &str, token: &str, name: &str) -> Result<WorkspaceInfo> {
    call(endpoint, token, "", |channel, auth| async move {
        let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
        let r = c
            .get(pb::WorkspaceName { name: name.to_string() })
            .await
            .context("Workspaces.Get RPC")?
            .into_inner();
        Ok(WorkspaceInfo {
            name: r.name,
            mode: r.mode,
            poll: r.poll,
            current_snapshot: r.current_snapshot,
            updated_at: r.updated_at,
            members: r
                .members
                .into_iter()
                .map(|m| {
                    let sha = if m.last_seen_sha.len() >= 8 { &m.last_seen_sha[..8] } else { &m.last_seen_sha };
                    (m.name, format!("{} @ {sha} [{}]", m.remote, m.sync_state))
                })
                .collect(),
        })
    })
}

/// `Workspaces.Fetch` — the "⟳ Sync now" button. Returns `(fetched, changed, errors)`.
pub fn fetch_workspace(
    endpoint: &str,
    token: &str,
    name: &str,
) -> Result<(u32, Vec<String>, Vec<String>)> {
    call(endpoint, token, "", |channel, auth| async move {
        let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
        let r = c
            .fetch(pb::WorkspaceName { name: name.to_string() })
            .await
            .context("Workspaces.Fetch RPC")?
            .into_inner();
        Ok((r.fetched, r.changed, r.errors))
    })
}

/// `Search.Query` — BM25 over the Tantivy corpora. `corpus`/`repo` empty = all.
pub fn search(
    endpoint: &str,
    token: &str,
    query: &str,
    corpus: &str,
    repo: &str,
    limit: u32,
    workspace: &str,
) -> Result<Vec<Hit>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::search_client::SearchClient::with_interceptor(channel, auth);
        let r = c
            .query(pb::SearchRequest {
                query: query.to_string(),
                corpus: corpus.to_string(),
                repo: repo.to_string(),
                limit,
            })
            .await
            .context("Search.Query RPC")?
            .into_inner();
        Ok(r.hits
            .into_iter()
            .map(|h| Hit {
                corpus: h.corpus,
                repo: h.repo,
                path: h.path,
                score: h.score,
                title: h.title,
                snippet: h.snippet,
            })
            .collect())
    })
}

/// `Knowledge.SymbolLookup` — item-name substring search over `symbol_facts`.
pub fn knowledge_lookup(
    endpoint: &str,
    token: &str,
    repo: &str,
    arg: &str,
    limit: u32,
    workspace: &str,
) -> Result<Vec<KnownSym>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
        let r = c
            .symbol_lookup(pb::KnowledgeSymbolQuery {
                repo: repo.to_string(),
                arg: arg.to_string(),
                limit,
            })
            .await
            .context("Knowledge.SymbolLookup RPC")?
            .into_inner();
        Ok(r.symbols
            .into_iter()
            .map(|s| KnownSym {
                crate_name: s.crate_name,
                item_kind: s.item_kind,
                item_name: s.item_name,
                visibility: s.visibility,
                file: s.file,
                line: s.line,
                signature: s.signature,
            })
            .collect())
    })
}

/// `Release.GateAll` — run every release gate for `repo`, pass/fail split.
pub fn gate_all(endpoint: &str, token: &str, repo: &str, workspace: &str) -> Result<GateReport> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::release_client::ReleaseClient::with_interceptor(channel, auth);
        let r = c
            .gate_all(pb::RepoOnly { repo: repo.to_string() })
            .await
            .context("Release.GateAll RPC")?
            .into_inner();
        Ok(GateReport {
            repo: r.repo,
            passed: r.passed,
            failed: r.failed.into_iter().map(|f| (f.name, f.error)).collect(),
        })
    })
}

/// `Release.Trace` — regression time-bisect JSON for `repo` (raw, UI pretty-prints).
pub fn trace(
    endpoint: &str,
    token: &str,
    repo: &str,
    workspace: &str,
) -> Result<String> {
    let ws = workspace.to_string();
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::release_client::ReleaseClient::with_interceptor(channel, auth);
        let r = c
            .trace(pb::TraceQuery { repo: repo.to_string(), workspace: ws })
            .await
            .context("Release.Trace RPC")?
            .into_inner();
        Ok(r.json)
    })
}

/// `Bench.History` — flatten every run's metrics into chartable points.
pub fn bench_history(
    endpoint: &str,
    token: &str,
    repo: &str,
    workspace: &str,
) -> Result<Vec<BenchPoint>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::bench_client::BenchClient::with_interceptor(channel, auth);
        let r = c
            .history(pb::RepoOnly { repo: repo.to_string() })
            .await
            .context("Bench.History RPC")?
            .into_inner();
        let mut pts = Vec::new();
        for run in r.runs {
            for res in run.results {
                for kvf in res.metrics {
                    pts.push(BenchPoint {
                        date: run.date.clone(),
                        version: run.version.clone(),
                        metric: format!("{}::{}", res.name, kvf.key),
                        value: kvf.value,
                    });
                }
            }
        }
        Ok(pts)
    })
}

/// One semantic-search hit (`Vector.Search`).
#[derive(Clone, Debug)]
pub struct VecHit {
    pub score: f64,
    pub file: String,
    pub start_line: u64,
    pub end_line: u64,
}

/// `Vector.Search` — GPU/CPU semantic search over a repo's embedded snapshot.
/// `repo` required; `sha` empty = latest. Returns hits or a clear error (the
/// server replies UNIMPLEMENTED when built without an embedder).
pub fn vector_search(
    endpoint: &str,
    token: &str,
    repo: &str,
    query: &str,
    limit: u32,
    workspace: &str,
) -> Result<Vec<VecHit>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::vector_client::VectorClient::with_interceptor(channel, auth);
        let r = c
            .search(pb::VectorSearchRequest {
                repo: repo.to_string(),
                query: query.to_string(),
                sha: String::new(),
                limit,
            })
            .await
            .context("Vector.Search RPC")?
            .into_inner();
        let v: serde_json::Value =
            serde_json::from_str(&r.json).context("decode vector hits json")?;
        let hits = v
            .get("hits")
            .and_then(|h| h.as_array())
            .map(|arr| {
                arr.iter()
                    .map(|h| VecHit {
                        score: h.get("score").and_then(|x| x.as_f64()).unwrap_or(0.0),
                        file: h.get("file").and_then(|x| x.as_str()).unwrap_or("").to_string(),
                        start_line: h.get("start_line").and_then(|x| x.as_u64()).unwrap_or(0),
                        end_line: h.get("end_line").and_then(|x| x.as_u64()).unwrap_or(0),
                    })
                    .collect()
            })
            .unwrap_or_default();
        Ok(hits)
    })
}

/// `Index.Stats` — `(total_docs, per-corpus counts)` for the search status panel.
pub fn index_stats(endpoint: &str, token: &str, workspace: &str) -> Result<(u64, Vec<(String, String)>)> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::index_client::IndexClient::with_interceptor(channel, auth);
        let r = c.stats(pb::Empty {}).await.context("Index.Stats RPC")?.into_inner();
        Ok((r.total, r.by_corpus.into_iter().map(|kv| (kv.key, kv.value)).collect()))
    })
}

/// `Knowledge.Callers` / `Callees` — who-calls / who-is-called-by `name`.
/// `callers = true` → Callers, else Callees. Returns `(caller_path, callee_ident,
/// file, line)` rows.
pub fn knowledge_calls(
    endpoint: &str,
    token: &str,
    repo: &str,
    name: &str,
    callers: bool,
    limit: u32,
    workspace: &str,
) -> Result<Vec<(String, String, String, u32)>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
        let q = pb::KnowledgeCallQuery { repo: repo.to_string(), name: name.to_string(), limit };
        let calls = if callers {
            c.callers(q).await.context("Knowledge.Callers RPC")?
        } else {
            c.callees(q).await.context("Knowledge.Callees RPC")?
        }
        .into_inner()
        .calls;
        Ok(calls.into_iter().map(|k| (k.caller_path, k.callee_ident, k.file, k.line)).collect())
    })
}

/// `Knowledge.CallPath` — a call path from `from` to `to` (empty = none found).
pub fn knowledge_call_path(
    endpoint: &str,
    token: &str,
    repo: &str,
    from: &str,
    to: &str,
    workspace: &str,
) -> Result<Vec<String>> {
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
        let r = c
            .call_path(pb::KnowledgeCallPathQuery {
                repo: repo.to_string(),
                from: from.to_string(),
                to: to.to_string(),
            })
            .await
            .context("Knowledge.CallPath RPC")?
            .into_inner();
        Ok(r.names)
    })
}

/// `Funnel.Show` — the whole idea→plan funnel, flattened into a render-ready
/// [`FunnelView`]. The `FunnelDump` already carries per-node `deps`, so the
/// DAG is fully reconstructable without any extra RPC.
pub fn funnel_show(
    endpoint: &str,
    token: &str,
    workspace: &str,
) -> Result<super::funnel_view::FunnelView> {
    use super::funnel_view::{FunnelView, NodeStat, NodeView, PlanView};
    call(endpoint, token, workspace, |channel, auth| async move {
        let mut c = pb::funnel_client::FunnelClient::with_interceptor(channel, auth);
        let dump = c.show(pb::Empty {}).await.context("Funnel.Show RPC")?.into_inner();
        let mut plans = Vec::new();
        for idea in dump.ideas {
            for plan in idea.plans {
                let nodes = plan
                    .nodes
                    .into_iter()
                    .map(|n| NodeView {
                        id: n.id,
                        kind: n.kind,
                        title: n.title,
                        status: NodeStat::parse(&n.status),
                        targets: Vec::new(), // FunnelDumpNode carries no targets
                        deps: n.deps,
                    })
                    .collect();
                plans.push(PlanView {
                    id: plan.id,
                    summary: plan.summary,
                    status: plan.status.to_ascii_lowercase(),
                    idea_text: idea.text.clone(),
                    nodes,
                });
            }
        }
        plans.sort_by(|a, b| a.id.cmp(&b.id));
        Ok(FunnelView { plans })
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    /// Live smoke against `$NORNIR_SERVER` (+ `$NORNIR_SERVER_TOKEN`). Ignored by
    /// default (needs a running server). Run with:
    ///   NORNIR_SERVER=http://oden:7878 NORNIR_SERVER_TOKEN=… \
    ///     cargo test --features viz --lib viz::remote::tests -- --ignored --nocapture
    #[test]
    #[ignore]
    fn live_list_workspaces() {
        let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
        let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
        let ws = list_workspaces(&ep, &tok).expect("list_workspaces should succeed");
        eprintln!("live workspaces: {ws:?}");
        assert!(!ws.is_empty(), "server returned no workspaces");
    }
}

/// Convert a proto `ProgressEvent` (the oneof) into the viz [`LiveEvent`] —
/// the same enum the local NDJSON tail deserializes, so the pane is wire-source
/// agnostic. Unknown/empty kinds are dropped (forward-compat).
fn to_live(ev: pb::ProgressEvent) -> Option<LiveEvent> {
    use pb::progress_event::Kind;
    Some(match ev.kind? {
        Kind::RunStart(x) => LiveEvent::RunStart { run_id: x.run_id, workspace: x.workspace },
        Kind::RepoStart(x) => LiveEvent::RepoStart { repo: x.repo, sha: x.sha },
        Kind::PhaseStart(x) => LiveEvent::PhaseStart { repo: x.repo, phase: x.phase },
        Kind::PhaseEnd(x) => LiveEvent::PhaseEnd {
            repo: x.repo,
            phase: x.phase,
            ok: x.ok,
            duration_ms: x.duration_ms,
        },
        Kind::BinaryStart(x) => LiveEvent::BinaryStart { repo: x.repo, binary: x.binary },
        Kind::TestPass(x) => LiveEvent::TestPass { repo: x.repo, binary: x.binary, name: x.name },
        Kind::TestFail(x) => LiveEvent::TestFail { repo: x.repo, binary: x.binary, name: x.name },
        Kind::BinaryDone(x) => LiveEvent::BinaryDone {
            repo: x.repo,
            binary: x.binary,
            passed: x.passed,
            failed: x.failed,
        },
        Kind::RepoEnd(x) => LiveEvent::RepoEnd { repo: x.repo, ok: x.ok },
        Kind::RunEnd(x) => LiveEvent::RunEnd { run_id: x.run_id, ok: x.ok },
    })
}