dwarves-gui/src/flow3d.rs
//! 🌐 **Flow3D** — the call graph as a LIVING 3D CLOUD: the 3D sibling of the
//! 🌊 Codeflow tab, on the shared [`facett_graph3d::Graph3D`] viewer (the
//! orbiting, depth-shaded node cloud) with its **edge-rate overlay** (feature
//! `rate`): every rated call edge is a smooth 3D arc with pulses travelling
//! along it, faster and brighter where the software's measured traffic runs
//! hotter.
//!
//! Same room, same answers, same provenance law as codeflow:
//!
//! * **Nodes + arcs** = the SELECTED workspace's call graph (`Viz.CallGraph`,
//! the Data view's wall answer; an edge's `from`/`to` index the reader's
//! node order).
//! * **The pulse ON an arc** = measured traffic. With a traced run in the
//! warehouse (`Viz.CodeMap` lod `trace_runs` / `trace`) each edge pulses at
//! ITS calls-per-second from that run; without one, the static call-site
//! count stands in — and the pane SAYS which it is showing (`rated_by`).
//!
//! The layout is [`Layout3D::Force`] — the crate's own "call graph look"
//! (deterministic 3D Fruchterman–Reingold: callers pull their callees close,
//! unrelated subsystems repel apart), the one placement where a call graph's
//! CLUSTERS are the picture. Group (the moniker's first segment) → colour.
//!
//! States that must never draw alike: un-scanned (reads itself on open),
//! asking, a refusal (the state + the reader's sentence, never an empty cloud
//! that reads as "no calls"), and the living cloud itself.
use std::sync::Arc;
use facett_core::Facet;
use facett_graph3d::{EdgeRates, Graph3D, Layout3D};
use crate::room::{self, Pending, RoomAnswer, RoomSource};
use crate::trace_replay;
/// The call-graph surface — dataview's, spelled once there (codeflow's law).
pub use crate::dataview::S_CALLS;
/// The replay surfaces (traced runs + one run's events) — codeflow's constant.
pub use crate::codeflow::S_MAP;
/// Edges asked of the room — the same cut codeflow asks for.
pub use crate::codeflow::EDGE_LIMIT;
/// Events of the rated run read in one page — the same window codeflow reads.
pub use crate::codeflow::TRACE_PAGE;
/// One scan's answers, landed together (the codeflow shape).
struct Flow3dRead {
calls: RoomAnswer,
/// The latest traced run and its event window, when the warehouse holds one.
trace: Option<(trace_replay::RunRow, RoomAnswer)>,
}
/// The 🌐 Flow3D tab.
pub struct Flow3dPane {
room: Option<Arc<dyn RoomSource>>,
graph: Option<Graph3D>,
/// What the pulse 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<Flow3dRead>>,
auto_read: bool,
}
impl Flow3dPane {
pub fn new() -> Self {
Self {
room: None,
graph: 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 cloud**: the call graph, the latest traced run, that run's
/// events — one worker, off the UI thread (codeflow's scan, verbatim).
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("flow3d-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)
});
Flow3dRead { 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 flow3d 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 flow3d read worker died before it answered".into(),
}
}
}
fn land(&mut self, read: Flow3dRead) {
let a = &read.calls;
if !a.is_measured() {
// A refusal draws its state, never an empty cloud that reads as "no calls".
self.graph = 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.
// id = moniker-or-label, DEDUPED (codeflow's law): the trace's stable
// ids land on these, so a duplicate moniker must not swallow a node.
let rows = a.rows("nodes");
let mut nodes: Vec<(String, egui::Color32)> = 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);
// group → colour: one stable hue per crate/module, the house rule.
nodes.push((label, facett_core::hash_color(&group)));
}
if nodes.is_empty() {
self.graph = None;
self.status = "the room's call graph names no node".into();
return;
}
let mut edges: Vec<(usize, usize)> = 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;
};
if f >= id_of.len() || t >= id_of.len() {
continue;
}
edges.push((f, t));
sites.push(e["sites"].as_u64().unwrap_or(0) as f32);
}
let (n, m) = (nodes.len(), edges.len());
// The arcs' rates: the traced run's measured calls/s when its stable ids
// land on this graph's nodes; the static call sites stand in otherwise —
// and `rated_by` SAYS which, never implies (codeflow's provenance law).
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>> = edges
.iter()
.map(|&(f, t)| {
count
.get(&(id_of[f].as_str(), id_of[t].as_str()))
.map(|&c| c 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 mut graph = Graph3D::new(nodes, edges, Layout3D::Force).with_title("flow3d");
// The viewer carries the table; its rate overlay pulses from it + the
// injected clock the pane ticks every frame.
let _ = graph.set_rates(EdgeRates { edges: rates });
self.graph = Some(graph);
self.status = format!("{n} node(s) · {m} arc(s) · pulse = {}", self.rated_by);
}
}
impl Default for Flow3dPane {
fn default() -> Self {
Self::new()
}
}
impl Facet for Flow3dPane {
fn title(&self) -> &str {
"Flow3D"
}
fn ui(&mut self, ui: &mut egui::Ui) {
self.poll();
if !self.auto_read && self.pending.is_none() && self.graph.is_none() {
self.auto_read = true;
self.scan();
}
ui.horizontal_wrapped(|ui| {
ui.heading("🌐 Flow3D");
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(graph) = &mut self.graph {
// The host clock: the injected FC-7 clock ticks every frame (the
// codeflow stable_dt clamp), so the spin AND the pulse phase are
// pure functions of it — continuously smooth, never stepped.
let dt = ui.input(|i| i.stable_dt).clamp(0.0, 0.1);
let _ = graph.update(facett_graph3d::Msg::Tick(f64::from(dt)));
ui.ctx().request_repaint();
graph.ui(ui);
}
}
fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "flow3d",
"scanned": self.graph.is_some(),
"status": self.status,
"nodes": self.graph.as_ref().map(|g| g.nodes.len()).unwrap_or(0),
"edges": self.graph.as_ref().map(|g| g.edges.len()).unwrap_or(0),
"rated": self.rated,
"rated_by": self.rated_by,
"layout": self.graph.as_ref().map(|g| g.state().layout.label()).unwrap_or("force"),
"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 🌐 Flow3D tab, bound to the SELECTED workspace.
pub fn flow3d_pane(endpoint: &str, token: &str, workspace: &str) -> Flow3dPane {
Flow3dPane::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
/// codeflow test double, verbatim: 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 3D arcs 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 over 2 s; the
/// pane's cloud must hold that graph on the force layout, rate the edges
/// from the TRACE (checkout→charge 1.0/s, charge→audit 0.5/s — the run's
/// calls over its duration, not the static `sites`), and say which run
/// rated it.
#[test]
fn the_cloud_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 = Flow3dPane::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 cloud's graph: {s}");
assert_eq!(s["edges"], 2, "{s}");
assert_eq!(s["layout"], "force", "the call-graph layout is the force cloud: {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}");
// The rates LANDED ON the viewer (not just the pane's counters): the
// overlay's table holds the run's calls/s per edge, in edge order.
let g = p.graph.as_ref().expect("the cloud landed");
let rates = g.rates().expect("the viewer carries the rate table");
assert_eq!(rates.edges.len(), 2, "one slot per edge");
assert_eq!(rates.edges[0], Some(1.0), "checkout→charge: 2 calls / 2 s");
assert_eq!(rates.edges[1], Some(0.5), "charge→audit: 1 call / 2 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 pulse
/// (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 = Flow3dPane::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}");
let g = p.graph.as_ref().expect("the cloud landed");
assert_eq!(
g.rates().expect("sites stand in as the rate table").edges[0],
Some(7.0),
"the static site count is the stand-in rate"
);
}
}