nordisk/dwarves
crates/nornir-funnel-embed/src/lib.rs
//! **nornir-funnel-embed** — the LIVE funnel as an embeddable `Panel` (backend (2)).
//!
//! The reusable `<guest>-embed` adapter shape from
//! `.nornir/wasm-ui-panels-design.md` §3 (holger-embed is the template): a **thin
//! tonic gRPC client** fetches the funnel from a running `nornir-server`'s read-only
//! [`Funnel.Show`](pb) RPC, **reconstructs** a [`funnel_core::state::Funnel`]
//! client-side (mirroring nornir's `viz::remote::funnel_reconstruct`), projects it
//! with the pure [`funnel_core::plan_dag::PlanDag`], and renders it with the shared
//! [`facett_plandag::PlanDagView`] (the metric-bearing Plan-DAG facet).
//!
//! **No `nornir` dependency.** nornir already depends on dwarves, so a `dwarves →
//! nornir` edge would be a cycle. Instead this crate compiles ONLY nornir's funnel
//! `Funnel.Show` proto (a trimmed copy under `proto/`, package `nornir.v1`, so the
//! gRPC method path `/nornir.v1.Funnel/Show` matches the live server) and reconstructs
//! against `funnel-core` + the generated `pb` types directly.
//!
//! ## The seam
//! [`panel`] returns a `Box<dyn Panel>` a host drops into a tab. It connects to the
//! given endpoint (default `http://127.0.0.1:7878`), binds to the **selected
//! warehouse** ([`FunnelPanel::set_workspace`], mirroring nornir's per-tab
//! `set_workspace`), and on any failure shows a clean, DISTINCT state — **no warehouse
//! selected** / **server unreachable** / **auth failed** / **warehouse not served** —
//! never the old catch-all *"offline"* (which lied when the server was up). It never
//! panics. A **Refresh** action re-fetches.
use std::time::Duration;
/// **dwarves's REACTION to a facett UI error code** (facett error-codes, Phase 2) —
/// namespaces a `facet-<component>-<n>` with dwarves' mount
/// (`dwarves/funnel:facet-graphview-1`) and returns the canonical reaction.
/// COMPOSES with `dwarves-<subsystem>-<n>`, never collides.
pub mod facet_react;
/// The funnel DAG rendered as a node-link graph on the shared `facett-graphview`
/// engine — the second render beside the `facett-plandag` card lane.
pub mod graph;
pub use graph::FunnelGraphPane;
/// The funnel DAG rendered as a **3D layered graph** — the third render (a camera-only
/// 3D placement with optional auto-rotate + clickable drill), reusing the world
/// graph's orthographic 3D projection pattern. See [`graph3d`].
pub mod graph3d;
pub use graph3d::{graph3d_build_calls, FunnelGraph3d};
use anyhow::{Context, Result};
use facett_core::{Facet, Panel};
use facett_plandag::{Metrics, PlanDagView, PlanEdge, PlanModel, PlanNode, PlanVersion};
use funnel_core::plan_dag::{DagStatus, PlanDag};
/// The generated thin client + wire types for nornir's `Funnel.Show` (+ the server
/// binding the `testkit` mock mounts). PUBLIC (like `nornir_warehouse_embed::pb`) so
/// the testkit's seed shapes are nameable by a host's robot tests, and so a future
/// dwarves-side `Funnel` proxy can serve these exact types.
pub mod pb {
tonic::include_proto!("nornir.v1");
}
/// The default `nornir-server` endpoint on this box ("dwarf").
pub const DEFAULT_ENDPOINT: &str = "http://127.0.0.1:7878";
// ── wire → funnel-core reconstruction (mirrors nornir viz::remote) ────────────
/// Lenient wire→[`funnel_core::event::NodeStatus`] parse. Accepts the server's
/// snake_case serde token, the `Debug` repr (lower-cased), and the CLI synonyms
/// (`active` ⇒ in-progress, `abandoned` ⇒ failed). Unknown / `pending` ⇒ `Pending`
/// (the safe default — an unrecognised status never panics).
fn node_status_from_wire(s: &str) -> funnel_core::event::NodeStatus {
use funnel_core::event::NodeStatus::*;
match s.trim().to_ascii_lowercase().as_str() {
"ready" => Ready,
"inprogress" | "in_progress" | "active" => InProgress,
"done" => Done,
"blocked" => Blocked,
"failed" | "abandoned" => Failed,
_ => Pending,
}
}
/// Lenient wire→[`funnel_core::event::PlanStatus`] parse (the dump carries the
/// `Debug` repr). Unknown ⇒ `Draft`.
fn plan_status_from_wire(s: &str) -> funnel_core::event::PlanStatus {
use funnel_core::event::PlanStatus::*;
match s.trim().to_ascii_lowercase().as_str() {
"active" => Active,
"done" => Done,
"abandoned" => Abandoned,
_ => Draft,
}
}
/// **Reconstruct a [`funnel_core::state::Funnel`] from a `Funnel.Show` dump.**
///
/// Pure (no network) so it is unit-testable: maps the render-ready `FunnelDump`
/// (ideas → plans → nodes, each node carrying its `deps`) back onto the in-memory
/// `Funnel` the pure DAG projections consume. Re-implemented against `funnel-core`
/// and the generated `pb` types (NOT by importing nornir) — a faithful twin of
/// nornir's `viz::remote::funnel_from_dump`.
///
/// Fidelity notes (additive, never a failure): the dump carries no timestamps, so
/// every card reconstructs with `now` (staleness collapses to a uniform 0 — the
/// topology, statuses, and edges are exact); `refs`/assignments/lanes are absent
/// from the wire and default empty.
pub fn funnel_from_dump(dump: &pb::FunnelDump) -> funnel_core::state::Funnel {
use funnel_core::event::ItemKind;
use funnel_core::ids::{IdeaId, NodeId, PlanId};
use funnel_core::state::{Funnel, Idea, Plan, PlanNode as CorePlanNode};
use std::collections::{BTreeMap, BTreeSet};
let now = chrono::Utc::now();
let mut f = Funnel::default();
for idea in &dump.ideas {
let iid = IdeaId::new(&idea.id);
f.ideas.insert(
iid.clone(),
Idea {
id: iid.clone(),
source: idea.source.clone(),
text: idea.text.clone(),
refs: Vec::new(),
item_kind: ItemKind::parse(&idea.item_kind),
submitted_at: now,
triage: None,
mode: None,
},
);
for plan in &idea.plans {
let pid = PlanId::new(&plan.id);
// Prefer the dump's explicit idea_id, fall back to the enclosing idea.
let plan_idea = if plan.idea_id.is_empty() {
iid.clone()
} else {
IdeaId::new(&plan.idea_id)
};
let mut nodes: BTreeMap<NodeId, CorePlanNode> = BTreeMap::new();
for n in &plan.nodes {
let nid = NodeId::new(&n.id);
let mut params = serde_json::Map::new();
if !n.title.is_empty() {
params.insert("title".into(), serde_json::json!(n.title));
}
nodes.insert(
nid.clone(),
CorePlanNode {
id: nid.clone(),
kind: n.kind.clone(),
params,
targets: Vec::new(),
prompt_excerpt: (!n.title.is_empty()).then(|| n.title.clone()),
created_at: now,
status: node_status_from_wire(&n.status),
last_status_change: now,
},
);
}
// Edges: each node lists its upstream `deps` (dep → node). Skip any edge
// whose endpoints aren't both present (defensive — the pure projections
// tolerate a partial dump, never panic).
let mut edges: BTreeSet<(NodeId, NodeId)> = BTreeSet::new();
let present: BTreeSet<&str> = plan.nodes.iter().map(|n| n.id.as_str()).collect();
for n in &plan.nodes {
let to = NodeId::new(&n.id);
for d in &n.deps {
if present.contains(d.as_str()) && d != &n.id {
edges.insert((NodeId::new(d), to.clone()));
}
}
}
f.plans.insert(
pid.clone(),
Plan {
id: pid,
idea_id: plan_idea,
summary: plan.summary.clone(),
planner: "server".into(),
created_at: now,
status: plan_status_from_wire(&plan.status),
nodes,
edges,
},
);
}
}
f
}
// ── thin gRPC fetch ───────────────────────────────────────────────────────────
/// **The bound on a READ, which `connect_timeout` is not.** A server that ACCEPTS the
/// socket and then never answers clears the connect bound instantly and then waits
/// forever — the shape that left `dwarves-ui-gpu` with no window at all (measured
/// against the live oden forge, 2026-09-06). Pairing the connect bound with a request
/// bound is what makes an unresponsive host degrade to an honest unavailable state.
const READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
/// Normalize an endpoint: prefix `http://` if the scheme is missing.
fn norm_endpoint(endpoint: &str) -> String {
if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
}
}
/// The ONE auth interceptor every thin funnel wire call shares: dwarves' bearer as
/// `authorization` plus the selected workspace as `nornir-workspace` metadata.
/// A concrete type (not a closure) so [`dial`] can NAME the intercepted client and
/// the three wire legs ([`fetch_funnel`] / [`submit_idea`] / [`persist_plan`]) share
/// one connect body instead of three twins (L5).
#[derive(Clone)]
struct FunnelAuth {
bearer: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
workspace: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
}
impl tonic::service::Interceptor for FunnelAuth {
fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
if let Some(b) = &self.bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(ws) = &self.workspace {
req.metadata_mut().insert("nornir-workspace", ws.clone());
}
Ok(req)
}
}
/// The connected thin client [`dial`] hands back (channel + [`FunnelAuth`]).
type WireClient = pb::funnel_client::FunnelClient<
tonic::service::interceptor::InterceptedService<tonic::transport::Channel, FunnelAuth>,
>;
/// Connect to `nornir-server` at `endpoint` and return the authenticated thin
/// `Funnel` client — the ONE connect+bearer+workspace body all three wire legs
/// share. 3s connect timeout so a dead server never wedges the caller; the
/// "connect to nornir-server" context is what `classify_conn` keys `Unreachable` on.
async fn dial(endpoint: &str, token: &str, workspace: &str) -> Result<WireClient> {
// The ONE shared client bearer-builder (Law #1) — no per-site
// `format!("Bearer …").parse()` twin. `None` ⇒ open (empty token).
let bearer = nornir_service::client::bearer(token).context("parse bearer token")?;
let workspace = (!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten();
let endpoint = norm_endpoint(endpoint);
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
Ok(
pb::funnel_client::FunnelClient::with_interceptor(
channel,
FunnelAuth { bearer, workspace },
)
.max_decoding_message_size(64 << 20),
)
}
/// Drive a wire future to completion on a private current-thread runtime — safe to
/// call from the synchronous egui loop (shared by the three wire legs, L5).
fn block_on<T>(what: &str, fut: impl std::future::Future<Output = Result<T>>) -> Result<T> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.with_context(|| format!("build tokio runtime for {what}"))?;
rt.block_on(fut)
}
/// Fetch the funnel from `nornir-server` at `endpoint` over the thin `Funnel.Show`
/// gRPC and reconstruct a [`funnel_core::state::Funnel`]. Authenticates with
/// `Bearer <token>` (empty ⇒ no real auth); if `workspace` is non-empty it rides in
/// the `nornir-workspace` metadata header (selects which served workspace's funnel).
/// Readiness is derived (`promote_ready`) so the DAG shows the same Pending→Ready
/// frontier the local path does. Runs on a private current-thread runtime so it is
/// safe to call from the synchronous egui loop; a 3s connect timeout keeps a dead
/// server from wedging the UI thread.
pub fn fetch_funnel(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<funnel_core::state::Funnel> {
block_on("funnel client", async move {
let mut client = dial(endpoint, token, workspace).await?;
let dump = client
.show(pb::Empty {})
.await
.context("Funnel.Show RPC")?
.into_inner();
let mut f = funnel_from_dump(&dump);
f.promote_ready();
Ok(f)
})
}
/// The intake kind the submit strip carries: an **idea** (a thing to build) or an
/// **error** (a defect to triage). Both are first-class funnel items — the server's
/// `SubmitIdeaRequest.item_kind` routes them (`funnel-intake`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum IntakeKind {
Idea,
Error,
}
impl IntakeKind {
/// The wire token the server expects (`""` ⇒ idea; we send it explicitly).
fn wire(self) -> &'static str {
match self {
IntakeKind::Idea => "idea",
IntakeKind::Error => "error",
}
}
}
/// **Submit an idea/error into nornir's funnel over thin gRPC (the WRITE leg).**
///
/// Calls `Funnel.SubmitIdea` on `nornir-server` at `endpoint`, returning the new
/// item's id. Same auth/workspace metadata + private current-thread runtime + 3s
/// connect timeout as [`fetch_funnel`], so it is safe to call from the synchronous
/// egui loop and never wedges the UI thread. Typing an idea here grows the server's
/// plan-DAG, which the pane's auto-poll `Show` then re-renders — the funnel's
/// read→write round-trip, all over gRPC (no `nornir` crate dependency).
pub fn submit_idea(
endpoint: &str,
token: &str,
workspace: &str,
text: &str,
kind: IntakeKind,
) -> Result<String> {
block_on("funnel submit", async move {
let mut client = dial(endpoint, token, workspace).await?;
let id = client
.submit_idea(pb::SubmitIdeaRequest {
text: text.to_string(),
source: "dwarves:funnel-pane".into(),
item_kind: kind.wire().into(),
})
.await
.context("Funnel.SubmitIdea RPC")?
.into_inner()
.id;
Ok(id)
})
}
// ── the decompose PERSIST leg (plan → CreatePlan + AddNode + Link) ────────────
/// One node of a plan to PERSIST — the neutral spec shape a host (dwarves'
/// `funnel_action::FunnelPlan`) lowers its plan into. `local_id` is the plan's OWN
/// node id (e.g. `"n1"`); `needs` references sibling `local_id`s. The server
/// assigns its own sequential node ids, so [`persist_plan`] keeps the local→server
/// map and translates the edges.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PlanNodeSpec {
/// The plan's own node id (referenced by sibling `needs`).
pub local_id: String,
/// Free verb/kind, e.g. `"code:write"`.
pub kind: String,
/// Human-readable title (what `Funnel.Show` re-renders).
pub title: String,
/// `local_id`s of sibling nodes this node depends on.
pub needs: Vec<String>,
}
/// The server ids a persisted plan landed under — returned so the caller can echo
/// them (and a test can follow them back out of `Funnel.Show`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PersistedPlan {
/// The idea the plan attached to (given, or newly submitted).
pub idea_id: String,
/// The server-assigned plan id (e.g. `"p-003"`).
pub plan_id: String,
/// `local_id → server node id`, in submission order.
pub node_ids: Vec<(String, String)>,
}
/// **Persist a decomposed plan into nornir's funnel (the "funnel-no-plan" fix).**
///
/// Drives the SAME event-sourced write RPCs the nornir CLI uses — `CreatePlan`
/// (⇒ `PlanCreated` + `PlanStatusChanged(Active)` events), `AddNode` (⇒ `NodeAdded`)
/// and `Link` (⇒ `EdgeAdded`) — so the persisted plan reappears in `Funnel.Show`
/// exactly like every other plan; no new server surface, no side store.
///
/// `idea_id: Some(..)` attaches the plan to an EXISTING funnel idea; `None` first
/// submits `idea_text` (source `"dwarves:decompose"`) — nornir's store rejects a
/// plan on an unknown idea, and a plan whose idea is missing would be invisible in
/// `Show`. The plan spec is VALIDATED before the first write (empty plan / dangling
/// `needs` / duplicate `local_id` ⇒ an honest error, no partial persist from a bad
/// spec). Same auth/workspace + timeout posture as [`fetch_funnel`]/[`submit_idea`].
pub fn persist_plan(
endpoint: &str,
token: &str,
workspace: &str,
idea_id: Option<&str>,
idea_text: &str,
summary: &str,
nodes: &[PlanNodeSpec],
) -> Result<PersistedPlan> {
// Validate BEFORE any write — a malformed spec must not half-persist.
if nodes.is_empty() {
anyhow::bail!("refusing to persist an empty plan (no nodes)");
}
let locals: std::collections::BTreeSet<&str> =
nodes.iter().map(|n| n.local_id.as_str()).collect();
if locals.len() != nodes.len() {
anyhow::bail!("plan has duplicate node ids");
}
for n in nodes {
for d in &n.needs {
if !locals.contains(d.as_str()) {
anyhow::bail!("plan node `{}` needs unknown sibling `{d}`", n.local_id);
}
}
}
block_on("funnel persist", async move {
let mut client = dial(endpoint, token, workspace).await?;
let idea_id = match idea_id {
Some(id) if !id.trim().is_empty() => id.to_string(),
_ => {
client
.submit_idea(pb::SubmitIdeaRequest {
text: idea_text.to_string(),
source: "dwarves:decompose".into(),
item_kind: IntakeKind::Idea.wire().into(),
})
.await
.context("Funnel.SubmitIdea RPC (persist: intake the idea)")?
.into_inner()
.id
}
};
let plan_id = client
.create_plan(pb::CreatePlanRequest {
idea_id: idea_id.clone(),
summary: summary.to_string(),
})
.await
.context("Funnel.CreatePlan RPC")?
.into_inner()
.id;
// Pass 1 — nodes (needs left empty: the server maps `needs` onto edges at
// add time, but a spec may reference a sibling not yet added, so ALL edges
// go through `Link` below — order-independent, same `EdgeAdded` event).
let mut server_id: std::collections::BTreeMap<&str, String> =
std::collections::BTreeMap::new();
let mut node_ids = Vec::with_capacity(nodes.len());
for n in nodes {
let id = client
.add_node(pb::AddNodeRequest {
plan_id: plan_id.clone(),
kind: n.kind.clone(),
title: n.title.clone(),
prompt: String::new(),
targets: Vec::new(),
needs: Vec::new(),
})
.await
.with_context(|| format!("Funnel.AddNode RPC (node `{}`)", n.local_id))?
.into_inner()
.id;
server_id.insert(n.local_id.as_str(), id.clone());
node_ids.push((n.local_id.clone(), id));
}
// Pass 2 — edges (dep → node), through the local→server id map.
for n in nodes {
for d in &n.needs {
client
.link(pb::LinkRequest {
plan_id: plan_id.clone(),
from: server_id[d.as_str()].clone(),
to: server_id[n.local_id.as_str()].clone(),
})
.await
.with_context(|| format!("Funnel.Link RPC (edge `{d}` → `{}`)", n.local_id))?;
}
}
Ok(PersistedPlan {
idea_id,
plan_id,
node_ids,
})
})
}
// ── the EMBEDDED pane's LIVE-source seam (host injection) ─────────────────────
/// **The embedded funnel pane's LIVE-source injection seam** — the same host-injection
/// pattern as the Warehouse Navigator's `RealFnSource`: the pane stays FAT/embedded
/// (offline-first, immediate — construction never dials), and a HOST that owns an
/// endpoint injects a source that is consulted LAZILY on each refresh/poll. A healthy
/// consult REPLACES the rendered funnel with the server's real, `funnel_events`-backed
/// state (provenance `live: …` in `state_json.funnel_source`); a dead/refused/denied
/// consult KEEPS the embedded state and NAMES the outage with the SAME distinct
/// failure vocabulary the thin pane uses (`state_json.live_status`:
/// unreachable / auth_failed / not_served / server_error / no_warehouse) — never
/// fabricated, never a silent fallback.
pub trait LiveFunnelSource: Send + Sync {
/// Fetch the real funnel for `workspace`. The error chain is classified by the
/// SAME [`classify_conn`](crate) rules the thin pane uses (a `tonic::Status` in
/// the chain names auth vs not-served vs server error; a connect failure is
/// `unreachable`).
fn fetch(&self, workspace: &str) -> Result<funnel_core::state::Funnel>;
/// Submit an idea/error into the live funnel (the WRITE leg), returning the new
/// item's id. Only consulted while the live leg is UP — an outage falls back to
/// the in-process engine's offline submit.
fn submit(&self, workspace: &str, text: &str, kind: IntakeKind) -> Result<String>;
/// Where the live funnel comes from — rides verbatim in `funnel_source`.
fn describe(&self) -> String;
}
/// **The SHIPPED live source**: nornir's `Funnel.Show` / `Funnel.SubmitIdea` over the
/// existing thin gRPC pair ([`fetch_funnel`] / [`submit_idea`]) — pure reuse of the
/// one wire path (the same `submit_idea` the dwarves MCP `funnel_submit_idea` tool
/// calls), never a second client.
pub struct GrpcFunnelSource {
endpoint: String,
token: String,
}
impl GrpcFunnelSource {
pub fn new(endpoint: impl Into<String>, token: impl Into<String>) -> Self {
Self {
endpoint: endpoint.into(),
token: token.into(),
}
}
}
impl LiveFunnelSource for GrpcFunnelSource {
fn fetch(&self, workspace: &str) -> Result<funnel_core::state::Funnel> {
fetch_funnel(&self.endpoint, &self.token, workspace)
}
fn submit(&self, workspace: &str, text: &str, kind: IntakeKind) -> Result<String> {
submit_idea(&self.endpoint, &self.token, workspace, text, kind)
}
fn describe(&self) -> String {
format!("nornir funnel @ {}", self.endpoint)
}
}
// ── the FAT / EMBEDDED funnel engine (the fat-mode default) ───────────────────
/// **The in-process (FAT / embedded) funnel** — the fat-mode default so dwarves
/// works self-contained with **`nornir-server` DOWN**. It maintains a
/// [`funnel_core::state::Funnel`] mutated purely in-process through funnel-core's
/// OWN event [`apply`](funnel_core::state::Funnel::apply) — the SAME engine
/// `nornir`'s funnel `Store` replays, so this is genuine reuse, not a twin — and
/// renders + grows a real Plan-DAG with **no network at all**.
///
/// The funnel is seeded over the LOCAL workspace, never the wire:
/// * if `$NORNIR_FUNNEL_LOG` points at a JSON-lines log of
/// [`funnel_core::event::Event`], it is replayed (a real local funnel);
/// * otherwise a small `nordisk` demo funnel is seeded so the embedded DAG renders
/// non-empty — mirroring the release pane's endorsed "nordisk fixture" fallback.
///
/// Submitting an idea/error [`applies`](EmbeddedFunnel::submit) real funnel-core
/// events in-process, so the DAG grows offline exactly like the live one does.
pub struct EmbeddedFunnel {
funnel: funnel_core::state::Funnel,
/// The workspace the embedded funnel is scoped to (for display/state only).
workspace: String,
}
impl EmbeddedFunnel {
/// Build the embedded funnel for `workspace` — replaying a local event log if
/// one is configured, else seeding the `nordisk` demo funnel. Never touches the
/// network; never panics.
pub fn new(workspace: impl Into<String>) -> Self {
let workspace = workspace.into();
let mut funnel = funnel_core::state::Funnel::default();
if let Some(path) = local_funnel_log() {
if replay_log(&mut funnel, &path).is_ok() && !funnel.ideas.is_empty() {
funnel.promote_ready();
return Self { funnel, workspace };
}
// A partial/failed replay must not leave a half-built funnel — reset and
// fall through to the self-contained seed.
funnel = funnel_core::state::Funnel::default();
}
// NO seed: with no local log the embedded funnel is EMPTY. It used to be filled
// with a sample idea and its plan, which a person looking at a pane whose server
// was down could not tell from their own funnel.
Self { funnel, workspace }
}
/// The embedded funnel over the SAMPLE content ([`seed_nordisk_funnel`]) — tests
/// only (`testkit`): no shipped build contains it.
#[cfg(any(test, feature = "testkit"))]
pub fn seeded(workspace: impl Into<String>) -> Self {
let mut funnel = funnel_core::state::Funnel::default();
seed_nordisk_funnel(&mut funnel);
funnel.promote_ready();
Self {
funnel,
workspace: workspace.into(),
}
}
/// The workspace the embedded funnel is scoped to.
pub fn workspace(&self) -> &str {
&self.workspace
}
/// The current in-process funnel (readiness promoted), for rendering.
fn funnel(&self) -> funnel_core::state::Funnel {
let mut f = self.funnel.clone();
f.promote_ready();
f
}
/// **The EVENT-SOURCED run history** — every `RunRecorded` folded into
/// `funnel_core::Funnel::runs`, newest first, as the wire shape the pane
/// renders.
///
/// This is the funnel's OWN history, projected by funnel-core, not a
/// second store kept alongside it: the state engine is the one source, so a
/// replayed log (`$NORNIR_FUNNEL_LOG`) yields exactly the history the engine
/// folded. Distinct from `FunnelRunLog`, which snapshots the *rendered DAG*
/// per poll for the run PICKER (a view concern, and the only history a remote
/// `Funnel.Show` can support — that RPC returns the current funnel with no
/// run ledger on the wire).
pub fn runs_json(&self) -> serde_json::Value {
self.funnel.runs_to_json()
}
/// How many runs the event log recorded.
pub fn run_count(&self) -> usize {
self.funnel.runs().len()
}
/// **Submit an idea/error into the in-process funnel (the offline WRITE leg).**
/// Applies real funnel-core events (idea → one-node plan, node kind from
/// [`ItemKind::node_kind`](funnel_core::event::ItemKind::node_kind)) so the DAG
/// grows locally, and returns the new idea id. No server involved.
fn submit(&mut self, text: &str, kind: IntakeKind) -> Result<String> {
use funnel_core::event::{Event, ItemKind};
use funnel_core::ids::{IdeaId, NodeId, PlanId};
let item_kind = match kind {
IntakeKind::Idea => ItemKind::Idea,
IntakeKind::Error => ItemKind::Error,
};
let ts = chrono::Utc::now();
let iid = IdeaId::new(format!("i-{:03}", self.funnel.next_idea));
self.funnel.apply(&Event::IdeaSubmitted {
id: iid.clone(),
source: "dwarves:funnel-pane(embedded)".into(),
text: text.to_string(),
refs: Vec::new(),
item_kind,
ts,
})?;
let pid = PlanId::new(format!("p-{:03}", self.funnel.next_plan));
self.funnel.apply(&Event::PlanCreated {
id: pid.clone(),
idea_id: iid.clone(),
summary: text.chars().take(60).collect::<String>(),
planner: "embedded".into(),
ts,
})?;
let nid = NodeId::new(format!("n-{:03}", self.funnel.next_node));
self.funnel.apply(&Event::NodeAdded {
plan_id: pid,
node_id: nid,
kind: item_kind.node_kind().to_string(),
params: serde_json::Map::new(),
targets: Vec::new(),
prompt_excerpt: None,
ts,
})?;
self.funnel.promote_ready();
Ok(iid.as_str().to_string())
}
}
/// The optional local funnel event-log path (`$NORNIR_FUNNEL_LOG`), when it names an
/// existing file. Kept env-only so the embedded default pulls in no heavy warehouse
/// deps — a real local funnel is opt-in, the `nordisk` seed is the zero-config default.
fn local_funnel_log() -> Option<std::path::PathBuf> {
let p = std::env::var_os("NORNIR_FUNNEL_LOG")?;
let p = std::path::PathBuf::from(p);
p.is_file().then_some(p)
}
/// Replay a JSON-lines log of [`funnel_core::event::Event`] into `funnel` (one event
/// per line; blank lines skipped; a line that fails to parse/apply is skipped rather
/// than aborting the whole replay — defensive, exactly like nornir's loader).
fn replay_log(funnel: &mut funnel_core::state::Funnel, path: &std::path::Path) -> Result<()> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("read funnel log {}", path.display()))?;
for line in text.lines() {
let line = line.trim();
if line.is_empty() {
continue;
}
if let Ok(ev) = serde_json::from_str::<funnel_core::event::Event>(line) {
let _ = funnel.apply(&ev);
}
}
Ok(())
}
/// A small SAMPLE funnel for tests (`testkit`): one feature idea decomposed into a
/// write→fix→test chain, plus a triaged error item — built from the SAME funnel-core
/// events the live engine records. Not compiled into a shipped build: an embedded pane
/// with no local log is empty.
#[cfg(any(test, feature = "testkit"))]
fn seed_nordisk_funnel(funnel: &mut funnel_core::state::Funnel) {
use funnel_core::event::{Event, ItemKind, NodeStatus};
use funnel_core::ids::{IdeaId, NodeId, PlanId};
let ts = chrono::Utc::now();
let mut apply = |ev: Event| {
let _ = funnel.apply(&ev);
};
// ── idea: the release engine — decomposed into a write → fix → test chain ──
let i1 = IdeaId::new("i-001");
apply(Event::IdeaSubmitted {
id: i1.clone(),
source: "human:alice".into(),
text: "world-class release engine".into(),
refs: Vec::new(),
item_kind: ItemKind::Idea,
ts,
});
let p1 = PlanId::new("p-001");
apply(Event::PlanCreated {
id: p1.clone(),
idea_id: i1,
summary: "decompose the release engine".into(),
planner: "embedded".into(),
ts,
});
let n_write = NodeId::new("n-001");
let n_fix = NodeId::new("n-002");
let n_test = NodeId::new("n-003");
for (nid, kind) in [
(&n_write, "code:write"),
(&n_fix, "code:fix"),
(&n_test, "test:write"),
] {
apply(Event::NodeAdded {
plan_id: p1.clone(),
node_id: nid.clone(),
kind: kind.into(),
params: serde_json::Map::new(),
targets: Vec::new(),
prompt_excerpt: None,
ts,
});
}
apply(Event::EdgeAdded {
plan_id: p1.clone(),
from_node: n_write.clone(),
to_node: n_fix.clone(),
ts,
});
apply(Event::EdgeAdded {
plan_id: p1.clone(),
from_node: n_fix.clone(),
to_node: n_test.clone(),
ts,
});
apply(Event::NodeStatusChanged {
plan_id: p1.clone(),
node_id: n_write.clone(),
status: NodeStatus::Done,
why: None,
ts,
});
apply(Event::NodeStatusChanged {
plan_id: p1.clone(),
node_id: n_fix.clone(),
status: NodeStatus::InProgress,
why: None,
ts,
});
// Real execution history, as EVENTS — the write node ran once and passed, the
// fix node ran twice (a failure, then a retry still in flight). These fold into
// `funnel_core::Funnel::runs`, which is what the pane's run-history strip reads;
// the seed carries them so the offline demo shows a non-empty history without a
// server (the same events a live engine records — no fixture-only shape).
use funnel_core::event::RunOutcome as CoreRunOutcome;
use funnel_core::ids::RunId;
apply(Event::RunRecorded {
plan_id: p1.clone(),
node_id: n_write.clone(),
run_id: RunId::new("r-001"),
ran_by: "agent:brokkr".into(),
outcome: CoreRunOutcome::Ok,
log_ref: None,
produced_commits: Vec::new(),
produced_test_runs: Vec::new(),
ts,
});
apply(Event::RunRecorded {
plan_id: p1.clone(),
node_id: n_fix.clone(),
run_id: RunId::new("r-002"),
ran_by: "agent:brokkr".into(),
outcome: CoreRunOutcome::Failed,
log_ref: None,
produced_commits: Vec::new(),
produced_test_runs: Vec::new(),
ts,
});
apply(Event::RunRecorded {
plan_id: p1,
node_id: n_fix,
run_id: RunId::new("r-003"),
ran_by: "human:alice".into(),
outcome: CoreRunOutcome::Aborted,
log_ref: None,
produced_commits: Vec::new(),
produced_test_runs: Vec::new(),
ts,
});
// ── error item: a first-class funnel entry, triaged into a fix node ──
let i2 = IdeaId::new("i-002");
apply(Event::IdeaSubmitted {
id: i2.clone(),
source: "human:alice".into(),
text: "funnel defaulted to remote nornir".into(),
refs: Vec::new(),
item_kind: ItemKind::Error,
ts,
});
let p2 = PlanId::new("p-002");
apply(Event::PlanCreated {
id: p2.clone(),
idea_id: i2,
summary: "make the funnel fat-by-default".into(),
planner: "embedded".into(),
ts,
});
apply(Event::NodeAdded {
plan_id: p2,
node_id: NodeId::new("n-004"),
kind: ItemKind::Error.node_kind().into(),
params: serde_json::Map::new(),
targets: Vec::new(),
prompt_excerpt: None,
ts,
});
}
// ── funnel-core PlanDag → facett-plandag model (mirrors nornir funnel_plandag) ─
/// Map the neutral DAG status onto the plandag's operator status.
fn op_status(s: DagStatus) -> facett_plandag::OpStatus {
use facett_plandag::OpStatus;
match s {
DagStatus::Idle => OpStatus::Idle,
DagStatus::Running => OpStatus::Running,
DagStatus::Done => OpStatus::Done,
DagStatus::Failed => OpStatus::Failed,
}
}
/// Build a [`PlanDagView`] from an assembled [`PlanDag`]. The three neutral node
/// scalars map onto the plandag's Time/Memory/Rows cost-heat: **staleness → Time**,
/// **weight → Memory**, **fan-out → Rows** (same mapping as nornir's funnel_plandag).
fn view_from_dag(dag: &PlanDag) -> PlanDagView {
let idx: std::collections::HashMap<&str, usize> = dag
.nodes
.iter()
.enumerate()
.map(|(i, n)| (n.id.as_str(), i))
.collect();
let nodes: Vec<PlanNode> = dag
.nodes
.iter()
.map(|n| {
let mut pn = PlanNode::new(n.id.clone(), n.title.clone())
.status(op_status(n.status))
.metrics(Metrics {
time_ms: n.staleness_hours,
peak_mem: n.weight_bytes,
rows: n.fanout,
spill: 0,
});
if let Some(g) = &n.group {
pn = pn.group(g.clone());
}
if n.is_bug {
pn = pn.photon();
}
pn
})
.collect();
let edges: Vec<PlanEdge> = dag
.edges
.iter()
.filter_map(|e| {
let (Some(&from), Some(&to)) = (idx.get(e.from.as_str()), idx.get(e.to.as_str()))
else {
return None;
};
Some(PlanEdge { from, to })
})
.collect();
let version = PlanVersion::new("funnel", nodes, edges);
view_from_model(PlanModel::new(version))
}
/// Build the funnel's [`PlanDagView`] from an already-assembled [`PlanModel`] — the
/// one place the pane names the view, so the live path and a re-rendered past-run
/// snapshot are byte-identical (same title, same renderer).
fn view_from_model(model: PlanModel) -> PlanDagView {
PlanDagView::new("funnel-plandag", model)
}
// ── the embeddable Panel ────────────────────────────────────────────────────
// ── growth history ("addera historik") ────────────────────────────────────────
/// One recorded funnel snapshot: the DAG's size at a poll tick. Robot-observable
/// (rides in `state_json.history`) so a test can watch the trail grow.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FunnelSample {
pub node_count: usize,
pub edge_count: usize,
}
/// Cap on the growth-history ring — a long live session can't leak memory.
pub const HISTORY_CAP: usize = 64;
/// **Bounded, de-duplicated growth history of the funnel's DAG size.** Each
/// SUCCESSFUL poll records the (node, edge) counts; a consecutive duplicate collapses
/// so the trail shows only the moments the DAG actually *changed* (grew) — the
/// "addera historik" trajectory the live pane draws under the current frame. Capped at
/// [`HISTORY_CAP`] (oldest dropped). Pure (no clock, no network) ⇒ unit-testable.
#[derive(Clone, Debug, Default)]
pub struct FunnelHistory {
samples: Vec<FunnelSample>,
}
impl FunnelHistory {
/// Record a snapshot. Returns `true` if it was appended, `false` if it collapsed
/// into the previous (identical) sample. Keeps at most [`HISTORY_CAP`] samples.
pub fn record(&mut self, node_count: usize, edge_count: usize) -> bool {
let s = FunnelSample {
node_count,
edge_count,
};
if self.samples.last() == Some(&s) {
return false;
}
self.samples.push(s);
if self.samples.len() > HISTORY_CAP {
let overflow = self.samples.len() - HISTORY_CAP;
self.samples.drain(0..overflow);
}
true
}
/// Number of recorded (change) samples.
pub fn len(&self) -> usize {
self.samples.len()
}
/// True when nothing has been recorded yet.
pub fn is_empty(&self) -> bool {
self.samples.is_empty()
}
/// The recorded samples, oldest→newest.
pub fn samples(&self) -> &[FunnelSample] {
&self.samples
}
/// The most recently recorded node count (0 if the trail is empty).
pub fn latest_node_count(&self) -> usize {
self.samples.last().map(|s| s.node_count).unwrap_or(0)
}
/// Largest node count ever recorded (the peak of the trail).
pub fn peak_nodes(&self) -> usize {
self.samples.iter().map(|s| s.node_count).max().unwrap_or(0)
}
/// Net node growth from the first recorded sample to the latest (can be negative
/// if the plan shrank; usually the DAG only grows).
pub fn node_growth(&self) -> i64 {
match (self.samples.first(), self.samples.last()) {
(Some(a), Some(b)) => b.node_count as i64 - a.node_count as i64,
_ => 0,
}
}
/// Drop the whole trail (the pane's "clear history" control).
pub fn clear(&mut self) {
self.samples.clear();
}
/// The trail as `state_json` data: an array of `{node_count,edge_count}`, oldest→newest.
pub fn to_json(&self) -> serde_json::Value {
serde_json::Value::Array(
self.samples
.iter()
.map(|s| serde_json::json!({ "node_count": s.node_count, "edge_count": s.edge_count }))
.collect(),
)
}
}
// ── run history ("addera historik", the NEXT layer) ───────────────────────────
/// The outcome of a captured funnel run, derived from its DAG's node statuses:
/// any Failed ⇒ `Failed`; else every node Done ⇒ `Done`; else `Running`. Robot-
/// observable (rides in `state_json.runs[].outcome`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RunOutcome {
Running,
Done,
Failed,
}
impl RunOutcome {
/// A short, stable JSON/UI token.
pub fn as_str(self) -> &'static str {
match self {
RunOutcome::Running => "running",
RunOutcome::Done => "done",
RunOutcome::Failed => "failed",
}
}
}
/// Derive the run outcome from a projected [`PlanDag`]'s node statuses. An empty DAG
/// is `Running` (nothing done yet). Pure ⇒ unit-testable.
fn run_outcome_from_dag(dag: &PlanDag) -> RunOutcome {
let mut any = false;
let mut any_failed = false;
let mut all_done = true;
for n in &dag.nodes {
any = true;
match n.status {
DagStatus::Failed => any_failed = true,
DagStatus::Done => {}
_ => all_done = false,
}
}
if any_failed {
RunOutcome::Failed
} else if any && all_done {
RunOutcome::Done
} else {
RunOutcome::Running
}
}
/// One recorded funnel **run**: a full point-in-time snapshot of the plan-DAG the pane
/// can re-select and re-draw (distinct from [`FunnelHistory`], which keeps only the
/// size trajectory). Carries the captured [`PlanModel`] (the DAG itself, Clone) so a
/// selected past run renders through the SAME [`facett_plandag::PlanDagView`] as the
/// live view — no second renderer (L5).
#[derive(Clone, Debug)]
pub struct FunnelRun {
/// 1-based capture sequence (stable id the picker + `state_json` address).
pub seq: usize,
pub node_count: usize,
pub edge_count: usize,
pub outcome: RunOutcome,
/// egui-input time (seconds) when captured (0.0 if none supplied). Started marker.
pub captured_at_s: f64,
/// The captured DAG snapshot — re-rendered on selection via `PlanDagView::new`.
pub model: PlanModel,
}
impl FunnelRun {
/// The run summary AS DATA (rides in `state_json.runs`), WITHOUT the heavy model.
fn summary_json(&self) -> serde_json::Value {
serde_json::json!({
"seq": self.seq,
"node_count": self.node_count,
"edge_count": self.edge_count,
"outcome": self.outcome.as_str(),
"captured_at_s": self.captured_at_s,
})
}
}
/// Cap on the run-history ring — a long live session can't leak the (heavier, model-
/// bearing) run snapshots.
pub const RUN_CAP: usize = 32;
/// **Bounded, de-duplicated log of past funnel runs** ("addera historik", the run
/// dimension). Each SUCCESSFUL poll whose DAG *changed* (node/edge count or outcome
/// differs from the previous capture) records a full snapshot; a duplicate collapses
/// so the log holds only distinct run states. Capped at [`RUN_CAP`] (oldest dropped).
/// Pure (no clock, no network) ⇒ unit-testable.
#[derive(Clone, Debug, Default)]
pub struct FunnelRunLog {
runs: Vec<FunnelRun>,
/// Monotonic seq counter (never reused, even after the ring drops old entries).
next_seq: usize,
}
impl FunnelRunLog {
/// Record a run snapshot. Returns `Some(seq)` if appended, `None` if it collapsed
/// into the previous (same node/edge count + outcome) capture. The model's initial
/// version supplies the node/edge counts (self-contained — no external counts).
pub fn record(&mut self, model: PlanModel, outcome: RunOutcome, at_s: f64) -> Option<usize> {
let node_count = model.initial.nodes.len();
let edge_count = model.initial.edges.len();
if let Some(last) = self.runs.last() {
if last.node_count == node_count
&& last.edge_count == edge_count
&& last.outcome == outcome
{
return None;
}
}
self.next_seq += 1;
let seq = self.next_seq;
self.runs.push(FunnelRun {
seq,
node_count,
edge_count,
outcome,
captured_at_s: at_s,
model,
});
if self.runs.len() > RUN_CAP {
let overflow = self.runs.len() - RUN_CAP;
self.runs.drain(0..overflow);
}
Some(seq)
}
/// Number of recorded runs.
pub fn len(&self) -> usize {
self.runs.len()
}
/// True when no run has been recorded yet.
pub fn is_empty(&self) -> bool {
self.runs.is_empty()
}
/// The recorded runs, oldest→newest.
pub fn runs(&self) -> &[FunnelRun] {
&self.runs
}
/// Look up a run by its stable `seq`.
pub fn get(&self, seq: usize) -> Option<&FunnelRun> {
self.runs.iter().find(|r| r.seq == seq)
}
/// Drop the whole log (the pane's "clear runs" control).
pub fn clear(&mut self) {
self.runs.clear();
}
/// The run summaries as `state_json` data (oldest→newest, each WITHOUT the model).
pub fn to_json(&self) -> serde_json::Value {
serde_json::Value::Array(self.runs.iter().map(|r| r.summary_json()).collect())
}
}
/// Connection outcome of the last fetch (drives the status line + `state_json`).
///
/// The failure arms are DISTINCT on purpose (the funnel-offline mislabel fix): the
/// pane must never claim *"server offline"* when the server is actually up. A funnel
/// fetch can fail for four very different reasons and the UI + `state_json` name each
/// one so a human (and a robot) can tell them apart:
/// * [`NoWarehouse`](Conn::NoWarehouse) — nothing is SELECTED to fetch (empty
/// workspace). The server is irrelevant; pick a warehouse.
/// * [`Unreachable`](Conn::Unreachable) — the server could not be connected to at all
/// (wrong port / not running).
/// * [`AuthFailed`](Conn::AuthFailed) — connected, but the bearer token was missing or
/// rejected (`Unauthenticated` / `PermissionDenied`).
/// * [`NotServed`](Conn::NotServed) — connected + authed, but the server does not serve
/// the selected warehouse (`NotFound`).
/// * [`ServerError`](Conn::ServerError) — any other RPC/server error.
#[derive(Clone, Debug, PartialEq, Eq)]
enum Conn {
/// Never fetched yet.
Idle,
/// The server returned a funnel (possibly empty).
Loaded,
/// No warehouse is selected (workspace empty) — nothing to fetch. NOT a server
/// problem; the pane guides the user to select one.
NoWarehouse,
/// The nornir-server could not be reached at all (connect failed).
Unreachable,
/// Reached the server, but the bearer token was missing or rejected.
AuthFailed,
/// Reached + authed, but the server does not serve the selected warehouse.
NotServed,
/// Any other RPC/server error (a clean, non-panicking state).
ServerError,
}
/// **Classify a failed fetch into a distinct [`Conn`] state** — the crux of the
/// funnel-offline mislabel fix. Never returns [`Conn::Loaded`]/[`Conn::Idle`]/
/// [`Conn::NoWarehouse`] (those are decided before the fetch). Downcasts the anyhow
/// error chain to a [`tonic::Status`] to read the gRPC code (auth vs not-served vs
/// other); if there is none, a connect-context error is [`Conn::Unreachable`], else
/// [`Conn::ServerError`]. Pure ⇒ unit-testable.
fn classify_conn(e: &anyhow::Error) -> Conn {
for cause in e.chain() {
if let Some(st) = cause.downcast_ref::<tonic::Status>() {
use tonic::Code;
return match st.code() {
Code::Unauthenticated | Code::PermissionDenied => Conn::AuthFailed,
Code::NotFound => Conn::NotServed,
_ => Conn::ServerError,
};
}
}
let msg = format!("{e:#}").to_ascii_lowercase();
if msg.contains("connect to nornir-server")
|| msg.contains("tcp connect")
|| msg.contains("transport error")
{
Conn::Unreachable
} else {
Conn::ServerError
}
}
/// The LIVE funnel pane: fetches nornir-server's funnel over thin gRPC and renders
/// the reconstructed Plan-DAG with [`facett_plandag`]. A [`Facet`] (hence a [`Panel`]
/// via the blanket impl) a host mounts in a tab.
pub struct FunnelPanel {
/// The FAT / EMBEDDED engine — `Some` ⇒ the fat-mode default (in-process funnel,
/// **no network**, `nornir-server` may be DOWN); `None` ⇒ thin/remote (gRPC to an
/// explicitly-configured `--server`/`NORNIR_SERVER`). Embedded is the default so
/// the pane renders + polls its own DAG with no server running — never the old
/// "no remote nornir"/unreachable error.
embedded: Option<EmbeddedFunnel>,
/// HOST-injected LIVE source (embedded mode only — [`LiveFunnelSource`]): consulted
/// lazily on each refresh/poll. `None` ⇒ purely local (the classic embedded pane).
live: Option<std::sync::Arc<dyn LiveFunnelSource>>,
/// Outcome of the LAST live consult: `None` = not consulted yet (lazy — nothing
/// dialed at construction); `Some(Conn::Loaded)` = the rendered funnel IS the
/// server's; any other `Some(_)` = a NAMED outage (the embedded state is shown).
live_conn: Option<Conn>,
/// The last live consult's error display text (state_json + hover — honesty).
live_error: Option<String>,
/// **A live consult IN FLIGHT** (the worker's receiver + when it started).
///
/// The consult is a full gRPC dial + `Funnel.Show` round trip
/// ([`fetch_funnel`] → `block_on`), and [`Panel::ui`] used to run it INLINE on
/// every auto-poll tick. The frame that carried the poll could not paint until
/// the server answered — and against a dead endpoint it held for the whole 3 s
/// connect timeout, every interval, forever.
///
/// MEASURED (oden, 2026-08-27, `dwarves-gui/tests/robot_funnel_off_thread.rs`
/// driving the REAL `Panel::ui` through the robot harness with a `SlowLive`
/// source standing in for the gRPC consult at a deliberate 1 500 ms; the number
/// is the wall clock of the frame that carries the auto-poll):
///
/// ```text
/// inline: 1 508.97 ms per polling frame — 100.6% of the consult; the window cannot paint
/// off-thread: 2.43 ms per polling frame — 0.2% of it; the worker is started and the frame returns
/// ```
///
/// 620x off the polling frame, and — the actual invariant — independent of what
/// the consult costs. The 3 s connect timeout against a DEAD endpoint was the
/// worst arm of the same defect: it froze the window for the full timeout, every
/// interval, for as long as the pane was open.
///
/// The frame polls this ([`Self::poll_live`]); the scripted/robot seam blocks on
/// it ([`Self::wait_live`]). A refresh pressed while one is in flight is not a
/// second round trip.
live_pending: Option<(
std::sync::mpsc::Receiver<Result<funnel_core::state::Funnel>>,
std::time::Instant,
)>,
endpoint: String,
token: String,
workspace: String,
view: Option<PlanDagView>,
conn: Conn,
node_count: usize,
edge_count: usize,
/// The last error's display text (folded into `state_json` for tests).
error: Option<String>,
/// LIVE auto-poll: when on, the pane re-fetches `Funnel.Show` every
/// `poll_interval_s` so the plan-DAG **grows on its own** as ideas land (no
/// manual Refresh needed). On by default.
auto_poll: bool,
/// Seconds between auto-poll re-fetches.
poll_interval_s: f64,
/// egui-input time (seconds) of the last (auto or manual) fetch; the poll is
/// due once `now - last_poll_s >= poll_interval_s`.
last_poll_s: f64,
/// The submit strip's text buffer (the idea/error you type into the pane).
draft: String,
/// Which intake kind the submit strip is set to (idea | error).
intake_kind: IntakeKind,
/// Outcome of the last submit, folded into `state_json` for robot assertions:
/// `Some(Ok(id))` = accepted, `Some(Err(msg))` = rejected, `None` = none yet.
last_submit: Option<std::result::Result<String, String>>,
/// LIVE growth trail ("addera historik"): the DAG's (node,edge) size recorded on
/// every successful poll (consecutive duplicates collapsed), so the pane draws the
/// funnel's growth trajectory under the current frame — not just the latest state.
history: FunnelHistory,
/// RUN HISTORY ("addera historik", the run dimension): a de-duplicated log of past
/// funnel runs, each a full DAG snapshot the user can re-select to view a prior
/// run's final plan. The live view stays the default ("latest").
runs: FunnelRunLog,
/// The selected past run's `seq` (`None` ⇒ the live/latest view is shown).
selected_run: Option<usize>,
/// The re-rendered DAG of the selected past run (rebuilt on selection from the
/// run's `model`), so a historical run draws through the SAME `PlanDagView`.
snapshot_view: Option<PlanDagView>,
/// The latest egui-input time (seconds), captured at the top of `ui()`; stamps a
/// recorded run's `captured_at_s` on the next poll. 0.0 until the pane renders.
now_s: f64,
/// The Claude credential SOURCE, as reported by the host over
/// `{"claude_source": "..."}` (e.g. `env:ANTHROPIC_API_KEY`, `vault:anthropic`,
/// `none`). The funnel drives Claude to expand ideas, so the pane shows whether a
/// credential is configured and points at Settings when it is not.
///
/// A SOURCE LABEL ONLY — this crate never reads, stores or renders key material,
/// and has no vault dependency. The host resolves it (dwarves does so via
/// `dwarves::claude_token::status`) and pushes the label down.
claude_source: String,
/// The SAME DAG as [`view`](Self::view), rendered as a node-link graph on the
/// shared `facett-graphview` engine. Rebuilt on every successful poll from the
/// same `PlanDag`, so the two renders never disagree about the funnel.
graph: FunnelGraphPane,
/// The SAME DAG rendered as a **3D layered graph** — a camera-only 3D placement
/// (ranks stacked in space) with an optional auto-rotate + clickable drill. Rebuilt
/// (cached) on every successful poll from the same `PlanDag`, so all three renders
/// agree about the funnel.
graph3d: FunnelGraph3d,
/// Which render of the DAG is on screen — the card lane, the flat graph, or the 3D graph.
dag_view: FunnelDagView,
}
/// Which render of the funnel DAG the pane is showing. Two renders, one DAG: the
/// card lane (`facett-plandag`, metric-bearing) answers *what state is each node in*;
/// the graph (`facett-graphview`) answers *what needs what*.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FunnelDagView {
/// The `facett-plandag` metric-bearing card lane (the default).
Cards,
/// The `facett-graphview` node-link graph (flat 2D).
Graph,
/// The 3D layered node-link graph (camera-only 3D + optional auto-rotate).
Graph3d,
}
impl FunnelDagView {
fn key(self) -> &'static str {
match self {
FunnelDagView::Cards => "cards",
FunnelDagView::Graph => "graph",
FunnelDagView::Graph3d => "graph3d",
}
}
fn parse(s: &str) -> Option<Self> {
Some(match s {
"cards" | "plandag" => FunnelDagView::Cards,
"graph" | "graphview" => FunnelDagView::Graph,
"graph3d" | "3d" | "graph_3d" => FunnelDagView::Graph3d,
_ => return None,
})
}
}
/// Default seconds between live auto-poll re-fetches of the funnel.
pub const DEFAULT_POLL_INTERVAL_S: f64 = 1.5;
impl FunnelPanel {
/// The EVENT-SOURCED run history (newest first) — funnel-core's own
/// `RunRecorded` ledger, projected by [`EmbeddedFunnel::runs_json`]. An empty
/// array in remote mode (the `Funnel.Show` RPC carries no run ledger).
pub fn event_runs(&self) -> serde_json::Value {
self.embedded
.as_ref()
.map(|e| e.runs_json())
.unwrap_or_else(|| serde_json::Value::Array(Vec::new()))
}
/// How many runs the event log recorded (0 in remote mode).
pub fn event_run_count(&self) -> usize {
self.embedded.as_ref().map(|e| e.run_count()).unwrap_or(0)
}
/// The Claude credential source label the host reported.
pub fn claude_source(&self) -> &str {
&self.claude_source
}
/// Set the Claude credential source label (the host's resolved status).
/// A blank value normalises to `none` so the seam never carries an empty label.
pub fn set_claude_source(&mut self, source: impl Into<String>) {
let s = source.into();
self.claude_source = if s.trim().is_empty() {
"none".to_string()
} else {
s
};
}
}
impl FunnelPanel {
/// Build the pane for `endpoint` and eagerly attempt the first fetch. A failed
/// fetch is captured as the `Unavailable` state (never a panic).
pub fn new(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Self {
let mut p = Self {
embedded: None,
live: None,
live_conn: None,
live_error: None,
live_pending: None,
endpoint: endpoint.into(),
token: token.into(),
workspace: workspace.into(),
view: None,
conn: Conn::Idle,
node_count: 0,
edge_count: 0,
error: None,
auto_poll: true,
poll_interval_s: DEFAULT_POLL_INTERVAL_S,
last_poll_s: 0.0,
draft: String::new(),
intake_kind: IntakeKind::Idea,
last_submit: None,
history: FunnelHistory::default(),
runs: FunnelRunLog::default(),
selected_run: None,
snapshot_view: None,
now_s: 0.0,
claude_source: "none".to_string(),
graph: FunnelGraphPane::new(),
graph3d: FunnelGraph3d::new(),
dag_view: FunnelDagView::Cards,
};
p.refresh();
p
}
/// [`new_embedded`](Self::new_embedded) over the SAMPLE funnel — tests only
/// (`testkit`): the pane a test drives when it needs a non-empty Plan-DAG.
#[cfg(any(test, feature = "testkit"))]
pub fn new_embedded_seeded(workspace: impl Into<String>) -> Self {
let workspace = workspace.into();
let mut p = Self::new_embedded(workspace.clone());
p.embedded = Some(EmbeddedFunnel::seeded(workspace));
p.refresh();
p
}
/// **Build the pane in FAT / EMBEDDED mode (the default)** — an in-process funnel
/// engine over the local workspace, **no `nornir-server`**. The pane renders +
/// polls its own Plan-DAG with no server running (the fat-mode doctrine: dwarves is
/// self-contained; nornir-server can be DOWN). Only [`new`](Self::new) /
/// [`panel_with`] opt into thin/remote gRPC, and only when a real endpoint is set.
pub fn new_embedded(workspace: impl Into<String>) -> Self {
let workspace = workspace.into();
let mut p = Self {
embedded: Some(EmbeddedFunnel::new(workspace.clone())),
live: None,
live_conn: None,
live_error: None,
live_pending: None,
endpoint: "embedded (local · no server)".to_string(),
token: String::new(),
workspace,
view: None,
conn: Conn::Idle,
node_count: 0,
edge_count: 0,
error: None,
auto_poll: true,
poll_interval_s: DEFAULT_POLL_INTERVAL_S,
last_poll_s: 0.0,
draft: String::new(),
intake_kind: IntakeKind::Idea,
last_submit: None,
history: FunnelHistory::default(),
runs: FunnelRunLog::default(),
selected_run: None,
snapshot_view: None,
now_s: 0.0,
claude_source: "none".to_string(),
graph: FunnelGraphPane::new(),
graph3d: FunnelGraph3d::new(),
dag_view: FunnelDagView::Cards,
};
p.refresh();
p
}
/// True when the pane is in the FAT / EMBEDDED (no-network) mode.
pub fn is_embedded(&self) -> bool {
self.embedded.is_some()
}
/// **Attach a HOST-injected live source** (the [`LiveFunnelSource`] seam) to an
/// embedded pane. LAZY on purpose: nothing is dialed here — the next refresh
/// (the auto-poll tick, a manual ⟳, a submit's re-fetch, a workspace re-bind)
/// consults it, so construction stays immediate/offline-first even against a
/// dead or slow endpoint (the Warehouse Navigator's injection posture).
pub fn with_live_source(mut self, source: Box<dyn LiveFunnelSource>) -> Self {
// Taken as a `Box` (every caller's spelling is unchanged) and HELD as an
// `Arc` — which is what the off-thread consult worker clones.
self.live = Some(source.into());
self.live_conn = None;
self.live_error = None;
self
}
/// True when the rendered funnel is the LIVE server's (embedded pane whose last
/// live consult loaded) — the provenance the submit routing keys on.
fn live_loaded(&self) -> bool {
self.embedded.is_some() && self.live_conn == Some(Conn::Loaded)
}
/// **Submit the current draft** as an idea/error into nornir's funnel, then
/// re-fetch so the pane's DAG immediately reflects the new item. A blank draft is
/// a no-op. The outcome (accepted id or error text) is recorded in `last_submit`
/// for the status line + `state_json`. Never panics — an offline server is
/// captured as a rejected submit, exactly like a failed fetch.
pub fn submit_draft(&mut self) {
let text = self.draft.trim().to_string();
if text.is_empty() {
return;
}
// EMBEDDED + LIVE, leg UP: the submit rides the SAME wire the MCP write leg
// uses (`Funnel.SubmitIdea` via the injected source), so the idea lands in
// the server's funnel_events-backed store — and the refresh below re-pulls
// the live funnel, making it VISIBLE at once (submit → visible). A rejected
// wire submit is recorded honestly, never silently retried locally (that
// would show "submitted" for an idea the server does not hold).
if self.live_loaded() {
if let Some(src) = &self.live {
match src.submit(&self.workspace, &text, self.intake_kind) {
Ok(id) => {
self.last_submit = Some(Ok(id));
self.draft.clear();
// SUBMIT -> VISIBLE is the contract this leg exists for, so the
// re-pull is the BLOCKING one: the idea must be on screen when
// the submit returns, not one poll interval later. The submit
// itself already went over the wire on this thread — waiting for
// its read-back adds no freeze the write did not already have.
self.refresh_blocking();
}
Err(e) => {
self.last_submit = Some(Err(format!("{e:#}")));
}
}
return;
}
}
// FAT / EMBEDDED (default, or the live leg is DOWN): apply the idea/error to
// the in-process funnel — no network, so it grows the DAG offline exactly
// like the live server would.
if let Some(emb) = &mut self.embedded {
match emb.submit(&text, self.intake_kind) {
Ok(id) => {
self.last_submit = Some(Ok(id));
self.draft.clear();
self.refresh();
}
Err(e) => {
self.last_submit = Some(Err(format!("{e:#}")));
}
}
return;
}
match submit_idea(
&self.endpoint,
&self.token,
&self.workspace,
&text,
self.intake_kind,
) {
Ok(id) => {
self.last_submit = Some(Ok(id));
self.draft.clear();
// Grow the DAG now, don't wait for the next poll tick.
self.refresh();
}
Err(e) => {
self.last_submit = Some(Err(format!("{e:#}")));
}
}
}
/// True when auto-poll is on AND at least one interval has elapsed since the
/// last fetch (`now_s` = egui input time). Pure/deterministic → unit-testable
/// without a clock or the network.
pub fn poll_due(&self, now_s: f64) -> bool {
self.auto_poll && (now_s - self.last_poll_s) >= self.poll_interval_s
}
/// Turn live auto-poll on/off (the ● live / ⏸ paused toggle).
pub fn set_auto_poll(&mut self, on: bool) {
self.auto_poll = on;
}
/// The growth-history trail ("addera historik"), oldest→newest.
pub fn history(&self) -> &FunnelHistory {
&self.history
}
/// The run history log (past funnel runs), oldest→newest.
pub fn runs(&self) -> &FunnelRunLog {
&self.runs
}
/// **Capture a plan-DAG snapshot as a run** — the exact entry [`refresh`] uses on a
/// live poll (de-duped by node/edge count + outcome). Exposed so a host, or a test
/// without a live server, can feed the run history. Returns the new `seq` (or `None`
/// if it collapsed into the previous capture).
pub fn record_run(
&mut self,
model: PlanModel,
outcome: RunOutcome,
at_s: f64,
) -> Option<usize> {
self.runs.record(model, outcome, at_s)
}
/// **Select a past run to view** (`Some(seq)`) or return to the live/latest view
/// (`None`). Selecting a known run rebuilds its DAG snapshot through a fresh
/// [`PlanDagView`]; an unknown seq is ignored (stays on whatever was shown). This
/// is the run-picker's handler — the headless `{"select_run": …}` seam drives it.
pub fn select_run(&mut self, seq: Option<usize>) {
match seq {
Some(s) => {
if let Some(run) = self.runs.get(s) {
self.selected_run = Some(s);
self.snapshot_view = Some(view_from_model(run.model.clone()));
}
}
None => {
self.selected_run = None;
self.snapshot_view = None;
}
}
}
/// The plan-DAG view currently shown: the selected snapshot when a past run is
/// picked, else the live view. `None` when neither exists (offline, no selection).
fn current_view(&self) -> Option<&PlanDagView> {
if self.selected_run.is_some() {
self.snapshot_view.as_ref()
} else {
self.view.as_ref()
}
}
fn current_view_mut(&mut self) -> Option<&mut PlanDagView> {
if self.selected_run.is_some() {
self.snapshot_view.as_mut()
} else {
self.view.as_mut()
}
}
/// **The current bottleneck** `(id, title)` — CONSUMED from the shown plan-DAG
/// view's `state_json` (`facett_plandag`'s `top_bottleneck` + the node titles),
/// never recomputed here (L5). `None` when there is no constrained node (empty /
/// unavailable DAG). The pane renders it as a *"clear next · <title>"* hint.
fn bottleneck_hint(&self) -> Option<(String, String)> {
let st = Facet::state_json(self.current_view()?);
let top = st.get("top_bottleneck")?.as_str()?.to_string();
let title = st
.get("nodes")
.and_then(|n| n.as_array())
.and_then(|arr| {
arr.iter()
.find(|n| n.get("id").and_then(|i| i.as_str()) == Some(top.as_str()))
.and_then(|n| n.get("title").and_then(|t| t.as_str()))
.map(|s| s.to_string())
})
.unwrap_or_else(|| top.clone());
Some((top, title))
}
/// **Bind the pane to a (newly) selected warehouse** and re-fetch. This is the
/// mirror of nornir's per-tab `set_workspace` (viz `switch_workspace` calls it on
/// every tab): the host's warehouse selector routes the selected name here, the
/// funnel re-scopes to it, and the next fetch loads THAT warehouse's funnel. An
/// empty name is the legitimate *no selection* resting state (→ [`Conn::NoWarehouse`]),
/// never an error.
pub fn set_workspace(&mut self, workspace: impl Into<String>) {
self.workspace = workspace.into();
self.selected_run = None;
self.snapshot_view = None;
// In embedded mode, re-scope the in-process funnel to the newly-selected
// workspace (no network) — the fat-mode mirror of the remote re-fetch.
if self.embedded.is_some() {
self.embedded = Some(EmbeddedFunnel::new(self.workspace.clone()));
}
self.refresh();
}
/// Set the bearer token used for the `Authorization` header and re-fetch (the host
/// may resolve the token after construction — e.g. from `~/.nornir/token`).
pub fn set_token(&mut self, token: impl Into<String>) {
self.token = token.into();
self.refresh();
}
/// The currently-bound warehouse/workspace name (`""` = none selected).
pub fn workspace(&self) -> &str {
&self.workspace
}
/// **Build the Plan-DAG from `funnel` and record it** — the ONE place the pane
/// turns a funnel into a render, whether that funnel came from the in-process
/// engine or off the wire. Split out of [`Self::refresh`] so the off-thread live
/// consult lands through the IDENTICAL path the synchronous embedded one takes
/// (LAW 5: one writer, never a second lowering that can drift).
fn land_funnel(&mut self, funnel: funnel_core::state::Funnel) {
let dag = PlanDag::from_funnel(&funnel, None);
self.node_count = dag.node_count();
self.edge_count = dag.edge_count();
let outcome = run_outcome_from_dag(&dag);
// The graph renders are the SAME dag — lowered before the dag is consumed.
self.graph.set_dag(&dag);
self.graph3d.set_dag(&dag);
let view = view_from_dag(&dag);
let model = view.model().clone();
self.view = Some(view);
self.conn = Conn::Loaded;
self.error = None;
self.history.record(self.node_count, self.edge_count);
self.record_run(model, outcome, self.now_s);
}
/// **Start the live consult on a worker thread.** The worker owns a clone of the
/// `Arc`d source and does the gRPC dial + `Funnel.Show`; the frame keeps painting
/// and [`Self::poll_live`] lands the result.
///
/// A consult already in flight is NOT restarted — an auto-poll tick that arrives
/// while the previous one is still dialing is not a second round trip (the same
/// rule the Coverage / Data view / Coderoom / Lenses panes hold).
///
/// A blank workspace short-circuits to [`Conn::NoWarehouse`] WITHOUT dialing:
/// nothing is selected to fetch, so it is not a server problem.
fn start_live_consult(&mut self) {
if self.live_pending.is_some() {
return;
}
let Some(src) = &self.live else { return };
if self.workspace.trim().is_empty() {
// Nothing SELECTED to fetch live — the thin pane's NoWarehouse rule,
// mirrored (not a server problem; no dial, and no worker).
self.live_conn = Some(Conn::NoWarehouse);
self.live_error = None;
return;
}
let src = std::sync::Arc::clone(src);
let workspace = self.workspace.clone();
let (tx, rx) = std::sync::mpsc::channel();
// ROOT LAW #0: a NAMED worker. A spawn failure must not leave the pane
// believing a consult is in flight that nothing will ever answer — the
// `live_pending` slot is only filled once the thread actually started.
match std::thread::Builder::new()
.name("funnel-live-consult".into())
.spawn(move || {
let _ = tx.send(src.fetch(&workspace));
}) {
Ok(_) => self.live_pending = Some((rx, std::time::Instant::now())),
Err(e) => {
self.live_conn = Some(Conn::Unreachable);
self.live_error = Some(format!("could not start the funnel consult worker: {e}"));
}
}
}
/// **Non-blocking: land an in-flight live consult if the worker answered.** Called
/// once per frame from [`Panel::ui`], so the window never waits on the wire.
///
/// Returns `true` when a consult landed this call (the caller may want a repaint).
/// A consult still dialing leaves every rendered field alone — offline-first: the
/// embedded content stays on screen with `live_status` reporting the leg's state.
pub fn poll_live(&mut self) -> bool {
let Some((rx, _)) = &self.live_pending else {
return false;
};
match rx.try_recv() {
Ok(res) => {
self.live_pending = None;
self.land_live(res);
true
}
Err(std::sync::mpsc::TryRecvError::Empty) => false,
Err(std::sync::mpsc::TryRecvError::Disconnected) => {
self.live_pending = None;
// A dead worker is a FAILED consult, which by this pane's own contract
// KEEPS the embedded content and names the outage. Blanking a rendered
// funnel because a thread died would be the fabrication the pane exists
// to prevent.
self.live_conn = Some(Conn::Unreachable);
self.live_error =
Some("the funnel consult worker died before it answered".to_string());
true
}
}
}
/// **Blocking: wait for the in-flight live consult.** The SCRIPTED seam — the robot
/// harness (`update_json`) and a submit's re-fetch use it so a driven refresh is
/// complete when the call returns. The egui frame never calls this; it calls
/// [`Self::poll_live`].
pub fn wait_live(&mut self) {
let Some((rx, _)) = self.live_pending.take() else {
return;
};
match rx.recv() {
Ok(res) => self.land_live(res),
Err(_) => {
self.live_conn = Some(Conn::Unreachable);
self.live_error =
Some("the funnel consult worker died before it answered".to_string());
}
}
}
/// Land a consult's outcome: classify it with the SAME [`classify_conn`] rules the
/// synchronous path used (the error travels whole, so nothing is re-worded on the
/// way off the worker), and rebuild the DAG from whichever funnel now applies.
fn land_live(&mut self, res: Result<funnel_core::state::Funnel>) {
match res {
Ok(f) => {
self.live_conn = Some(Conn::Loaded);
self.live_error = None;
self.land_funnel(f);
}
Err(e) => {
self.live_conn = Some(classify_conn(&e));
self.live_error = Some(format!("{e:#}"));
// The outage KEEPS the in-process state on screen (offline-first) and
// re-lands it, so the render matches what `funnel_source` claims.
if let Some(emb) = &self.embedded {
let f = emb.funnel();
self.land_funnel(f);
}
}
}
}
/// Whether a live consult is in flight (the robot seam + `state_json`).
pub fn is_live_pending(&self) -> bool {
self.live_pending.is_some()
}
/// **Refresh and WAIT for the live leg** — the synchronous contract the scripted
/// paths want: when this returns, the consult (if any) has landed. `ui()` must
/// never call it; it calls [`Self::refresh`] + [`Self::poll_live`].
pub fn refresh_blocking(&mut self) {
self.refresh();
self.wait_live();
}
/// (Re-)fetch the funnel and rebuild the Plan-DAG. On error, drops into the clean,
/// DISTINCT failure state ([`classify_conn`]) — never the old catch-all "offline".
/// A blank workspace short-circuits to [`Conn::NoWarehouse`] WITHOUT touching the
/// network: nothing is selected to fetch, so it is not a server problem.
pub fn refresh(&mut self) {
// FAT / EMBEDDED (default): build the DAG from the in-process funnel engine
// over the local workspace — NO network, so `nornir-server` can be DOWN and the
// pane still renders + polls a live DAG. Never the "no remote nornir" error.
if self.embedded.is_some() {
// EMBEDDED + LIVE (the host-injected read leg): START the consult OFF the
// caller's thread. A healthy consult renders the SERVER's funnel — the
// `funnel_events`-backed store an MCP `funnel_submit_idea` writes into —
// through the very same DAG path `land_funnel` runs. A failed consult keeps
// the in-process state and NAMES the outage (`live_conn` → `live_status`),
// with the same distinct vocabulary as the thin pane. Never fabricated.
//
// It used to be consulted RIGHT HERE, synchronously, and `Panel::ui` calls
// `refresh()` on every auto-poll tick — so the gRPC round trip ran on the
// egui thread and the window could not paint until the server answered.
if self.live.is_some() {
self.start_live_consult();
// Nothing has ever been rendered (the pane's very first refresh): show
// the embedded funnel NOW rather than a blank frame while the consult
// is in flight. Offline-first, exactly as before — the live answer
// replaces it when it lands. `history`/`runs` both collapse consecutive
// duplicates, so this costs no spurious trail entry.
if self.view.is_none() {
let f = self
.embedded
.as_ref()
.expect("embedded checked above")
.funnel();
self.land_funnel(f);
}
return;
}
let f = self
.embedded
.as_ref()
.expect("embedded checked above")
.funnel();
self.land_funnel(f);
return;
}
// THIN / REMOTE: only reached when an explicit endpoint was configured.
if self.workspace.trim().is_empty() {
self.view = None;
self.node_count = 0;
self.edge_count = 0;
self.conn = Conn::NoWarehouse;
self.error = None;
return;
}
match fetch_funnel(&self.endpoint, &self.token, &self.workspace) {
Ok(funnel) => {
let dag = PlanDag::from_funnel(&funnel, None);
self.node_count = dag.node_count();
self.edge_count = dag.edge_count();
let outcome = run_outcome_from_dag(&dag);
// The graph renders are the SAME dag — lowered before the dag is consumed.
self.graph.set_dag(&dag);
self.graph3d.set_dag(&dag);
let view = view_from_dag(&dag);
// Capture this poll as a run snapshot (de-duped): the model is the DAG
// itself, so a selected past run re-renders through the same view.
let model = view.model().clone();
self.view = Some(view);
self.conn = Conn::Loaded;
self.error = None;
// Grow the history trail. A transient offline blip (the Err arm) never
// records, so it can't wipe the growth trajectory — only real loads add.
self.history.record(self.node_count, self.edge_count);
self.record_run(model, outcome, self.now_s);
}
Err(e) => {
self.view = None;
self.node_count = 0;
self.edge_count = 0;
self.conn = classify_conn(&e);
self.error = Some(format!("{e:#}"));
}
}
}
fn conn_key(&self) -> &'static str {
let embedded = self.embedded.is_some();
let live = self.live_loaded();
match self.conn {
Conn::Idle => "idle",
Conn::Loaded if self.node_count == 0 => {
// Provenance in the status itself: live-from-server vs in-process.
if live {
"live_empty"
} else if embedded {
"embedded_empty"
} else {
"empty"
}
}
Conn::Loaded => {
if live {
"live"
} else if embedded {
"embedded"
} else {
"loaded"
}
}
ref c => conn_token(c),
}
}
/// `state_json.live_status` — the injected live leg's own state, DISTINCT from the
/// content `status`: `none` (no source injected), `pending` (injected, not yet
/// consulted — lazy), `live` (the rendered funnel IS the server's), or one of the
/// thin pane's named failure tokens (the outage, while embedded content is shown).
fn live_key(&self) -> &'static str {
match (&self.live, &self.live_conn) {
(None, _) => "none",
(Some(_), None) => "pending",
(Some(_), Some(Conn::Loaded)) => "live",
(Some(_), Some(c)) => conn_token(c),
}
}
/// **Provenance of the rendered funnel content** (`state_json.funnel_source`) —
/// the Warehouse Navigator's `skeleton_source` vocabulary: `live: …` when the
/// server's funnel is on screen, `embedded (…)` when the in-process one is (with
/// the live outage NAMED inside the parenthesis when a source was injected).
fn funnel_source(&self) -> String {
if self.embedded.is_none() {
return format!("remote: {}", self.endpoint);
}
match (&self.live, &self.live_conn) {
(Some(src), Some(Conn::Loaded)) => format!("live: {}", src.describe()),
(Some(src), Some(c)) => {
format!("embedded (live {} — {})", conn_token(c), src.describe())
}
(Some(src), None) => format!("embedded (live pending — {})", src.describe()),
(None, _) => "embedded (local · no server)".to_string(),
}
}
}
/// The stable token for a [`Conn`] arm — ONE vocabulary shared by the pane's content
/// `status` and the injected live leg's `live_status` (the funnel-offline mislabel
/// fix's five distinct failure names, not a second set).
fn conn_token(c: &Conn) -> &'static str {
match c {
Conn::Idle => "idle",
Conn::Loaded => "loaded",
Conn::NoWarehouse => "no_warehouse",
Conn::Unreachable => "unreachable",
Conn::AuthFailed => "auth_failed",
Conn::NotServed => "not_served",
Conn::ServerError => "server_error",
}
}
impl Facet for FunnelPanel {
fn title(&self) -> &str {
"Funnel"
}
fn ui(&mut self, ui: &mut egui::Ui) {
// LIVE auto-poll: re-fetch on the interval so the DAG grows on its own.
let now_s = ui.input(|i| i.time);
self.now_s = now_s; // stamp any run captured by the poll below
// Land a consult that finished since the last frame. NON-blocking: this is the
// egui thread, and the wire is not allowed to hold it.
if self.poll_live() {
ui.ctx().request_repaint();
}
if self.poll_due(now_s) {
// STARTS the consult (see `refresh`); it does not wait for it.
self.refresh();
self.last_poll_s = now_s;
}
if self.auto_poll {
// Keep the egui loop ticking even with no user input, so the next
// poll fires. Throttled to the poll interval — never a busy-spin.
ui.ctx()
.request_repaint_after(std::time::Duration::from_secs_f32(
self.poll_interval_s as f32,
));
}
let mut do_refresh = false;
ui.horizontal(|ui| {
if ui.button("⟳ Refresh").on_hover_text(format!("re-fetch the funnel from {}", self.endpoint)).clicked() {
do_refresh = true;
}
// Live / paused toggle for the auto-poll.
let (glyph, hint) = if self.auto_poll {
("● live", "auto-polling — click to pause")
} else {
("⏸ paused", "auto-poll paused — click to resume live")
};
if ui.button(glyph).on_hover_text(hint).clicked() {
self.auto_poll = !self.auto_poll;
self.last_poll_s = now_s;
}
ui.separator();
match self.conn {
Conn::Idle => {
ui.weak("connecting…");
}
Conn::Loaded if self.node_count == 0 => {
if self.embedded.is_some() {
ui.weak("embedded funnel is empty — add an idea to grow it (local · no server)");
} else {
ui.weak("funnel is empty (no ideas/plans yet)");
}
}
Conn::Loaded => {
// PLAIN NUMBERS (no thousands separators) — a nordisk law.
let live = self.live_loaded();
let lead = if live {
"live funnel (via host) · "
} else if self.embedded.is_some() {
"embedded funnel · "
} else {
"live funnel · "
};
ui.label(format!("{lead}{} nodes / {} edges", self.node_count, self.edge_count))
.on_hover_text(if live {
"the server's real funnel (funnel_events-backed), pulled through the host-injected live source"
} else if self.embedded.is_some() {
"in-process funnel engine over the local workspace — no nornir-server needed (fat mode)"
} else {
"live funnel from nornir-server (thin/remote)"
});
}
// ── DISTINCT failure states (the funnel-offline mislabel fix) ──
// Amber = "you/selection" problem (not the server); red = a real
// server/transport fault. NEVER "offline" when the server is up.
Conn::NoWarehouse => {
ui.colored_label(
egui::Color32::from_rgb(224, 170, 90),
"no warehouse selected — pick one to load its funnel",
)
.on_hover_text("the funnel belongs to a warehouse; select one in the warehouse picker");
}
Conn::NotServed => {
ui.colored_label(
egui::Color32::from_rgb(224, 170, 90),
format!("warehouse '{}' not served by nornir-server", self.workspace),
)
.on_hover_text(format!(
"nornir-server at {} is up but does not serve '{}' — register/populate it (nornir workspace add)",
self.endpoint, self.workspace
));
}
Conn::AuthFailed => {
let resp = ui.colored_label(
egui::Color32::from_rgb(224, 120, 90),
"authentication failed — check the nornir-server token",
);
if let Some(e) = &self.error {
resp.on_hover_text(format!("set $NORNIR_SERVER_TOKEN (or ~/.nornir/token): {e}"));
}
}
Conn::Unreachable => {
let resp = ui.colored_label(
egui::Color32::from_rgb(224, 120, 90),
format!("nornir-server unreachable at {}", self.endpoint),
);
if let Some(e) = &self.error {
resp.on_hover_text(e);
}
}
Conn::ServerError => {
let resp = ui.colored_label(
egui::Color32::from_rgb(224, 120, 90),
"funnel error (nornir-server returned an error)",
);
if let Some(e) = &self.error {
resp.on_hover_text(e);
}
}
}
});
if do_refresh {
self.refresh();
self.last_poll_s = now_s;
}
// ── the injected live leg's OUTAGE, NAMED (never silently absorbed) ──
// The embedded content is still shown (offline-first), but the pane says so:
// the host wired a live source and it could not be honoured this poll.
if self.embedded.is_some() {
if let (Some(_), Some(c)) = (&self.live, &self.live_conn) {
if *c != Conn::Loaded {
let msg = match c {
Conn::NoWarehouse => {
"live funnel: no warehouse selected — showing the embedded funnel"
.to_string()
}
c => format!(
"live funnel {} — showing the embedded funnel",
conn_token(c)
),
};
let resp = ui.colored_label(egui::Color32::from_rgb(224, 170, 90), msg);
if let Some(e) = &self.live_error {
resp.on_hover_text(e.as_str());
}
}
}
}
// ── the submit strip (WRITE leg): type an idea/error → grow the DAG ──
let mut do_submit = false;
ui.horizontal(|ui| {
// idea | error kind toggle.
egui::ComboBox::from_id_salt("funnel-intake-kind")
.selected_text(match self.intake_kind {
IntakeKind::Idea => "💡 idea",
IntakeKind::Error => "🐛 error",
})
.show_ui(ui, |ui| {
ui.selectable_value(&mut self.intake_kind, IntakeKind::Idea, "💡 idea");
ui.selectable_value(&mut self.intake_kind, IntakeKind::Error, "🐛 error");
});
let hint = match self.intake_kind {
IntakeKind::Idea => "describe an idea to plan…",
IntakeKind::Error => "describe an error to triage…",
};
let resp = ui.add(
egui::TextEdit::singleline(&mut self.draft)
.hint_text(hint)
.desired_width(ui.available_width() - 90.0),
);
// Enter in the field submits (and keeps focus flow natural).
if resp.lost_focus() && ui.input(|i| i.key_pressed(egui::Key::Enter)) {
do_submit = true;
}
let enabled = !self.draft.trim().is_empty();
if ui
.add_enabled(enabled, egui::Button::new("＋ Submit"))
.on_hover_text(format!(
"submit to nornir-server at {} (grows the plan-DAG)",
self.endpoint
))
.clicked()
{
do_submit = true;
}
});
// Last-submit feedback line.
match &self.last_submit {
Some(Ok(id)) => {
ui.colored_label(
egui::Color32::from_rgb(120, 190, 120),
format!("submitted · {id}"),
);
}
Some(Err(e)) => {
ui.colored_label(
egui::Color32::from_rgb(224, 120, 90),
"submit failed (server offline?)",
)
.on_hover_text(e.as_str());
}
None => {}
}
if do_submit {
self.submit_draft();
self.last_poll_s = now_s;
}
// ── the growth trail ("addera historik"): draw HOW the DAG grew over polls ──
if !self.history.is_empty() {
ui.horizontal(|ui| {
let growth = self.history.node_growth();
let sign = if growth >= 0 { "+" } else { "" };
ui.weak(format!(
"history · {} samples · peak {} nodes · {sign}{growth} grown",
self.history.len(),
self.history.peak_nodes(),
));
// The tail of the trail as node counts joined by → (last 12 for width).
let samples = self.history.samples();
let tail = &samples[samples.len().saturating_sub(12)..];
let trail = tail
.iter()
.map(|s| s.node_count.to_string())
.collect::<Vec<_>>()
.join(" → ");
ui.weak(trail)
.on_hover_text("funnel DAG node-count over successive polls (oldest → newest)");
if ui
.small_button("clear")
.on_hover_text("clear the growth history")
.clicked()
{
self.history.clear();
}
});
}
// ── EVENT-SOURCED run history: what actually RAN, from funnel-core's ledger ─
//
// Distinct from the run PICKER below (which snapshots the rendered DAG per
// poll): these are the funnel's own `RunRecorded` events, folded by
// funnel-core, so the strip shows real executions — who ran what, and how it
// came out — rather than view states.
if let Some(rows) = self.event_runs().as_array().filter(|r| !r.is_empty()) {
ui.horizontal_wrapped(|ui| {
ui.weak(format!("runs · {} recorded", rows.len()));
// Newest first (the projection's order); cap the strip's width.
for row in rows.iter().take(6) {
let outcome = row["outcome"].as_str().unwrap_or("?");
let (glyph, colour) = match outcome {
"ok" => ("✔", egui::Color32::from_rgb(0x4c, 0xaf, 0x50)),
"failed" => ("✖", egui::Color32::from_rgb(0xd9, 0x53, 0x4f)),
_ => ("⊘", egui::Color32::from_rgb(0xe0, 0xa8, 0x00)),
};
let node = row["node_id"].as_str().unwrap_or("?");
let run = row["run_id"].as_str().unwrap_or("?");
ui.colored_label(colour, format!("{glyph} {run}"))
.on_hover_text(format!(
"{run} on {node} — {outcome}, by {}",
row["ran_by"].as_str().unwrap_or("?")
));
}
if rows.len() > 6 {
ui.weak(format!("+{} older", rows.len() - 6));
}
});
}
// ── the Claude credential the funnel expands ideas with ──────────────────
//
// The funnel drives Claude, so a missing credential is the pane's problem to
// surface — silently producing no plan is the failure mode this prevents.
// A SOURCE LABEL only; the key never enters this crate.
if self.claude_source == "none" {
ui.horizontal(|ui| {
ui.colored_label(egui::Color32::from_rgb(0xe0, 0xa8, 0x00), "○");
ui.weak("no Claude credential — configure it in Settings to expand ideas");
});
}
// ── run history ("addera historik"): pick a PAST run to view its final DAG ──
if !self.runs.is_empty() {
let mut pick: Option<Option<usize>> = None;
ui.horizontal(|ui| {
let selected_text = match self.selected_run {
None => "◷ latest (live)".to_string(),
Some(seq) => match self.runs.get(seq) {
Some(r) => format!(
"run #{seq} · {} nodes · {}",
r.node_count,
r.outcome.as_str()
),
None => "◷ latest (live)".to_string(),
},
};
egui::ComboBox::from_id_salt("funnel-run-picker")
.selected_text(selected_text)
.show_ui(ui, |ui| {
if ui
.selectable_label(self.selected_run.is_none(), "◷ latest (live)")
.clicked()
{
pick = Some(None);
}
// Newest run first — the most recent history entry on top.
for r in self.runs.runs().iter().rev() {
let label = format!(
"run #{} · {} nodes / {} edges · {}",
r.seq,
r.node_count,
r.edge_count,
r.outcome.as_str()
);
if ui
.selectable_label(self.selected_run == Some(r.seq), label)
.clicked()
{
pick = Some(Some(r.seq));
}
}
});
ui.weak(format!("history · {} runs", self.runs.len()));
if self.selected_run.is_some()
&& ui
.small_button("● live")
.on_hover_text("return to the live/latest view")
.clicked()
{
pick = Some(None);
}
});
if let Some(sel) = pick {
self.select_run(sel);
}
// The selected past run's summary (started / node count / outcome).
if let Some(seq) = self.selected_run {
if let Some(r) = self.runs.get(seq) {
ui.weak(format!(
"viewing run #{seq} · {} nodes / {} edges · {}",
r.node_count,
r.edge_count,
r.outcome.as_str()
));
}
}
}
// ── bottleneck annotation: WHERE the funnel is constrained (consumed) ──
if let Some((_, title)) = self.bottleneck_hint() {
ui.horizontal(|ui| {
ui.colored_label(egui::Color32::from_rgb(224, 170, 90), format!("clear next · {title}"))
.on_hover_text("the top bottleneck (unblocks the most downstream work) — from facett-plandag's constraint analysis");
});
}
// ── which render of the DAG: card lane or node-link graph ────────────────
// Two renders of ONE dag. Only offered when there IS a dag to render, and the
// graph only for the LIVE view (a historical run snapshots the card view).
if self.current_view().is_some() {
ui.horizontal(|ui| {
ui.label("View:");
if ui.selectable_label(self.dag_view == FunnelDagView::Cards, "Cards").clicked() {
self.dag_view = FunnelDagView::Cards;
}
let can_graph = self.selected_run.is_none();
if ui
.add_enabled(can_graph, egui::Button::selectable(self.dag_view == FunnelDagView::Graph, "Graph"))
.on_disabled_hover_text("the graph render follows the LIVE funnel — return to latest to see it")
.clicked()
{
self.dag_view = FunnelDagView::Graph;
}
// The 3D layered render (live view only) — a camera-only 3D DAG.
if ui
.add_enabled(can_graph, egui::Button::selectable(self.dag_view == FunnelDagView::Graph3d, "3D Graph"))
.on_hover_text("the funnel Plan-DAG as a 3D layered graph (ranks stacked in space) — reuses the world graph's orthographic 3D camera")
.on_disabled_hover_text("the 3D graph follows the LIVE funnel — return to latest to see it")
.clicked()
{
self.dag_view = FunnelDagView::Graph3d;
}
if self.dag_view == FunnelDagView::Graph && can_graph {
ui.separator();
// The measurement, on screen: nodes drawn and the last frame cost,
// and that there is NO node cap (nornir's CallGraph caps at 160).
let us = self.graph.last_frame_us().unwrap_or(0);
ui.weak(format!(
"{} node(s) · {} edge(s) · {:.1} ms/frame · no node cap",
self.graph.node_count(),
self.graph.edge_count(),
us as f64 / 1000.0,
));
}
if self.dag_view == FunnelDagView::Graph3d && can_graph {
ui.separator();
// The "3D rotating" toggle — spins the CAMERA, never the layout.
let mut rot = self.graph3d.is_rotating();
if ui
.checkbox(&mut rot, "3D rotating")
.on_hover_text("auto-spin the camera about the vertical axis (projection-only — the layout stays cached). On by default.")
.changed()
{
self.graph3d.set_rotate(rot);
}
// The idea WORDS beside each node — on by default (the funnel is small;
// only the world graph's universe scale turns labels off).
let mut lab = self.graph3d.labels_on();
if ui
.checkbox(&mut lab, "labels")
.on_hover_text("each node's idea words, attached in 3D (depth-cued). On by default — funnel data is small.")
.changed()
{
self.graph3d.set_labels(lab);
}
let us = self.graph3d.last_frame_us().unwrap_or(0);
ui.weak(format!(
"{} node(s) · {} edge(s) · {:.1} ms/frame · no node cap · click a node to drill",
self.graph3d.node_count(),
self.graph3d.edge_count(),
us as f64 / 1000.0,
));
}
});
}
ui.separator();
// The GRAPH render (live view only) — the shared facett-graphview engine.
if self.dag_view == FunnelDagView::Graph
&& self.selected_run.is_none()
&& self.view.is_some()
{
self.graph.ui(ui);
return;
}
// The 3D layered GRAPH render (live view only) — camera-only 3D + rotate. A
// node click drills to the card detail: the id is forwarded to the SAME
// `facett-plandag` view (its selection), so switching to Cards shows that card's
// detail line — one DAG, one selection, reused across renders.
if self.dag_view == FunnelDagView::Graph3d
&& self.selected_run.is_none()
&& self.view.is_some()
{
if let Some(id) = self.graph3d.ui(ui) {
if let Some(v) = self.view.as_mut() {
// The card view keys resolved nodes as `node:<id>` (see plandag's
// `resolve`), so the raw funnel id is prefixed to hit the drill.
v.update(facett_plandag::Msg::Select(format!("node:{id}")));
}
}
return;
}
let endpoint = self.endpoint.clone();
let conn = self.conn.clone();
match self.current_view_mut() {
Some(v) => Facet::ui(v, ui),
None => {
let workspace = self.workspace.clone();
ui.centered_and_justified(|ui| match conn {
Conn::NoWarehouse => {
ui.weak("select a warehouse to load its funnel");
}
Conn::NotServed => {
ui.weak(format!("warehouse '{workspace}' is not served by nornir-server at {endpoint}"));
}
Conn::AuthFailed => {
ui.weak("authentication failed — set the nornir-server token, then press ⟳ Refresh");
}
Conn::Unreachable => {
ui.weak(format!("could not reach nornir-server at {endpoint} — press ⟳ Refresh"));
}
Conn::ServerError => {
ui.weak(format!("nornir-server at {endpoint} returned an error — press ⟳ Refresh"));
}
_ => {
ui.weak("no funnel to render");
}
});
}
}
}
fn state_json(&self) -> serde_json::Value {
// The DAG currently shown: the selected past-run snapshot, else the live view.
let plandag = match self.current_view() {
Some(v) => Facet::state_json(v),
None => serde_json::json!({ "node_count": 0, "edge_count": 0, "nodes": [] }),
};
// Bottleneck annotation, CONSUMED from that view's constraint analysis.
let (bottleneck, bottleneck_title) = match self.bottleneck_hint() {
Some((id, title)) => (Some(id), Some(title)),
None => (None, None),
};
let critical_path = plandag
.get("critical_path")
.cloned()
.unwrap_or(serde_json::Value::Null);
let (submit_status, submit_id, submit_error) = match &self.last_submit {
Some(Ok(id)) => ("ok", Some(id.clone()), None),
Some(Err(e)) => ("error", None, Some(e.clone())),
None => ("none", None, None),
};
serde_json::json!({
"kind": "funnel",
// FAT/embedded (in-process, no server) vs thin/remote (gRPC to nornir-server).
"mode": if self.embedded.is_some() { "embedded" } else { "remote" },
// PROVENANCE of the rendered content — `live: …` (the server's real
// funnel via the host-injected source) vs `embedded (…)` (in-process,
// with any live outage NAMED). The warehouse-nav `skeleton_source`
// vocabulary, applied to the funnel.
"funnel_source": self.funnel_source(),
// The injected live leg's own state: none | pending | live | one of the
// five distinct failure tokens (the outage, while embedded is shown).
"live_status": self.live_key(),
"live_error": self.live_error,
"endpoint": self.endpoint,
"workspace": self.workspace,
"status": self.conn_key(),
"node_count": self.node_count,
"edge_count": self.edge_count,
"error": self.error,
"auto_poll": self.auto_poll,
"poll_interval_s": self.poll_interval_s,
"draft": self.draft,
"intake_kind": match self.intake_kind { IntakeKind::Idea => "idea", IntakeKind::Error => "error" },
"submit_status": submit_status,
"submit_id": submit_id,
"submit_error": submit_error,
"history": self.history.to_json(),
"history_len": self.history.len(),
"history_peak_nodes": self.history.peak_nodes(),
"node_growth": self.history.node_growth(),
// RUN HISTORY (the run dimension of "addera historik"): past runs + which is shown.
"runs": self.runs.to_json(),
"run_count": self.runs.len(),
"selected_run": self.selected_run,
// EVENT-SOURCED run history — funnel-core's OWN `RunRecorded` ledger,
// projected by `Funnel::runs_to_json` (newest first). Embedded mode only:
// the remote `Funnel.Show` RPC returns the current funnel with no run
// ledger on the wire, so a thin pane has no events to fold. Empty (not
// absent) there, so the seam shape is the same in both modes.
"event_runs": self.event_runs(),
"event_run_count": self.event_run_count(),
// The Claude credential SOURCE, as reported by the host (never a value):
// the funnel drives Claude to expand ideas, so the pane shows whether a
// credential is configured and points at Settings when it is not.
"claude_source": self.claude_source,
"claude_configured": self.claude_source != "none",
// BOTTLENECK annotation (consumed from facett-plandag's constraint analysis).
"bottleneck": bottleneck,
"bottleneck_title": bottleneck_title,
"critical_path": critical_path,
"plandag": plandag,
// Which render of the DAG is on screen, and what the GRAPH render drew:
// node/edge counts, the measured frame cost, and node_cap:null (no
// MAX_NODES ceiling, unlike nornir's CallGraph pane).
"dag_view": self.dag_view.key(),
"graph": self.graph.report(),
// The 3D layered render: whether it's the active view, node/edge counts, the
// live camera azimuth (rotate_angle), the rotate toggle (spin — ON by
// default), the labels toggle + the exact label TEXTS painted (the idea
// words attached in 3D), the lane histogram, and the drilled selection —
// all AS DATA so a robot asserts the 3D render.
"graph3d_on": self.dag_view == FunnelDagView::Graph3d,
"graph3d": self.graph3d.report(),
})
}
fn update_json(&mut self, msg_json: &str) {
// The pane's own control surface (drives it headless from a robot test):
// {"set_workspace": "…"} → bind to a selected warehouse + refetch
// {"set_token": "…"} → set the bearer token + refetch
// {"refresh": true} / {"poll": true} → force a re-fetch
// {"auto_poll": bool} → toggle live polling
// {"draft": "…"} → set the submit-strip text
// {"intake_kind": "idea"|"error"} → set the intake kind
// {"submit": true} → submit the current draft (WRITE)
// {"clear_history": true} → clear the growth-history trail
// {"select_run": <seq>|null} → view a past run (null ⇒ live/latest)
// {"clear_runs": true} → drop the run history + back to live
// {"dag_view": "cards"|"graph"|"graph3d"} → switch the DAG render
// {"graph3d_rotate": bool} → auto-spin the 3D camera (projection-only)
// {"graph3d_labels": bool} → the idea words beside each 3D node
// {"graph3d_select": "<node-id>"} → drill: select a 3D node → card detail
// {"claude_source": "…"} → the host's resolved Claude credential
// SOURCE label (never a key value)
// {"submit_text": "…", ["submit_kind": "idea"|"error"]}
// → set draft(+kind) then submit
// Anything else is forwarded to the inner Plan-DAG facet (its own JSON seam).
if let Ok(v) = serde_json::from_str::<serde_json::Value>(msg_json) {
// Warehouse selection: the host's selector routes the chosen name here so
// the funnel re-binds to the SELECTED warehouse (nornir's per-tab
// set_workspace, mirrored). An empty string is the valid "none" state.
if let Some(ws) = v.get("set_workspace").and_then(|r| r.as_str()) {
self.set_workspace(ws);
return;
}
// The host's resolved Claude credential SOURCE (a label, never a key).
if let Some(src) = v.get("claude_source").and_then(|r| r.as_str()) {
self.set_claude_source(src);
return;
}
// Switch the DAG render: card lane ↔ flat graph ↔ 3D layered graph.
if let Some(dv) = v
.get("dag_view")
.and_then(|r| r.as_str())
.and_then(FunnelDagView::parse)
{
self.dag_view = dv;
return;
}
// Toggle the 3D graph's camera auto-spin (projection-only; layout stays cached).
if let Some(on) = v.get("graph3d_rotate").and_then(|r| r.as_bool()) {
self.graph3d.set_rotate(on);
return;
}
// Toggle the 3D graph's per-node labels (the idea words attached in 3D).
if let Some(on) = v.get("graph3d_labels").and_then(|r| r.as_bool()) {
self.graph3d.set_labels(on);
return;
}
// Drill: select a 3D node by id (the headless twin of a node click) and
// forward it to the SAME card view's selection, so switching to Cards shows
// that card's detail — the drill reuses the existing card detail.
if let Some(id) = v.get("graph3d_select").and_then(|r| r.as_str()) {
if let Some(sel) = self.graph3d.select_id(id) {
if let Some(vw) = self.view.as_mut() {
// Prefixed to the card view's `node:<id>` resolved key (the drill).
vw.update(facett_plandag::Msg::Select(format!("node:{sel}")));
}
}
return;
}
if let Some(tok) = v.get("set_token").and_then(|r| r.as_str()) {
self.set_token(tok);
return;
}
if let Some(on) = v.get("auto_poll").and_then(|r| r.as_bool()) {
self.auto_poll = on;
return;
}
if let Some(k) = v.get("intake_kind").and_then(|r| r.as_str()) {
self.intake_kind = if k.eq_ignore_ascii_case("error") {
IntakeKind::Error
} else {
IntakeKind::Idea
};
// fall through — a message may also set the draft / submit.
}
if let Some(t) = v.get("draft").and_then(|r| r.as_str()) {
self.draft = t.to_string();
}
if let Some(t) = v.get("submit_text").and_then(|r| r.as_str()) {
self.draft = t.to_string();
if let Some(k) = v.get("submit_kind").and_then(|r| r.as_str()) {
self.intake_kind = if k.eq_ignore_ascii_case("error") {
IntakeKind::Error
} else {
IntakeKind::Idea
};
}
self.submit_draft();
return;
}
if v.get("submit").and_then(|r| r.as_bool()) == Some(true) {
self.submit_draft();
return;
}
if v.get("clear_history").and_then(|r| r.as_bool()) == Some(true) {
self.history.clear();
return;
}
// Run picker: {"select_run": <seq>} views a past run; {"select_run": null}
// (or false) returns to the live/latest view.
if let Some(sr) = v.get("select_run") {
if sr.is_null() || sr.as_bool() == Some(false) {
self.select_run(None);
} else if let Some(seq) = sr.as_u64() {
self.select_run(Some(seq as usize));
}
return;
}
if v.get("clear_runs").and_then(|r| r.as_bool()) == Some(true) {
self.select_run(None);
self.runs.clear();
return;
}
if v.get("refresh").and_then(|r| r.as_bool()) == Some(true)
|| v.get("poll").and_then(|r| r.as_bool()) == Some(true)
{
// The SCRIPTED seam is synchronous by contract: a driven refresh is
// complete when this returns, so a robot assertion right after it sees
// the consult's result. Only the egui frame goes through the
// non-blocking `refresh` + `poll_live` pair.
self.refresh_blocking();
return;
}
// A message that only set draft/intake_kind (no submit) is handled.
if v.get("draft").is_some() || v.get("intake_kind").is_some() {
return;
}
}
// Forward anything else to the CURRENTLY-SHOWN Plan-DAG facet (the selected
// snapshot when a past run is picked, else the live view) — its own JSON seam.
if let Some(v) = self.current_view_mut() {
Facet::update_json(v, msg_json);
}
}
}
/// **Resolve the nornir-server bearer token** the way the `nornir` CLI does: prefer
/// `$NORNIR_SERVER_TOKEN`, then fall back to the user-readable `~/.nornir/token`
/// (trimmed). Empty ⇒ no auth. Exposed so a host can seed the same token the CLI uses
/// without re-implementing the ladder.
///
/// Folds onto the SHARED `nornir_service::resolve_token` env→file→open ladder (Law #5 —
/// the one helper korp/nornir/dwarves reuse), parameterised on nornir's env var + token
/// file. No `--token` flag here (a host seeds this programmatically), so the flag arm is
/// `None`.
pub fn resolve_token() -> String {
let token_file =
std::env::var_os("HOME").map(|h| std::path::Path::new(&h).join(".nornir").join("token"));
nornir_service::resolve_token(None, "NORNIR_SERVER_TOKEN", token_file.as_deref())
}
/// **The backend-(2) seam.** Build the live funnel pane as a `Box<dyn Panel>` a host
/// drops into a tab. `endpoint` is the `nornir-server` url (e.g.
/// `http://127.0.0.1:7878`). The bearer token is resolved by [`resolve_token`]
/// (`$NORNIR_SERVER_TOKEN` → `~/.nornir/token`) and the served workspace from
/// `$NORNIR_WORKSPACE` (empty ⇒ no warehouse selected — the pane shows *"no warehouse
/// selected"*, NOT "offline"). The first fetch happens eagerly; every failure yields
/// a clean, DISTINCT state (no-warehouse / unreachable / auth-failed / not-served),
/// never a panic. A host with a warehouse selector should call
/// [`FunnelPanel::set_workspace`] to bind the funnel to the selected warehouse.
pub fn panel(endpoint: impl Into<String>) -> Box<dyn Panel> {
let token = resolve_token();
let workspace = std::env::var("NORNIR_WORKSPACE").unwrap_or_default();
Box::new(FunnelPanel::new(endpoint.into(), token, workspace))
}
/// Like [`panel`] but with the token + workspace passed explicitly (for hosts that
/// manage their own credentials rather than reading the environment).
pub fn panel_with(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Box<dyn Panel> {
Box::new(FunnelPanel::new(
endpoint.into(),
token.into(),
workspace.into(),
))
}
/// **The FAT / EMBEDDED seam (the default).** Build the funnel pane in-process over the
/// local `workspace` — **no `nornir-server`**. The pane renders + polls its own
/// Plan-DAG with no server running and grows it as ideas are submitted, all offline
/// (the fat-mode doctrine: dwarves is self-contained; nornir-server can be DOWN). A
/// host only reaches for [`panel`]/[`panel_with`] when an explicit remote endpoint is
/// configured.
pub fn panel_embedded(workspace: impl Into<String>) -> Box<dyn Panel> {
Box::new(FunnelPanel::new_embedded(workspace.into()))
}
/// [`panel_embedded`] over the SAMPLE funnel — tests only (`testkit`).
#[cfg(any(test, feature = "testkit"))]
pub fn panel_embedded_seeded(workspace: impl Into<String>) -> Box<dyn Panel> {
Box::new(FunnelPanel::new_embedded_seeded(workspace.into()))
}
/// **The EMBEDDED + LIVE seam — the shipped host wiring.** The pane is built
/// FAT/embedded exactly like [`panel_embedded`] (offline-first, immediate, nothing
/// dialed at construction), with the host's endpoint + token injected as a
/// [`GrpcFunnelSource`] that the refresh/poll path consults LAZILY: when the host
/// serves `/nornir.v1.Funnel/Show`, the tab renders the server's REAL,
/// `funnel_events`-backed funnel (so an idea submitted over the MCP write leg becomes
/// VISIBLE here) and routes the submit strip over the same wire; a dead/denied/
/// not-serving host keeps the embedded state and NAMES the outage
/// (`state_json.live_status` + `funnel_source`) — honest fallback, never fabrication.
pub fn panel_embedded_live(endpoint: &str, token: &str, workspace: &str) -> Box<dyn Panel> {
Box::new(
FunnelPanel::new_embedded(workspace)
.with_live_source(Box::new(GrpcFunnelSource::new(endpoint, token))),
)
}
/// **The reusable MOCK `Funnel` gRPC server** (feature `testkit`) — a REAL tonic
/// `nornir.v1.Funnel` service on an ephemeral port whose in-memory store `SubmitIdea`
/// grows and `Show` returns, so a host's robot tests prove submit→visible over the
/// actual wire (never a stub of this crate's own client). Mounted through the
/// warehouse testkit's ONE `spawn_router` body (L5 — no second thread/runtime/wait
/// twin).
#[cfg(feature = "testkit")]
pub mod testkit {
use std::sync::{Arc, Mutex};
use crate::pb;
/// The running-mock handle (endpoint + shutdown-on-drop) — the SAME type the
/// warehouse testkit hands out, reused rather than twinned.
pub use nornir_warehouse_embed::testkit::MockWarehouseServer as MockFunnelServer;
/// An in-memory `nornir.v1.Funnel`: `Show` returns the current store; `SubmitIdea`
/// appends a plan-less idea (`i-001`, `i-002`, …) exactly as the live server's
/// intake does before any decompose — so a submitted idea round-trips the wire and
/// comes back visible in the very next `Show`. The plan-write trio (`CreatePlan` /
/// `AddNode` / `Link`) grows plans under those ideas with the live server's
/// sequential-id + needs→deps semantics, so a host's tests prove
/// decompose→persist→visible over the actual wire.
#[derive(Clone, Default)]
pub struct MockFunnel {
ideas: Arc<Mutex<Vec<pb::FunnelDumpIdea>>>,
}
#[tonic::async_trait]
impl pb::funnel_server::Funnel for MockFunnel {
async fn show(
&self,
_req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::FunnelDump>, tonic::Status> {
Ok(tonic::Response::new(pb::FunnelDump {
ideas: self.ideas.lock().unwrap().clone(),
}))
}
async fn submit_idea(
&self,
req: tonic::Request<pb::SubmitIdeaRequest>,
) -> Result<tonic::Response<pb::IdRef>, tonic::Status> {
let r = req.into_inner();
if r.text.trim().is_empty() {
// The live server rejects an empty intake; the mock must too, so a
// test cannot green on a swallowed no-op.
return Err(tonic::Status::invalid_argument("empty idea text"));
}
let mut ideas = self.ideas.lock().unwrap();
let id = format!("i-{:03}", ideas.len() + 1);
ideas.push(pb::FunnelDumpIdea {
id: id.clone(),
text: r.text,
plans: Vec::new(),
source: r.source,
item_kind: if r.item_kind.is_empty() {
"idea".into()
} else {
r.item_kind
},
});
Ok(tonic::Response::new(pb::IdRef { id }))
}
async fn create_plan(
&self,
req: tonic::Request<pb::CreatePlanRequest>,
) -> Result<tonic::Response<pb::IdRef>, tonic::Status> {
let r = req.into_inner();
let mut ideas = self.ideas.lock().unwrap();
// Live semantics: sequential plan id across the store; a plan on an
// unknown idea is REJECTED (funnel-core's apply bails), never dangled.
let id = format!(
"p-{:03}",
ideas.iter().map(|i| i.plans.len()).sum::<usize>() + 1
);
let Some(idea) = ideas.iter_mut().find(|i| i.id == r.idea_id) else {
return Err(tonic::Status::internal(format!(
"PlanCreated for unknown idea `{}`",
r.idea_id
)));
};
idea.plans.push(pb::FunnelDumpPlan {
id: id.clone(),
idea_id: r.idea_id,
summary: r.summary,
nodes: Vec::new(),
// The live server records PlanCreated + PlanStatusChanged(Active);
// Show carries the Debug repr.
status: "Active".into(),
});
Ok(tonic::Response::new(pb::IdRef { id }))
}
async fn add_node(
&self,
req: tonic::Request<pb::AddNodeRequest>,
) -> Result<tonic::Response<pb::IdRef>, tonic::Status> {
let r = req.into_inner();
let mut ideas = self.ideas.lock().unwrap();
let id = format!(
"n-{:03}",
ideas
.iter()
.flat_map(|i| &i.plans)
.map(|p| p.nodes.len())
.sum::<usize>()
+ 1
);
let Some(plan) = ideas
.iter_mut()
.flat_map(|i| &mut i.plans)
.find(|p| p.id == r.plan_id)
else {
return Err(tonic::Status::internal(format!(
"NodeAdded for unknown plan `{}`",
r.plan_id
)));
};
for need in &r.needs {
if !plan.nodes.iter().any(|n| &n.id == need) {
return Err(tonic::Status::internal(format!(
"EdgeAdded from unknown node `{need}`"
)));
}
}
plan.nodes.push(pb::FunnelDumpNode {
id: id.clone(),
kind: r.kind,
title: r.title,
status: "Pending".into(),
deps: r.needs,
});
Ok(tonic::Response::new(pb::IdRef { id }))
}
async fn link(
&self,
req: tonic::Request<pb::LinkRequest>,
) -> Result<tonic::Response<pb::Empty>, tonic::Status> {
let r = req.into_inner();
let mut ideas = self.ideas.lock().unwrap();
let Some(plan) = ideas
.iter_mut()
.flat_map(|i| &mut i.plans)
.find(|p| p.id == r.plan_id)
else {
return Err(tonic::Status::internal(format!(
"EdgeAdded for unknown plan `{}`",
r.plan_id
)));
};
if !plan.nodes.iter().any(|n| n.id == r.from) {
return Err(tonic::Status::internal(format!(
"EdgeAdded from unknown node `{}`",
r.from
)));
}
let Some(to) = plan.nodes.iter_mut().find(|n| n.id == r.to) else {
return Err(tonic::Status::internal(format!(
"EdgeAdded to unknown node `{}`",
r.to
)));
};
if !to.deps.contains(&r.from) {
to.deps.push(r.from);
}
Ok(tonic::Response::new(pb::Empty {}))
}
}
/// Spawn the mock `Funnel` server on an ephemeral port, pre-loaded with `seed`
/// ideas. Dropping the handle shuts it down.
pub fn spawn_mock_funnel(seed: Vec<pb::FunnelDumpIdea>) -> MockFunnelServer {
let mock = MockFunnel {
ideas: Arc::new(Mutex::new(seed)),
};
nornir_warehouse_embed::testkit::spawn_router(move |mut s| {
s.add_service(pb::funnel_server::FunnelServer::new(mock))
})
}
}
#[cfg(all(test, feature = "testkit"))]
mod persist_wire_tests {
//! The decompose PERSIST leg over the REAL wire: [`persist_plan`] drives
//! `SubmitIdea` + `CreatePlan` + `AddNode` + `Link` against the testkit's tonic
//! mock, and the plan's node TITLES come back out of `Funnel.Show` — applied
//! output through the socket, never a state round-trip.
use super::*;
fn spec(local: &str, kind: &str, title: &str, needs: &[&str]) -> PlanNodeSpec {
PlanNodeSpec {
local_id: local.into(),
kind: kind.into(),
title: title.into(),
needs: needs.iter().map(|s| s.to_string()).collect(),
}
}
/// RED-when-broken (proven by sabotage: dropping the `create_plan`/`add_node`
/// calls fails the title assertions): a persisted plan REAPPEARS in `Show` —
/// idea intaken, plan attached to it, both node titles present, and the edge
/// translated from local ids (`n1` → server `n-001`) into the dump's `deps`.
#[test]
fn persist_plan_reappears_in_show_with_titles_and_edges() {
let srv = testkit::spawn_mock_funnel(Vec::new());
let out = persist_plan(
&srv.endpoint,
"",
"nordisk",
None,
"add a cache to the store",
"cache plan",
&[
spec("n1", "code:write", "add LruCache to store.rs", &[]),
spec("n2", "test:run", "test cache hit/miss", &["n1"]),
],
)
.expect("persist over the wire");
assert_eq!(out.idea_id, "i-001", "the idea was intaken first");
assert_eq!(out.plan_id, "p-001");
// Applied output: the titles come back OUT of Funnel.Show over the wire.
let f = fetch_funnel(&srv.endpoint, "", "nordisk").expect("Show after persist");
let idea = f.ideas.values().next().expect("idea visible");
assert_eq!(idea.text, "add a cache to the store");
let plan = f
.plans
.values()
.find(|p| p.idea_id == idea.id)
.expect("plan attached");
assert_eq!(plan.summary, "cache plan");
let titles: Vec<&str> = plan
.nodes
.values()
.filter_map(|n| n.prompt_excerpt.as_deref())
.collect();
assert!(
titles.contains(&"add LruCache to store.rs") && titles.contains(&"test cache hit/miss"),
"both node titles round-trip Show: {titles:?}"
);
// The n1→n2 edge crossed as server ids.
assert_eq!(plan.edges.len(), 1, "the one dependency edge persisted");
// Attaching to the EXISTING idea (the id the first persist returned) must
// not intake a second idea.
let again = persist_plan(
&srv.endpoint,
"",
"nordisk",
Some(&out.idea_id),
"ignored — idea exists",
"follow-up plan",
&[spec("n1", "doc:update", "document the cache", &[])],
)
.expect("persist onto the existing idea");
assert_eq!(again.idea_id, out.idea_id);
let f = fetch_funnel(&srv.endpoint, "", "nordisk").expect("Show again");
assert_eq!(f.ideas.len(), 1, "no duplicate idea intaken");
assert_eq!(
f.plans.len(),
2,
"the second plan attached to the same idea"
);
}
/// A malformed spec is refused BEFORE the first write: dangling `needs` (and an
/// empty plan) bail with an honest error and the store stays untouched — no
/// half-persisted plan from a bad model reply.
#[test]
fn malformed_plan_is_refused_before_any_write() {
let srv = testkit::spawn_mock_funnel(Vec::new());
let err = persist_plan(
&srv.endpoint,
"",
"nordisk",
None,
"idea",
"summary",
&[spec("n1", "code:write", "something", &["n-does-not-exist"])],
)
.expect_err("dangling needs must refuse");
assert!(format!("{err:#}").contains("unknown sibling"), "{err:#}");
assert!(
persist_plan(&srv.endpoint, "", "nordisk", None, "idea", "s", &[]).is_err(),
"an empty plan is refused"
);
let f = fetch_funnel(&srv.endpoint, "", "nordisk").expect("Show");
assert!(
f.ideas.is_empty() && f.plans.is_empty(),
"nothing was written"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
// ── the EMBEDDED + LIVE seam (host-injected read leg) ────────────────────
/// A fake [`LiveFunnelSource`]: `up` serves a funnel holding `ideas` (grown by
/// `submit`, so a submit→fetch round-trip is observable); `dead` fails every
/// consult with the connect-shaped error a dead server produces. No network.
struct FakeLive {
up: bool,
ideas: std::sync::Mutex<Vec<String>>,
}
impl FakeLive {
fn up(texts: &[&str]) -> Self {
Self {
up: true,
ideas: std::sync::Mutex::new(texts.iter().map(|s| s.to_string()).collect()),
}
}
fn dead() -> Self {
Self {
up: false,
ideas: std::sync::Mutex::new(Vec::new()),
}
}
}
impl LiveFunnelSource for FakeLive {
fn fetch(&self, _workspace: &str) -> Result<funnel_core::state::Funnel> {
if !self.up {
return Err(anyhow::anyhow!("tcp connect: connection refused")
.context("connect to nornir-server at http://127.0.0.1:1"));
}
let ideas = self.ideas.lock().unwrap();
let dump = pb::FunnelDump {
ideas: ideas
.iter()
.enumerate()
.map(|(i, t)| pb::FunnelDumpIdea {
id: format!("i-{:03}", i + 1),
text: t.clone(),
plans: Vec::new(),
source: "test".into(),
item_kind: "idea".into(),
})
.collect(),
};
let mut f = funnel_from_dump(&dump);
f.promote_ready();
Ok(f)
}
fn submit(&self, _workspace: &str, text: &str, _kind: IntakeKind) -> Result<String> {
if !self.up {
anyhow::bail!("connect to nornir-server at http://127.0.0.1:1: refused");
}
let mut v = self.ideas.lock().unwrap();
v.push(text.to_string());
Ok(format!("i-{:03}", v.len()))
}
fn describe(&self) -> String {
"fake live funnel".into()
}
}
/// The rendered plan-DAG's node TITLES (applied output, not a count).
fn plandag_titles(st: &serde_json::Value) -> Vec<String> {
st["plandag"]["nodes"]
.as_array()
.expect("plandag nodes")
.iter()
.filter_map(|n| n["title"].as_str().map(|s| s.to_string()))
.collect()
}
/// RED-WHEN-BROKEN — the missing-read-leg fix. A healthy injected live source
/// REPLACES the embedded content with the server's funnel: the live idea's own
/// words render in the plan-DAG, the demo seed is gone, and provenance says
/// `live: …` / `status: live`. If `refresh` ignored the source (the original
/// defect: submitted ideas landed in a store no pane read), the seeded
/// "release engine" fixture would still be on screen and this fails.
#[test]
fn live_source_replaces_embedded_content_and_names_live_provenance() {
let panel = FunnelPanel::new_embedded_seeded("nordisk")
.with_live_source(Box::new(FakeLive::up(&["korp raven autopilot"])));
// LAZY: attaching dials nothing — provenance says so until the first poll.
let st = Facet::state_json(&panel);
assert_eq!(
st["live_status"], "pending",
"no consult before the first refresh: {st}"
);
assert_eq!(
st["status"], "embedded",
"embedded content until consulted: {st}"
);
let mut panel = panel;
panel.refresh_blocking();
let st = Facet::state_json(&panel);
assert_eq!(
st["status"], "live",
"the rendered funnel is the server's: {st}"
);
assert_eq!(st["live_status"], "live", "{st}");
assert_eq!(st["funnel_source"], "live: fake live funnel", "{st}");
assert!(st["live_error"].is_null(), "{st}");
let titles = plandag_titles(&st);
assert!(
titles.iter().any(|t| t.contains("korp raven autopilot")),
"the live idea's words render in the DAG: {titles:?}"
);
assert!(
!titles.iter().any(|t| t.contains("release engine")),
"the embedded demo seed is NOT shown while live: {titles:?}"
);
}
/// The honesty guard: a DEAD live source keeps the embedded state AND names the
/// outage — `live_status: unreachable`, `funnel_source: embedded (live
/// unreachable — …)`, the error text surfaced — never a fabricated funnel, never
/// a silent fallback.
#[test]
fn dead_live_source_keeps_embedded_state_and_names_the_outage() {
let mut panel = FunnelPanel::new_embedded_seeded("nordisk")
.with_live_source(Box::new(FakeLive::dead()));
panel.refresh_blocking();
let st = Facet::state_json(&panel);
assert_eq!(st["status"], "embedded", "embedded content is kept: {st}");
assert_eq!(
st["live_status"], "unreachable",
"the outage is NAMED: {st}"
);
let src = st["funnel_source"].as_str().unwrap();
assert!(
src.starts_with("embedded (live unreachable"),
"provenance names the outage: {src}"
);
assert!(
st["live_error"].as_str().is_some(),
"the error text is surfaced: {st}"
);
let titles = plandag_titles(&st);
assert!(
titles.iter().any(|t| t.contains("release engine")),
"the seeded embedded funnel still renders: {titles:?}"
);
assert!(
st["node_count"].as_u64().unwrap() > 0,
"never blank on an outage: {st}"
);
}
/// While the live leg is UP the submit strip rides the WIRE (the same write leg
/// MCP uses) and the refresh makes the idea VISIBLE — submit → visible in one
/// action. The fake source records the submit, so a locally-swallowed submit
/// (the embedded engine absorbing it) could not pass the title assertion.
#[test]
fn submit_routes_over_the_live_wire_and_lands_visible() {
let mut panel = FunnelPanel::new_embedded_seeded("nordisk")
.with_live_source(Box::new(FakeLive::up(&[])));
panel.refresh_blocking();
let st = Facet::state_json(&panel);
assert_eq!(
st["status"], "live_empty",
"live and empty before the submit: {st}"
);
Facet::update_json(&mut panel, r#"{"submit_text":"raven autopilot brain"}"#);
let st = Facet::state_json(&panel);
assert_eq!(st["submit_status"], "ok", "{st}");
assert_eq!(st["submit_id"], "i-001", "the wire's id came back: {st}");
assert_eq!(st["status"], "live", "{st}");
let titles = plandag_titles(&st);
assert!(
titles.iter().any(|t| t.contains("raven autopilot brain")),
"the submitted idea is VISIBLE in the rendered DAG: {titles:?}"
);
}
/// An empty workspace with a live source is the thin pane's NoWarehouse rule,
/// mirrored: NAMED (`live_status: no_warehouse`) and NOT dialed — the dead fake
/// would report `unreachable` if a dial had happened.
#[test]
fn live_source_with_no_workspace_is_named_not_dialed() {
let mut panel = FunnelPanel::new_embedded("").with_live_source(Box::new(FakeLive::dead()));
panel.refresh();
let st = Facet::state_json(&panel);
assert_eq!(
st["live_status"], "no_warehouse",
"named, and no dial happened: {st}"
);
// the pane stays on its in-process funnel, which holds nothing (no local log, no seed)
assert_eq!(
st["status"], "embedded_empty",
"the embedded funnel is what is on screen: {st}"
);
}
/// RED-WHEN-BROKEN — the fat-mode fix. With **NO remote nornir configured** the
/// pane must default to EMBEDDED, render its in-process DAG, and report an
/// embedded-mode state — NEVER an unreachable / "no remote nornir" error. If the
/// pane regresses to thin/remote-by-default this fails (it would report
/// `unreachable` / a null DAG).
#[test]
fn embedded_is_the_default_and_renders_without_a_server() {
let panel = FunnelPanel::new_embedded_seeded("nordisk");
assert!(
panel.is_embedded(),
"no server configured ⇒ FAT/embedded default"
);
let st = Facet::state_json(&panel);
assert_eq!(st["kind"], "funnel");
assert_eq!(
st["mode"], "embedded",
"must default to fat/embedded, not thin/remote"
);
// A real DAG is rendered with NO server running.
assert!(
st["node_count"].as_u64().unwrap() > 0,
"embedded DAG is non-empty: {st}"
);
let status = st["status"].as_str().unwrap();
assert!(
status.starts_with("embedded"),
"embedded status, got `{status}`"
);
for bad in [
"unreachable",
"auth_failed",
"not_served",
"server_error",
"no_warehouse",
] {
assert_ne!(
status, bad,
"the embedded default must NEVER report a remote-failure state"
);
}
assert!(
st["error"].is_null(),
"no error in the embedded default: {st}"
);
}
/// The offline WRITE leg: submitting an idea into the embedded funnel grows the
/// in-process DAG (no server) and is accepted locally.
#[test]
fn embedded_submit_grows_the_dag_offline() {
let mut panel = FunnelPanel::new_embedded_seeded("nordisk");
let before = Facet::state_json(&panel)["node_count"].as_u64().unwrap();
Facet::update_json(
&mut panel,
r#"{"submit_text":"an offline idea","submit_kind":"idea"}"#,
);
let st = Facet::state_json(&panel);
assert_eq!(
st["submit_status"], "ok",
"embedded submit accepted with no server: {st}"
);
let after = st["node_count"].as_u64().unwrap();
assert!(
after > before,
"the in-process DAG grew {before} -> {after}: {st}"
);
assert_eq!(st["mode"], "embedded");
}
/// The thin/remote path is still available when a real endpoint is explicitly
/// configured (a dead port ⇒ the honest `unreachable` state, NOT embedded).
#[test]
fn explicit_remote_endpoint_still_uses_thin_mode() {
let panel = FunnelPanel::new("http://127.0.0.1:0", "", "nordisk");
assert!(!panel.is_embedded());
let st = Facet::state_json(&panel);
assert_eq!(st["mode"], "remote");
assert_eq!(
st["status"], "unreachable",
"explicit remote + dead port ⇒ honest unreachable"
);
}
/// **A dead source keeps the honest fallback in the 3D render too.** With an
/// explicit remote endpoint on a dead port, the pane names the outage (`status:
/// unreachable`, an `error` naming the endpoint) and the 3D graph reports ZERO
/// nodes and ZERO label texts — it never fabricates a spinning graph out of
/// nothing. The embedded default, by contrast, seeds the honest local fixture
/// whose labels really carry idea words.
#[test]
fn dead_source_keeps_honest_fallback_in_3d() {
let mut panel = FunnelPanel::new("http://127.0.0.1:0", "", "nordisk");
Facet::update_json(&mut panel, r#"{"dag_view":"graph3d"}"#);
let st = Facet::state_json(&panel);
assert_eq!(st["mode"], "remote");
assert_eq!(
st["status"], "unreachable",
"the outage is NAMED, not papered over: {st}"
);
let err = st["error"].as_str().expect("the error text is surfaced");
assert!(
err.contains("nornir-server"),
"the error names the dead source: {err}"
);
// The 3D report is honestly empty — no fabricated nodes, no fabricated words.
assert_eq!(st["graph3d"]["nodes"], 0, "no fabricated 3D nodes: {st}");
assert_eq!(
st["graph3d"]["label_texts"].as_array().unwrap().len(),
0,
"no fabricated words: {st}"
);
// The embedded default is the honest local fallback: real seeded idea words.
let emb = FunnelPanel::new_embedded_seeded("nordisk");
let st = Facet::state_json(&emb);
assert_eq!(st["mode"], "embedded");
let words: Vec<String> = st["graph3d"]["label_texts"]
.as_array()
.unwrap()
.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect();
assert!(
words.iter().any(|w| w.contains("release engine")),
"the embedded fallback's 3D labels carry the seeded idea's words: {words:?}"
);
assert_eq!(
st["graph3d"]["labels"], true,
"labels are ON by default: {st}"
);
assert_eq!(
st["graph3d"]["rotate"], true,
"the 3D graph spins by default: {st}"
);
}
/// A `Funnel.Show` dump reconstructs into a `funnel_core::Funnel` faithful enough
/// that the pure Plan-DAG projects the identical topology the local path builds —
/// the crux of the thin/remote wiring (mirrors nornir's reconstruct test, but
/// against THIS crate's re-implementation + generated `pb`).
#[test]
fn dump_reconstructs_into_a_projectable_funnel() {
let dump = pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-001".into(),
text: "build the planner".into(),
source: "human:alice".into(),
item_kind: "idea".into(),
plans: vec![pb::FunnelDumpPlan {
id: "p-001".into(),
idea_id: "i-001".into(),
summary: "decompose".into(),
status: "Active".into(), // server ships the Debug repr
nodes: vec![
pb::FunnelDumpNode {
id: "n-001".into(),
kind: "code:write".into(),
title: "implement foo".into(),
status: "done".into(),
deps: vec![],
},
pb::FunnelDumpNode {
id: "n-002".into(),
kind: "code:fix".into(),
title: "implement bar".into(),
status: "in_progress".into(),
deps: vec!["n-001".into()],
},
pb::FunnelDumpNode {
id: "n-003".into(),
kind: "test:write".into(),
title: "test baz".into(),
status: "pending".into(),
deps: vec!["n-002".into(), "n-001".into()],
},
],
}],
}],
};
let f = funnel_from_dump(&dump);
assert_eq!(f.ideas.len(), 1);
assert_eq!(f.plans.len(), 1);
let plan = f.plans.values().next().unwrap();
assert_eq!(plan.status, funnel_core::event::PlanStatus::Active);
assert_eq!(plan.nodes.len(), 3);
use funnel_core::ids::NodeId;
assert_eq!(
plan.nodes[&NodeId::new("n-001")].status,
funnel_core::event::NodeStatus::Done
);
assert_eq!(
plan.nodes[&NodeId::new("n-002")].status,
funnel_core::event::NodeStatus::InProgress
);
assert_eq!(plan.edges.len(), 3);
// The pure Plan-DAG projects it: mother idea + 3 task nodes.
let dag = PlanDag::from_funnel(&f, None);
assert_eq!(dag.node_count(), 4, "mother + 3 task cards");
assert_eq!(dag.dep_edges(), 3, "the reconstructed dependency edges");
assert_eq!(
dag.mother_edges(),
3,
"one decomposition edge per task node"
);
let order = dag.topo_order().expect("acyclic");
let pos = |id: &str| order.iter().position(|x| x == id).unwrap();
assert!(pos("n-001") < pos("n-002"));
assert!(pos("n-002") < pos("n-003"));
}
/// A partial/edge-to-missing-node dump never panics (defensive reconstruction).
#[test]
fn dump_with_dangling_dep_is_tolerated() {
let dump = pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-002".into(),
text: "x".into(),
source: "".into(),
item_kind: "".into(),
plans: vec![pb::FunnelDumpPlan {
id: "p-002".into(),
idea_id: "".into(),
summary: "s".into(),
status: "weird-unknown".into(),
nodes: vec![pb::FunnelDumpNode {
id: "n-010".into(),
kind: "code:write".into(),
title: "".into(),
status: "".into(),
deps: vec!["n-999".into()],
}],
}],
}],
};
let f = funnel_from_dump(&dump);
let plan = f.plans.values().next().unwrap();
assert_eq!(
plan.idea_id.as_str(),
"i-002",
"empty idea_id falls back to the enclosing idea"
);
assert_eq!(plan.status, funnel_core::event::PlanStatus::Draft);
assert!(plan.edges.is_empty(), "the dangling n-999 edge is dropped");
assert_eq!(plan.nodes.len(), 1);
}
/// The reconstructed funnel maps onto the `facett-plandag` model and exposes the
/// same robot-observable shape (node count + per-node cost-heat AS DATA).
#[test]
fn reconstructed_funnel_maps_onto_plandag_model() {
let dump = pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-001".into(),
text: "planner".into(),
source: "".into(),
item_kind: "idea".into(),
plans: vec![pb::FunnelDumpPlan {
id: "p-001".into(),
idea_id: "i-001".into(),
summary: "decompose".into(),
status: "Active".into(),
nodes: vec![
pb::FunnelDumpNode {
id: "n-001".into(),
kind: "code:write".into(),
title: "foo".into(),
status: "done".into(),
deps: vec![],
},
pb::FunnelDumpNode {
id: "n-002".into(),
kind: "code:write".into(),
title: "bar".into(),
status: "in_progress".into(),
deps: vec!["n-001".into()],
},
],
}],
}],
};
let f = funnel_from_dump(&dump);
let dag = PlanDag::from_funnel(&f, None);
let view = view_from_dag(&dag);
let st = Facet::state_json(&view);
// mother + 2 task nodes.
assert_eq!(st["node_count"], 3);
for n in st["nodes"].as_array().unwrap() {
let h = n["heat"].as_f64().unwrap();
assert!(
(0.0..=1.0).contains(&h),
"cost-heat rides as data in [0,1]: {h}"
);
}
}
/// **DISTINCT failure states (the funnel-offline mislabel fix).** With a warehouse
/// SELECTED but the server on a dead port, the pane is cleanly `unreachable` (NOT
/// the old "offline" lie); with NO warehouse selected (empty workspace) it is
/// `no_warehouse` and never even touches the network. RED-when-broken: collapse the
/// two into one catch-all and one of these assertions fails.
#[test]
fn distinct_offline_and_no_warehouse_states() {
// A warehouse is selected, but the server is unreachable (dead port).
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "nordisk");
assert_eq!(
p.conn,
Conn::Unreachable,
"selected + dead port ⇒ unreachable, not 'offline'"
);
let st = Facet::state_json(&p);
assert_eq!(st["status"], "unreachable");
assert_eq!(st["node_count"], 0);
assert!(
st["error"].is_string(),
"the error is captured for the hover tooltip"
);
p.refresh();
assert_eq!(
p.conn,
Conn::Unreachable,
"a still-dead server stays cleanly unreachable"
);
// NO warehouse selected (empty workspace): not a server problem at all.
let np = FunnelPanel::new("http://127.0.0.1:0", "", "");
assert_eq!(
np.conn,
Conn::NoWarehouse,
"empty workspace ⇒ no_warehouse (no network touched)"
);
assert_eq!(Facet::state_json(&np)["status"], "no_warehouse");
assert!(
Facet::state_json(&np)["error"].is_null(),
"no-warehouse is not an error state"
);
}
/// **`classify_conn` maps each gRPC failure to the RIGHT distinct state** — the
/// crux of the mislabel fix, unit-tested without a server. RED-when-broken: mis-map
/// any code (e.g. Unauthenticated → ServerError) and an assert fails.
#[test]
fn classify_conn_distinguishes_the_failure_modes() {
let auth = anyhow::Error::from(tonic::Status::unauthenticated(
"missing or invalid bearer token",
))
.context("Funnel.Show RPC");
assert_eq!(classify_conn(&auth), Conn::AuthFailed);
let notfound = anyhow::Error::from(tonic::Status::not_found(
"workspace `nordisk` is not served",
))
.context("Funnel.Show RPC");
assert_eq!(classify_conn(¬found), Conn::NotServed);
let internal =
anyhow::Error::from(tonic::Status::internal("boom")).context("Funnel.Show RPC");
assert_eq!(classify_conn(&internal), Conn::ServerError);
let connect = anyhow::anyhow!("some io error")
.context("connect to nornir-server at http://127.0.0.1:0");
assert_eq!(classify_conn(&connect), Conn::Unreachable);
facett_core::testmatrix::emit(
"nornir-funnel-embed",
"classify_conn_distinguishes_failure_modes",
classify_conn(&auth) == Conn::AuthFailed
&& classify_conn(¬found) == Conn::NotServed
&& classify_conn(&connect) == Conn::Unreachable,
"the funnel names auth-failed vs not-served vs unreachable vs server-error distinctly — never the 'offline' mislabel when the server is up",
);
}
/// **`set_workspace` binds the pane to a selected warehouse** (nornir's per-tab
/// set_workspace, mirrored): selecting a name re-fetches for THAT warehouse;
/// selecting the empty name returns to the clean no_warehouse state. Drives the
/// dead-port pane so no real server is needed.
#[test]
fn set_workspace_rebinds_and_refetches() {
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "");
assert_eq!(p.conn, Conn::NoWarehouse);
// Select a warehouse over the JSON seam → binds + re-fetches (dead port ⇒ unreachable).
Facet::update_json(&mut p, r#"{"set_workspace":"nordisk"}"#);
assert_eq!(p.workspace(), "nordisk");
assert_eq!(Facet::state_json(&p)["workspace"], "nordisk");
assert_eq!(
p.conn,
Conn::Unreachable,
"a bound-but-unreachable warehouse ⇒ unreachable"
);
// Deselect (empty) → back to no_warehouse, no network.
Facet::update_json(&mut p, r#"{"set_workspace":""}"#);
assert_eq!(p.conn, Conn::NoWarehouse);
}
#[test]
fn poll_due_respects_interval_and_toggle() {
let p = FunnelPanel::new("http://127.0.0.1:0", "", ""); // offline; auto_poll on by default
assert!(p.auto_poll);
// last_poll_s starts at 0.0, interval = DEFAULT_POLL_INTERVAL_S (1.5s).
assert!(!p.poll_due(0.5), "not due before one interval");
assert!(
p.poll_due(DEFAULT_POLL_INTERVAL_S),
"due at exactly one interval"
);
assert!(p.poll_due(9.0), "due well past the interval");
let mut p = p;
p.set_auto_poll(false);
assert!(!p.poll_due(9.0), "never due while paused");
}
#[test]
fn update_json_toggles_auto_poll_in_state() {
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "");
Facet::update_json(&mut p, r#"{"auto_poll": false}"#);
assert_eq!(Facet::state_json(&p)["auto_poll"], serde_json::json!(false));
Facet::update_json(&mut p, r#"{"auto_poll": true}"#);
assert_eq!(Facet::state_json(&p)["auto_poll"], serde_json::json!(true));
assert_eq!(
Facet::state_json(&p)["poll_interval_s"],
serde_json::json!(DEFAULT_POLL_INTERVAL_S)
);
}
/// Build a `Funnel.Show` dump with one idea/plan carrying `task_titles` task
/// nodes wired in a chain (each depends on the previous). Lets a test grow the
/// plan by handing successively longer title lists.
fn dump_with_tasks(task_titles: &[&str]) -> pb::FunnelDump {
let nodes: Vec<pb::FunnelDumpNode> = task_titles
.iter()
.enumerate()
.map(|(i, t)| pb::FunnelDumpNode {
id: format!("n-{:03}", i + 1),
kind: "code:write".into(),
title: (*t).into(),
status: if i == 0 { "done" } else { "pending" }.into(),
deps: if i == 0 {
vec![]
} else {
vec![format!("n-{:03}", i)]
},
})
.collect();
pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-001".into(),
text: "grow the plan".into(),
source: "human:alice".into(),
item_kind: "idea".into(),
plans: vec![pb::FunnelDumpPlan {
id: "p-001".into(),
idea_id: "i-001".into(),
summary: "decompose".into(),
status: "Active".into(),
nodes,
}],
}],
}
}
/// Project a dump into the pane's Plan-DAG node count (mother idea + task
/// nodes) — the exact number the live pane renders + folds into `state_json`.
fn plandag_node_count(dump: &pb::FunnelDump) -> usize {
let f = funnel_from_dump(dump);
PlanDag::from_funnel(&f, None).node_count()
}
/// **The DAG GROWS as nodes land** (super-priority C): each successive
/// `Funnel.Show` poll that carries more task nodes reconstructs into a
/// strictly larger Plan-DAG — the exact behaviour the pane's auto-poll drives
/// on `dwarf`. Records ONE test-matrix cell for the growth invariant.
#[test]
fn plandag_grows_as_nodes_land() {
// 1 task → 2 tasks → 4 tasks. node_count = mother(1) + tasks.
let counts: Vec<usize> = [&["a"][..], &["a", "b"][..], &["a", "b", "c", "d"][..]]
.iter()
.map(|titles| plandag_node_count(&dump_with_tasks(titles)))
.collect();
assert_eq!(
counts,
vec![2, 3, 5],
"mother + N task nodes as the plan grows"
);
let monotonic = counts.windows(2).all(|w| w[1] > w[0]);
assert!(
monotonic,
"each poll with more nodes renders a strictly larger DAG: {counts:?}"
);
facett_core::testmatrix::emit(
"nornir-funnel-embed",
"plandag_grows_as_nodes_land",
monotonic && counts.first() == Some(&2) && counts.last() == Some(&5),
"successive Funnel.Show polls with more task nodes reconstruct into a strictly larger Plan-DAG (the pane's auto-poll growth)",
);
}
/// The submit strip's default state is robot-observable and empty: idea kind,
/// blank draft, no submit yet — the shape a robot asserts before typing.
#[test]
fn submit_strip_default_state_is_observable() {
let p = FunnelPanel::new("http://127.0.0.1:0", "", "");
let st = Facet::state_json(&p);
assert_eq!(st["draft"], "");
assert_eq!(st["intake_kind"], "idea");
assert_eq!(st["submit_status"], "none");
assert!(st["submit_id"].is_null());
}
/// Setting the draft + intake kind through the JSON seam (no keyboard) lands in
/// `state_json` — the headless path a robot uses to "type" into the pane.
#[test]
fn update_json_sets_draft_and_kind() {
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "");
Facet::update_json(
&mut p,
r#"{"draft": "make releases atomic", "intake_kind": "error"}"#,
);
let st = Facet::state_json(&p);
assert_eq!(st["draft"], "make releases atomic");
assert_eq!(st["intake_kind"], "error");
assert_eq!(
st["submit_status"], "none",
"setting the draft does not submit"
);
}
/// A blank draft submit is a no-op (never hits the network, never records an
/// outcome) — the disabled-button behaviour, verified headless.
#[test]
fn blank_submit_is_a_noop() {
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "");
p.submit_draft();
assert_eq!(Facet::state_json(&p)["submit_status"], "none");
Facet::update_json(&mut p, r#"{"draft": " "}"#); // whitespace only
Facet::update_json(&mut p, r#"{"submit": true}"#);
assert_eq!(
Facet::state_json(&p)["submit_status"],
"none",
"whitespace-only draft is blank"
);
}
/// **The WRITE leg is offline-safe.** Submitting against a dead server records a
/// clean `error` outcome (never a panic) and PRESERVES the draft so the typed
/// text isn't lost — the same non-panic discipline as the read path. Records the
/// test-matrix cell for the funnel write leg.
#[test]
fn submit_offline_is_clean_error_and_keeps_draft() {
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "nordisk");
Facet::update_json(
&mut p,
r#"{"submit_text": "grow the plan-DAG", "submit_kind": "idea"}"#,
);
let st = Facet::state_json(&p);
assert_eq!(
st["submit_status"], "error",
"offline submit → clean error, no panic"
);
assert!(
st["submit_error"].is_string(),
"the error is captured for the hover tooltip"
);
assert_eq!(
st["draft"], "grow the plan-DAG",
"a failed submit keeps the typed text"
);
// The pane stays usable (still cleanly unreachable, not wedged).
assert_eq!(st["status"], "unreachable");
facett_core::testmatrix::emit(
"nornir-funnel-embed",
"funnel_write_submit_idea_offline_safe",
st["submit_status"] == "error" && st["draft"] == "grow the plan-DAG",
"typing an idea/error and submitting calls Funnel.SubmitIdea over gRPC; an offline server → clean error, draft preserved, no panic (the WRITE leg)",
);
}
/// **The growth history de-dups + caps.** Consecutive identical polls collapse
/// (the trail records only *changes*), and the ring never exceeds `HISTORY_CAP`.
/// RED-when-broken: drop the de-dup and `len` jumps to 5; drop the cap and it
/// exceeds `HISTORY_CAP`.
#[test]
fn history_records_growth_dedups_and_caps() {
let mut h = FunnelHistory::default();
// Two identical polls collapse; then two growth steps.
assert!(h.record(2, 1));
assert!(!h.record(2, 1), "an identical consecutive poll collapses");
assert!(h.record(3, 2));
assert!(!h.record(3, 2));
assert!(h.record(5, 4));
assert_eq!(h.len(), 3, "only the change points are kept");
assert_eq!(
h.samples().iter().map(|s| s.node_count).collect::<Vec<_>>(),
vec![2, 3, 5]
);
assert_eq!(h.peak_nodes(), 5);
assert_eq!(h.node_growth(), 3, "net growth from first(2) to latest(5)");
// Cap: push HISTORY_CAP + 10 distinct samples; the ring holds exactly the cap
// and keeps the NEWEST (oldest dropped).
let mut h = FunnelHistory::default();
for i in 0..(HISTORY_CAP + 10) {
h.record(i, 0);
}
assert_eq!(h.len(), HISTORY_CAP, "capped at HISTORY_CAP");
assert_eq!(
h.latest_node_count(),
HISTORY_CAP + 9,
"the newest sample survives; the oldest were dropped"
);
h.clear();
assert!(h.is_empty(), "clear empties the trail");
}
/// **The DAG's growth is RECORDED as it grows** (the "addera historik" super-prio):
/// successive `Funnel.Show` polls carrying more task nodes push a strictly growing
/// node-count into the history trail (a re-poll with NO change collapses, so the
/// trail counts genuine growth moments, not ticks). Records the test-matrix cell.
#[test]
fn funnel_history_records_the_dag_growing() {
let plans: Vec<&[&str]> = vec![
&["a"][..],
&["a"][..], // an unchanged re-poll — must collapse
&["a", "b"][..],
&["a", "b", "c", "d"][..],
];
let mut h = FunnelHistory::default();
for titles in &plans {
let n = plandag_node_count(&dump_with_tasks(titles));
h.record(n, 0);
}
// node_count = mother(1) + tasks: 2, (2 collapses), 3, 5 → three change points.
let trail: Vec<usize> = h.samples().iter().map(|s| s.node_count).collect();
assert_eq!(
trail,
vec![2, 3, 5],
"the unchanged re-poll collapsed; growth recorded"
);
let monotonic = trail.windows(2).all(|w| w[1] > w[0]);
assert!(monotonic && h.node_growth() == 3);
facett_core::testmatrix::emit(
"nornir-funnel-embed",
"funnel_history_addera_historik",
monotonic && trail == vec![2, 3, 5] && h.node_growth() == 3,
"the live pane records the funnel DAG's node-count growth across polls (de-duped change trail) — the 'addera historik' growth history",
);
}
/// The offline pane exposes the history fields with the correct EMPTY shape (a
/// dead server never recorded a sample, but the trail is always present as data).
#[test]
fn offline_pane_exposes_empty_history_shape() {
let p = FunnelPanel::new("http://127.0.0.1:0", "", "");
let st = Facet::state_json(&p);
assert!(
st["history"].as_array().unwrap().is_empty(),
"no polls landed ⇒ empty trail"
);
assert_eq!(st["history_len"], 0);
assert_eq!(st["history_peak_nodes"], 0);
assert_eq!(st["node_growth"], 0);
}
#[test]
fn live_poll_offline_stays_unavailable_no_panic() {
// Headless, in-code: a forced live poll on a dead server (warehouse selected)
// keeps the clean `unreachable` state — the auto-poll loop can never wedge or
// panic the UI.
let mut p = FunnelPanel::new("http://127.0.0.1:0", "", "nordisk");
Facet::update_json(&mut p, r#"{"poll": true}"#);
let st = Facet::state_json(&p);
assert_eq!(st["status"], "unreachable");
assert_eq!(st["node_count"], 0);
// Test-matrix cell: the live-poll seam is exercised in code (self-driving,
// no AI, no manual run). No-op unless `--features testmatrix`.
facett_core::testmatrix::emit(
"nornir-funnel-embed",
"live_auto_poll_offline_safe",
st["status"] == "unreachable",
"forced poll on a dead server → clean unreachable, no panic; auto_poll toggles via state seam",
);
}
// ── run history + bottleneck ("addera historik", the NEXT layer) ────────────
The file goes on; the console shows all of it.