dwarves-gui/src/codeflow.rs
//! 🌊 **Codeflow** — the call graph as LIVING FLOW (the owner, 2026-10-18:
//! "facett has a dataflow component — would that work for codeflow, making the
//! lines software?" → "yes codeflow").
//!
//! Mounts the shared [`facett_flow::FlowGraph`] (the host-clock animated flow
//! graph) over the SELECTED workspace's code room:
//!
//! * **Nodes + lines** = the room's call graph, the same `Viz.CallGraph` answer
//! the Data view's 3D wall reads (nodes in the reader's order; an edge's
//! `from`/`to` are indexes into them).
//! * **The flow ON a line** = measured traffic. With a traced run in the
//! warehouse (`Viz.CodeMap` lod `trace_runs` / `trace`, the replay's own
//! surfaces) each edge flows at ITS calls-per-second from that run; without
//! one, the static call-site count stands in, and the pane SAYS which of the
//! two it is showing (`rated_by`). The lines carry the software's real
//! traffic — a hot path flows hard, a cold one barely moves.
//!
//! States that must never draw alike: un-scanned (reads itself on open),
//! asking, a refusal (the state + the reader's sentence, never an empty flow
//! that reads as "no calls"), and the flow itself.
use std::sync::Arc;
use facett_core::{Elm, Facet};
use crate::room::{self, Pending, RoomAnswer, RoomSource};
use crate::trace_replay;
/// The call-graph surface — dataview's, spelled once there.
pub use crate::dataview::S_CALLS;
/// The replay surfaces (traced runs + one run's events).
pub const S_MAP: &str = "Viz.CodeMap";
/// Edges asked of the room (the reader ranks; a bigger graph is cut, not wrong).
pub const EDGE_LIMIT: u64 = 4_000;
/// Events of the rated run read in one page (rates need counts, not a replay).
pub const TRACE_PAGE: u64 = 65_536;
/// One scan's answers, landed together.
struct FlowRead {
calls: RoomAnswer,
/// The latest traced run and its event window, when the warehouse holds one.
trace: Option<(trace_replay::RunRow, RoomAnswer)>,
}
/// The 🌊 Codeflow tab.
pub struct CodeflowPane {
room: Option<Arc<dyn RoomSource>>,
flow: Option<facett_flow::FlowGraph>,
/// What the line rates ARE: `trace <run_id>` or `static call sites` —
/// stated on the pane, never implied.
rated_by: String,
/// How many edges carry a non-zero rate.
rated: usize,
/// The read's state line (asking / the refusal / the counts).
status: String,
pending: Option<Pending<FlowRead>>,
auto_read: bool,
}
impl CodeflowPane {
pub fn new() -> Self {
Self {
room: None,
flow: None,
rated_by: String::new(),
rated: 0,
status: "un-scanned".into(),
pending: None,
auto_read: false,
}
}
/// The pane asking `room` (the forge in the shipped wiring; canned in tests).
pub fn with_room(mut self, room: Arc<dyn RoomSource>) -> Self {
self.room = Some(room);
self
}
/// **Read the flow**: the call graph, the latest traced run, that run's
/// events — one worker, off the UI thread.
pub fn scan(&mut self) {
let Some(room) = &self.room else {
self.status = "this pane was given no code room to ask".into();
return;
};
if self.pending.is_some() {
return;
}
self.status = "asking the room…".into();
let room = Arc::clone(room);
self.pending = Some(Pending::spawn("codeflow-scan", move || {
let calls = room.ask(S_CALLS, &serde_json::json!({ "limit": EDGE_LIMIT }));
// The latest traced run, when the warehouse holds one — its events
// are the rates; a workspace with no run is NOT an error here.
let runs = room.ask(S_MAP, &serde_json::json!({ "lod": "trace_runs", "limit": 1 }));
let trace = trace_replay::runs_of(&runs)
.ok()
.and_then(|rs| rs.into_iter().next())
.map(|run| {
let window = room.ask(
S_MAP,
&serde_json::json!({ "lod": "trace", "run_id": run.run_id, "limit": TRACE_PAGE }),
);
(run, window)
});
FlowRead { calls, trace }
}));
}
/// Non-blocking: land the in-flight read. Called once per frame.
pub fn poll(&mut self) {
let Some(p) = &self.pending else { return };
if let Some(res) = p.try_take() {
self.pending = None;
match res {
Ok(read) => self.land(read),
Err(()) => self.status = "the codeflow read worker died before it answered".into(),
}
}
}
/// Blocking: wait for the in-flight read (the robot seam; the window never blocks).
pub fn wait(&mut self) {
if let Some(p) = self.pending.take() {
match p.wait() {
Ok(read) => self.land(read),
Err(()) => self.status = "the codeflow read worker died before it answered".into(),
}
}
}
fn land(&mut self, read: FlowRead) {
let a = &read.calls;
if !a.is_measured() {
// A refusal draws its state, never an empty flow that reads as "no calls".
self.flow = None;
self.status = format!("[{}] {}", a.state, a.evidence_text());
return;
}
// The reader's nodes in ITS order — an edge's from/to index into them.
let rows = a.rows("nodes");
let mut nodes: Vec<facett_flow::Node> = Vec::with_capacity(rows.len());
let mut id_of: Vec<String> = Vec::with_capacity(rows.len());
let mut seen = std::collections::HashSet::new();
for (i, n) in rows.iter().enumerate() {
let label_full = n["id"].as_str().unwrap_or("?").to_string();
let mut id = n["moniker"].as_str().unwrap_or(&label_full).to_string();
if !seen.insert(id.clone()) {
id = format!("{id}#{i}");
seen.insert(id.clone());
}
let label = label_full.rsplit("::").next().unwrap_or(&label_full).to_string();
let group = label_full.split("::").next().unwrap_or("?").to_string();
id_of.push(id.clone());
nodes.push(facett_flow::Node { id, label, position: [0.0, 0.0], group, kind: 0 });
}
if nodes.is_empty() {
self.flow = None;
self.status = "the room's call graph names no node".into();
return;
}
let mut edges: Vec<facett_flow::Edge> = Vec::new();
let mut sites: Vec<f32> = Vec::new();
for e in a.rows("edges") {
let (Some(f), Some(t)) = (
e["from"].as_u64().map(|x| x as usize),
e["to"].as_u64().map(|x| x as usize),
) else {
continue;
};
let (Some(from), Some(to)) = (id_of.get(f), id_of.get(t)) else { continue };
edges.push(facett_flow::Edge { from: from.clone(), to: to.clone() });
sites.push(e["sites"].as_u64().unwrap_or(0) as f32);
}
let mut graph = facett_flow::Graph { nodes, edges };
if graph.layered(facett_flow::LayeredOpts::default()).is_err() {
// A cyclic call graph refuses the DAG layering; a ring is still a
// layout, never a refusal to draw.
ring(&mut graph);
}
// The lines' rates: the traced run's measured calls/s when its ids land
// on this graph's nodes; the static call sites stand in otherwise — and
// `rated_by` SAYS which, never implies.
let static_rates = || sites.iter().map(|&s| Some(s)).collect::<Vec<Option<f32>>>();
let (rates, rated_by) = match &read.trace {
Some((run, window)) => match trace_replay::trace_of(window) {
Ok(view) => {
let mut count: std::collections::HashMap<(&str, &str), u64> =
std::collections::HashMap::new();
for ev in &view.events {
if ev.exit {
continue;
}
let Some(c) = ev.caller else { continue };
let (Some(from), Some(to)) = (view.heads.get(c), view.heads.get(ev.head))
else {
continue;
};
*count
.entry((from.stable_id.as_str(), to.stable_id.as_str()))
.or_default() += 1;
}
let secs = run.duration_ns.max(1) as f32 / 1e9;
let rates: Vec<Option<f32>> = graph
.edges
.iter()
.map(|e| {
count
.get(&(e.from.as_str(), e.to.as_str()))
.map(|&n| n as f32 / secs)
})
.collect();
if rates.iter().any(Option::is_some) {
(rates, format!("trace {}", run.run_id))
} else {
(static_rates(), "static call sites".to_string())
}
}
Err(_) => (static_rates(), "static call sites".to_string()),
},
None => (static_rates(), "static call sites".to_string()),
};
self.rated = rates.iter().filter(|r| matches!(r, Some(v) if *v > 0.0)).count();
self.rated_by = rated_by;
let (n, m) = (graph.nodes.len(), graph.edges.len());
match facett_flow::FlowGraph::new("codeflow", graph) {
Ok(mut flow) => {
flow.set_labels(true);
flow.set_compact_state(true);
let _ = Elm::update(
&mut flow,
facett_flow::Msg::Rates(Arc::new(facett_flow::attr::Rates { edges: rates })),
);
let _ = Elm::update(&mut flow, facett_flow::Msg::Fit);
self.flow = Some(flow);
self.status = format!("{n} node(s) · {m} line(s) · flow = {}", self.rated_by);
}
Err(e) => {
self.flow = None;
self.status = format!("the flow refused the graph: {e}");
}
}
}
}
/// The fallback layout for a graph the DAG layering refuses (a cycle): the
/// nodes on a circle, reader's order kept.
fn ring(graph: &mut facett_flow::Graph) {
let n = graph.nodes.len().max(1) as f32;
let r = 60.0 + 40.0 * n.sqrt();
for (i, node) in graph.nodes.iter_mut().enumerate() {
let a = (i as f32 / n) * std::f32::consts::TAU;
node.position = [r * a.cos(), r * a.sin()];
}
}
impl Default for CodeflowPane {
fn default() -> Self {
Self::new()
}
}
impl Facet for CodeflowPane {
fn title(&self) -> &str {
"Codeflow"
}
fn ui(&mut self, ui: &mut egui::Ui) {
self.poll();
if !self.auto_read && self.pending.is_none() && self.flow.is_none() {
self.auto_read = true;
self.scan();
}
ui.horizontal_wrapped(|ui| {
ui.heading("🌊 Codeflow");
if self.pending.is_some() {
ui.spinner();
ui.ctx()
.request_repaint_after(std::time::Duration::from_millis(120));
}
ui.label(egui::RichText::new(&self.status).monospace().small());
});
if let Some(flow) = &mut self.flow {
// The host clock: the flow animates at the measured rates.
let dt = ui.input(|i| i.stable_dt).clamp(0.0, 0.1);
let _ = Elm::update(flow, facett_flow::Msg::Tick(f64::from(dt)));
ui.ctx().request_repaint();
flow.ui(ui);
}
}
fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "codeflow",
"scanned": self.flow.is_some(),
"status": self.status,
"nodes": self.flow.as_ref().map(|f| f.state().graph.nodes.len()).unwrap_or(0),
"edges": self.flow.as_ref().map(|f| f.state().graph.edges.len()).unwrap_or(0),
"rated": self.rated,
"rated_by": self.rated_by,
"pending": self.pending.is_some(),
})
}
fn update_json(&mut self, msg_json: &str) {
let Ok(v) = serde_json::from_str::<serde_json::Value>(msg_json) else {
return;
};
if v.get("scan").and_then(|b| b.as_bool()).unwrap_or(false) || v == serde_json::json!("scan") {
self.scan();
self.wait();
}
}
}
/// The shipped 🌊 Codeflow tab, bound to the SELECTED workspace.
pub fn codeflow_pane(endpoint: &str, token: &str, workspace: &str) -> CodeflowPane {
CodeflowPane::new().with_room(room::ForgeRoom::shared(endpoint, token, workspace))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::room::testing::CannedRoom;
/// A room that answers `Viz.CodeMap` BY LOD (runs vs the window) — the two
/// asks share one surface, which a per-surface canned room cannot split.
struct LodRoom {
calls: serde_json::Value,
runs: serde_json::Value,
trace: serde_json::Value,
}
impl RoomSource for LodRoom {
fn ask(&self, surface: &str, params: &serde_json::Value) -> RoomAnswer {
let body = match (surface, params["lod"].as_str()) {
(S_CALLS, _) => self.calls.clone(),
(S_MAP, Some("trace_runs")) => self.runs.clone(),
(S_MAP, Some("trace")) => self.trace.clone(),
_ => serde_json::json!({ "state": "invalid_argument", "refused": true }),
};
RoomAnswer::from_body(surface, "lod room (test)", body)
}
fn describe(&self) -> String {
"lod room (test)".to_string()
}
}
/// **ROBOT — the lines carry the traced run's real traffic.** A canned room
/// serves the call graph (3 nodes, 2 edges) and ONE traced run whose events
/// walk checkout→charge twice and charge→audit once; the pane's flow must
/// hold that graph, rate the edges from the TRACE (calls/s of the run, not
/// the static `sites`), and say which run rated it.
#[test]
fn the_flow_carries_the_traced_runs_traffic() {
let calls = serde_json::json!({
"state": "complete", "total_edges": 2,
"nodes": [
{ "id": "shop::checkout", "moniker": "m:checkout" },
{ "id": "shop::pay::charge", "moniker": "m:charge" },
{ "id": "shop::pay::audit", "moniker": "m:audit" },
],
"edges": [
{ "from": 0, "to": 1, "sites": 3 },
{ "from": 1, "to": 2, "sites": 2 },
],
"evidence": { "missing": [], "notes": [] },
});
let runs = serde_json::json!({
"state": "complete", "lod": "trace_runs", "repo": "",
"runs": [{ "run_id": "r-1", "target": "shop", "commit": "c1", "source": "runa",
"events": 6, "threads": 1, "duration_ns": 2_000_000_000u64,
"mapped_events": 6 }],
});
// heads: 0=checkout 1=charge 2=audit; events [seq,t,thread,exit,head,caller]
let trace = serde_json::json!({
"state": "complete", "lod": "trace", "run_id": "r-1", "from_ns": 0,
"heads": [
{ "card": 0, "at": [0.0, 0.0], "stable_id": "m:checkout", "path": "shop", "symbol": "checkout" },
{ "card": 1, "at": [0.0, 0.0], "stable_id": "m:charge", "path": "shop", "symbol": "charge" },
{ "card": 2, "at": [0.0, 0.0], "stable_id": "m:audit", "path": "shop", "symbol": "audit" },
],
"events": [
[1, 0, 0, 0, 0, null], [2, 10, 0, 0, 1, 0], [3, 20, 0, 1, 1, 0],
[4, 30, 0, 0, 1, 0], [5, 40, 0, 0, 2, 1], [6, 50, 0, 1, 2, 1],
],
"unplaced": 0, "more": false,
});
let room = Arc::new(LodRoom { calls, runs, trace });
let mut p = CodeflowPane::new().with_room(room);
p.update_json(r#"{"scan":true}"#);
let s = p.state_json();
assert_eq!(s["nodes"], 3, "the room's call graph is the flow's graph: {s}");
assert_eq!(s["edges"], 2, "{s}");
// checkout→charge happened twice in 2 s (1.0/s), charge→audit once (0.5/s):
// both edges rated, and rated BY THE TRACE, not by static sites.
assert_eq!(s["rated"], 2, "{s}");
assert_eq!(s["rated_by"], "trace r-1", "{s}");
assert_eq!(s["scanned"], true, "{s}");
}
/// **ROBOT — no traced run: the static call sites stand in, and the pane
/// says so.** The same graph with no run in the warehouse must still flow
/// (sites as the rate) and name `static call sites` as the provenance.
#[test]
fn without_a_trace_the_sites_stand_in_and_are_named() {
let calls = serde_json::json!({
"state": "complete", "total_edges": 1,
"nodes": [
{ "id": "a::f", "moniker": "m:f" },
{ "id": "a::g", "moniker": "m:g" },
],
"edges": [ { "from": 0, "to": 1, "sites": 7 } ],
"evidence": { "missing": [], "notes": [] },
});
let no_runs = serde_json::json!({
"state": "complete", "lod": "trace_runs", "repo": "", "runs": [],
});
let room = CannedRoom::new(&[(S_CALLS, calls), (S_MAP, no_runs)]);
let mut p = CodeflowPane::new().with_room(room);
p.update_json(r#"{"scan":true}"#);
let s = p.state_json();
assert_eq!(s["nodes"], 2, "{s}");
assert_eq!(s["edges"], 1, "{s}");
assert_eq!(s["rated"], 1, "{s}");
assert_eq!(s["rated_by"], "static call sites", "{s}");
}
}