tawasuyu/brahman

1//! `brahman-broker-explorer` — probe GUI del broker brahman.
2//!
3//! Cada [`POLL_INTERVAL`] arma un Card observer agnóstico y lo
4//! manda al broker via `brahman_sidecar::await_provider_blocking`
5//! (que internamente abre tokio runtime + Unix socket + handshake).
6//! Reporta 3 estados:
7//!
8//! - **Down**: connect failed (broker no escucha en el socket).
9//! - **Up sin provider**: connect OK, pero el broker no encontró
10//!   productor para el flow probado dentro del timeout.
11//! - **Up con provider**: connect OK + el broker matcheó algo →
12//!   muestra el `producer_service_socket` recibido.
13//!
14//! Configuración via env:
15//! - `BRAHMAN_INIT_SOCKET` — path del socket del broker (default
16//!   resuelto por `brahman_handshake::transport`).
17//! - `BRAHMAN_BROKER_PROBE_FLOW` — nombre del flow probe (default
18//!   `broker-health`).
19//! - `BRAHMAN_BROKER_PROBE_TYPE` — type name del flow probe
20//!   (default `ping`).
21//!
22//! Usá un type name probable (ej. `monad-list:json`,
23//! `event-log:tail`) para detectar productores específicos del
24//! ecosistema.
25
26use std::path::PathBuf;
27use std::time::{Duration, Instant};
28
29use brahman_handshake::messages::SessionList;
30use brahman_handshake::transport;
31use brahman_sidecar::{
32    await_provider_blocking, build_consumer_card, list_sessions_blocking, ConsumerError,
33};
34use gpui::{
35    div, prelude::*, px, Context, IntoElement, Render, SharedString, Window,
36};
37use yahweh_launcher::launch_app;
38use yahweh_theme::Theme;
39use yahweh_widget_app_header::app_header;
40use yahweh_widget_banner::{banner_themed, Banner};
41use yahweh_widget_stat_card::stat_card;
42
43const POLL_INTERVAL: Duration = Duration::from_secs(5);
44const PROBE_TIMEOUT: Duration = Duration::from_secs(1);
45
46fn main() {
47    launch_app("Brahman Broker — Probe", (720., 480.), Explorer::new);
48}
49
50/// Snapshot de un probe.
51#[derive(Clone, Debug)]
52enum ProbeState {
53    /// Aún no probó (boot, primer ciclo).
54    Pending,
55    /// Connect failed → broker no responde en el path.
56    Down { reason: String },
57    /// Connect OK, sin matching producer dentro del timeout.
58    UpNoProvider { flow: String },
59    /// Connect OK, broker matcheó al menos un producer.
60    UpWithProvider {
61        flow: String,
62        producer_socket: PathBuf,
63    },
64}
65
66struct Explorer {
67    socket_path: PathBuf,
68    flow: String,
69    type_name: String,
70    state: ProbeState,
71    last_probe_ms: u64,
72    last_probe_at: Option<Instant>,
73    /// Última `SessionList` recibida del broker (None = aún sin pedir
74    /// o último intento falló).
75    sessions: Option<SessionList>,
76}
77
78impl Explorer {
79    fn new(cx: &mut Context<Self>) -> Self {
80        let socket_path = transport::default_socket_path();
81        let flow = std::env::var("BRAHMAN_BROKER_PROBE_FLOW")
82            .unwrap_or_else(|_| "broker-health".to_string());
83        let type_name = std::env::var("BRAHMAN_BROKER_PROBE_TYPE")
84            .unwrap_or_else(|_| "ping".to_string());
85
86        let flow_for_loop = flow.clone();
87        let type_for_loop = type_name.clone();
88        cx.spawn(async move |this, cx| {
89            let timer = cx.background_executor().clone();
90            let bg = cx.background_executor().clone();
91            loop {
92                let card = build_consumer_card(
93                    "brahman-broker-explorer",
94                    flow_for_loop.clone(),
95                    type_for_loop.clone(),
96                );
97                let started = Instant::now();
98                // El probe es bloqueante (interno tokio runtime); va
99                // al background executor para no congelar el main.
100                let probe_flow = flow_for_loop.clone();
101                let result = bg
102                    .spawn(async move { await_provider_blocking(card, PROBE_TIMEOUT) })
103                    .await;
104                let elapsed = started.elapsed().as_millis() as u64;
105
106                let new_state = match result {
107                    Ok(socket) => ProbeState::UpWithProvider {
108                        flow: probe_flow.clone(),
109                        producer_socket: socket,
110                    },
111                    Err(ConsumerError::NoProvider { .. }) => ProbeState::UpNoProvider {
112                        flow: probe_flow.clone(),
113                    },
114                    Err(e) => ProbeState::Down {
115                        reason: e.to_string(),
116                    },
117                };
118
119                // Si el broker está reachable (UP*), aprovechar el
120                // round-trip para pedir la lista de sesiones. Si está
121                // DOWN, ni intentar — la lista serviría de nada con
122                // connect failed igual.
123                let sessions_snapshot = match &new_state {
124                    ProbeState::Down { .. } | ProbeState::Pending => None,
125                    _ => bg
126                        .spawn(async move {
127                            list_sessions_blocking("brahman-broker-explorer").ok()
128                        })
129                        .await,
130                };
131
132                let _ = this.update(cx, |me, cx| {
133                    me.state = new_state;
134                    me.sessions = sessions_snapshot;
135                    me.last_probe_ms = elapsed;
136                    me.last_probe_at = Some(Instant::now());
137                    cx.notify();
138                });
139
140                timer.timer(POLL_INTERVAL).await;
141            }
142        })
143        .detach();
144
145        Self {
146            socket_path,
147            flow,
148            type_name,
149            state: ProbeState::Pending,
150            last_probe_ms: 0,
151            last_probe_at: None,
152            sessions: None,
153        }
154    }
155}
156
157impl Render for Explorer {
158    fn render(&mut self, _w: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
159        let theme = Theme::global(cx).clone();
160        let bg = theme.bg_app.clone();
161        let text = theme.fg_text;
162        let text_dim = theme.fg_muted;
163        // Acentos por estado: verde = arriba, amber = arriba sin
164        // provider, rojo = abajo, gris = pending.
165        let accent_up = gpui::rgb(0xa3be8c);
166        let accent_partial = gpui::rgb(0xebcb8b);
167        let accent_down = gpui::rgb(0xbf616a);
168        let accent_pending = gpui::rgb(0x6a7280);
169
170        let header_text = format!(
171            "Probe: {}  ·  flow: {}/{}  ·  reload {} ms",
172            self.socket_path.display(),
173            self.flow,
174            self.type_name,
175            self.last_probe_ms,
176        );
177
178        // Header standard via widget compartido.
179        let header = app_header(cx, header_text);
180
181        // Banner permanente debajo del header con el estado actual.
182        // Severidad acorde al kind.
183        let status_banner = match &self.state {
184            ProbeState::Pending => None,
185            ProbeState::Down { reason } => Some(banner_themed(
186                cx,
187                Banner::Error,
188                SharedString::from(format!("Broker DOWN — {reason}")),
189            )),
190            ProbeState::UpNoProvider { .. } => Some(banner_themed(
191                cx,
192                Banner::Warning,
193                SharedString::from("Broker UP, sin provider para el flow"),
194            )),
195            ProbeState::UpWithProvider { .. } => Some(banner_themed(
196                cx,
197                Banner::Success,
198                SharedString::from("Broker UP, provider matcheado"),
199            )),
200        };
201
202        let sessions_items: Vec<String> = self
203            .sessions
204            .as_ref()
205            .map(|list| {
206                let mut entries: Vec<_> = list.entries.iter().collect();
207                // Orden estable por session id (Ulid es ordenable
208                // temporal); útil para que la UI no se reordene
209                // entre ticks aunque el HashMap del server sí.
210                entries.sort_by_key(|e| e.session);
211                entries
212                    .iter()
213                    .map(|e| {
214                        format!(
215                            "{}  ·  in:[{}]  out:[{}]{}",
216                            e.label,
217                            e.inputs.join(","),
218                            e.outputs.join(","),
219                            if e.conscious { "  (wit)" } else { "" }
220                        )
221                    })
222                    .collect()
223            })
224            .unwrap_or_default();
225
226        let sessions_count_value = self
227            .sessions
228            .as_ref()
229            .map(|l| l.entries.len().to_string())
230            .unwrap_or_else(|| "—".into());
231        let sessions_descr = match &self.sessions {
232            None => "lista no disponible (broker DOWN o pendiente)".to_string(),
233            Some(l) if l.entries.is_empty() => "sin sesiones registradas en el broker".into(),
234            Some(_) => "labels visibles + flows in/out · (wit) = consciente".into(),
235        };
236
237        let body = div()
238            .flex()
239            .flex_col()
240            .gap(px(8.))
241            .px(px(16.))
242            .py(px(16.))
243            .child(state_card(cx, &self.state, text, text_dim, accent_up,
244                accent_partial, accent_down, accent_pending))
245            .child(stat_card(
246                cx,
247                "Sesiones activas",
248                sessions_count_value,
249                &sessions_descr,
250                accent_up,
251                text,
252                text_dim,
253                &sessions_items,
254            ));
255
256        div()
257            .flex()
258            .flex_col()
259            .size_full()
260            .bg(bg)
261            .child(header)
262            .when_some(status_banner, |d, b| d.child(b))
263            .child(body)
264    }
265}
266
267/// Wrap del `stat_card` compartido con el mapeo de
268/// `ProbeState` → (label/accent/value/description). Mantenemos
269/// este helper local porque la traducción del enum a strings es
270/// específica del explorer (no es un patrón cross-app).
271#[allow(clippy::too_many_arguments)]
272fn state_card(
273    cx: &mut Context<Explorer>,
274    state: &ProbeState,
275    text: gpui::Hsla,
276    text_dim: gpui::Hsla,
277    accent_up: gpui::Rgba,
278    accent_partial: gpui::Rgba,
279    accent_down: gpui::Rgba,
280    accent_pending: gpui::Rgba,
281) -> impl IntoElement {
282    let (accent, value, description): (gpui::Rgba, String, String) = match state {
283        ProbeState::Pending => (
284            accent_pending,
285            "PENDING".into(),
286            "esperando primer probe…".into(),
287        ),
288        ProbeState::Down { reason } => (
289            accent_down,
290            "DOWN".into(),
291            format!("connect failed: {reason}"),
292        ),
293        ProbeState::UpNoProvider { flow } => (
294            accent_partial,
295            "UP / NO PROVIDER".into(),
296            format!("broker reachable; sin productor para flow `{flow}`"),
297        ),
298        ProbeState::UpWithProvider {
299            flow,
300            producer_socket,
301        } => (
302            accent_up,
303            "UP / PROVIDER".into(),
304            format!(
305                "flow `{flow}` matcheado en producer socket: {}",
306                producer_socket.display()
307            ),
308        ),
309    };
310
311    stat_card(cx, "Estado", value, &description, accent, text, text_dim, &[])
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317
318    #[test]
319    fn pending_is_default_state_at_boot() {
320        // El estado inicial DEBE ser Pending — sino el banner del
321        // header arrancaría con un Error o Warning sin haber probado
322        // (mensaje engañoso).
323        let s = ProbeState::Pending;
324        assert!(matches!(s, ProbeState::Pending));
325    }
326
327    #[test]
328    fn poll_and_probe_constants_are_sane() {
329        // Sanity: el timeout del probe DEBE ser menor que el
330        // intervalo de polling, sino los probes se solapan.
331        assert!(PROBE_TIMEOUT < POLL_INTERVAL);
332        // El intervalo no debería ser tan corto que sature al broker.
333        assert!(POLL_INTERVAL >= Duration::from_secs(2));
334    }
335}