refactor(monorepo): reorganización lógica + renames + SDDs + split CHANGELOG
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 std::collections::HashSet;
30
31use brahman_handshake::messages::SessionList;
32use brahman_handshake::transport;
33use brahman_sidecar::{
34 await_provider_blocking, build_consumer_card, list_matches_blocking, list_sessions_blocking,
35 ConsumerError,
36};
37use ulid::Ulid;
38use gpui::{
39 div, prelude::*, px, Context, IntoElement, Render, SharedString, Window,
40};
41use nahual_launcher::launch_app;
42use nahual_theme::Theme;
43use nahual_widget_app_header::app_header;
44use nahual_widget_banner::{banner_themed, Banner};
45use nahual_widget_stat_card::stat_card;
46
47const POLL_INTERVAL: Duration = Duration::from_secs(5);
48const PROBE_TIMEOUT: Duration = Duration::from_secs(1);
49
50/// Cap del buffer del timeline. Mantenemos las últimas N entries —
51/// más viejo se descarta. 50 cubre ~4 minutos de actividad densa
52/// sin saturar el panel; subir si hace falta historia más larga.
53const TIMELINE_CAP: usize = 50;
54
55fn main() {
56 launch_app("Brahman Broker — Probe", (720., 480.), Explorer::new);
57}
58
59/// Snapshot de un probe.
60#[derive(Clone, Debug)]
61enum ProbeState {
62 /// Aún no probó (boot, primer ciclo).
63 Pending,
64 /// Connect failed → broker no responde en el path.
65 Down { reason: String },
66 /// Connect OK, sin matching producer dentro del timeout.
67 UpNoProvider { flow: String },
68 /// Connect OK, broker matcheó al menos un producer.
69 UpWithProvider {
70 flow: String,
71 producer_socket: PathBuf,
72 },
73}
74
75struct Explorer {
76 socket_path: PathBuf,
77 flow: String,
78 type_name: String,
79 state: ProbeState,
80 last_probe_ms: u64,
81 last_probe_at: Option<Instant>,
82 /// Última `SessionList` recibida del broker (None = aún sin pedir
83 /// o último intento falló).
84 sessions: Option<SessionList>,
85 /// Set de matches presentes en el último snapshot del broker.
86 /// Identificado por `(consumer.session, consumer.flow,
87 /// producer.session, producer.flow)` para que la diff entre
88 /// ticks distinga "nuevo match" vs "match perdido". Un cambio
89 /// de producer (otro session/flow para mismo consumer) cuenta
90 /// como Lost del previo + Available del nuevo.
91 last_match_keys: HashSet<MatchKey>,
92 /// Timeline FIFO: los más nuevos al frente. Cada entry tiene un
93 /// timestamp local + el evento sintético (Available/Lost) que
94 /// surgió de la diff del tick.
95 timeline: std::collections::VecDeque<TimelineEntry>,
96}
97
98/// Key estable para un match. Tupla porque (consumer, producer)
99/// determina el match unívocamente; los campos derivados (`label`,
100/// `via`, `pinned`) viajan en la entry pero no en la key.
101type MatchKey = (Ulid, String, Ulid, String);
102
103#[derive(Clone, Debug)]
104struct TimelineEntry {
105 /// Cuándo lo observó el explorer. Es tiempo local de wall-clock,
106 /// no del broker — el broker no timestampa los matches.
107 at: std::time::SystemTime,
108 /// Available = nuevo en este tick. Lost = estaba en el tick
109 /// anterior y desapareció.
110 kind: brahman_handshake::messages::MatchEventKind,
111 consumer_label: String,
112 consumer_flow: String,
113 producer_label: String,
114 producer_flow: String,
115 via: brahman_broker::MatchStrategy,
116 pinned: bool,
117}
118
119impl Explorer {
120 fn new(cx: &mut Context<Self>) -> Self {
121 let socket_path = transport::default_socket_path();
122 let flow = std::env::var("BRAHMAN_BROKER_PROBE_FLOW")
123 .unwrap_or_else(|_| "broker-health".to_string());
124 let type_name = std::env::var("BRAHMAN_BROKER_PROBE_TYPE")
125 .unwrap_or_else(|_| "ping".to_string());
126
127 let flow_for_loop = flow.clone();
128 let type_for_loop = type_name.clone();
129 cx.spawn(async move |this, cx| {
130 let timer = cx.background_executor().clone();
131 let bg = cx.background_executor().clone();
132 loop {
133 let card = build_consumer_card(
134 "brahman-broker-explorer",
135 flow_for_loop.clone(),
136 type_for_loop.clone(),
137 );
138 let started = Instant::now();
139 // El probe es bloqueante (interno tokio runtime); va
140 // al background executor para no congelar el main.
141 let probe_flow = flow_for_loop.clone();
142 let result = bg
143 .spawn(async move { await_provider_blocking(card, PROBE_TIMEOUT) })
144 .await;
145 let elapsed = started.elapsed().as_millis() as u64;
146
147 let new_state = match result {
148 Ok(socket) => ProbeState::UpWithProvider {
149 flow: probe_flow.clone(),
150 producer_socket: socket,
151 },
152 Err(ConsumerError::NoProvider { .. }) => ProbeState::UpNoProvider {
153 flow: probe_flow.clone(),
154 },
155 Err(e) => ProbeState::Down {
156 reason: e.to_string(),
157 },
158 };
159
160 // Si el broker está reachable (UP*), aprovechar el
161 // round-trip para pedir la lista de sesiones + matches.
162 // Si está DOWN, ni intentar — la lista serviría de nada
163 // con connect failed igual.
164 let (sessions_snapshot, matches_snapshot) = match &new_state {
165 ProbeState::Down { .. } | ProbeState::Pending => (None, None),
166 _ => {
167 let s = bg
168 .spawn(async move {
169 list_sessions_blocking("brahman-broker-explorer").ok()
170 })
171 .await;
172 let m = bg
173 .spawn(async move {
174 list_matches_blocking("brahman-broker-explorer").ok()
175 })
176 .await;
177 (s, m)
178 }
179 };
180
181 let _ = this.update(cx, |me, cx| {
182 me.state = new_state;
183 me.sessions = sessions_snapshot;
184 if let Some(matches) = matches_snapshot {
185 me.diff_matches_into_timeline(&matches);
186 }
187 me.last_probe_ms = elapsed;
188 me.last_probe_at = Some(Instant::now());
189 cx.notify();
190 });
191
192 timer.timer(POLL_INTERVAL).await;
193 }
194 })
195 .detach();
196
197 Self {
198 socket_path,
199 flow,
200 type_name,
201 state: ProbeState::Pending,
202 last_probe_ms: 0,
203 last_probe_at: None,
204 sessions: None,
205 last_match_keys: HashSet::new(),
206 timeline: std::collections::VecDeque::new(),
207 }
208 }
209
210 /// Diffea el snapshot recibido contra el último set de keys.
211 /// Genera entries `Available` para keys nuevas y `Lost` para
212 /// keys que estaban antes y no están ahora. Cada entry se
213 /// prepende al timeline; el cap se aplica desde la cola.
214 ///
215 /// El primer tick del explorer (cuando `last_match_keys` está
216 /// vacío) hace que TODOS los matches actuales aparezcan como
217 /// `Available` — es el comportamiento querido (UI muestra el
218 /// estado al boot sin que parezca "no pasa nada").
219 fn diff_matches_into_timeline(
220 &mut self,
221 list: &brahman_handshake::messages::MatchList,
222 ) {
223 let (new_entries, new_keys) = diff_matches(&self.last_match_keys, list);
224 for entry in new_entries {
225 self.push_timeline(entry);
226 }
227 self.last_match_keys = new_keys;
228 }
229
230 fn push_timeline(&mut self, entry: TimelineEntry) {
231 self.timeline.push_front(entry);
232 while self.timeline.len() > TIMELINE_CAP {
233 self.timeline.pop_back();
234 }
235 }
236}
237
238impl Render for Explorer {
239 fn render(&mut self, _w: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
240 let theme = Theme::global(cx).clone();
241 let bg = theme.bg_app.clone();
242 let text = theme.fg_text;
243 let text_dim = theme.fg_muted;
244 // Acentos por estado: verde = arriba, amber = arriba sin
245 // provider, rojo = abajo, gris = pending.
246 let accent_up = gpui::rgb(0xa3be8c);
247 let accent_partial = gpui::rgb(0xebcb8b);
248 let accent_down = gpui::rgb(0xbf616a);
249 let accent_pending = gpui::rgb(0x6a7280);
250
251 let header_text = format!(
252 "Probe: {} · flow: {}/{} · reload {} ms",
253 self.socket_path.display(),
254 self.flow,
255 self.type_name,
256 self.last_probe_ms,
257 );
258
259 // Header standard via widget compartido.
260 let header = app_header(cx, header_text);
261
262 // Banner permanente debajo del header con el estado actual.
263 // Severidad acorde al kind.
264 let status_banner = match &self.state {
265 ProbeState::Pending => None,
266 ProbeState::Down { reason } => Some(banner_themed(
267 cx,
268 Banner::Error,
269 SharedString::from(format!("Broker DOWN — {reason}")),
270 )),
271 ProbeState::UpNoProvider { .. } => Some(banner_themed(
272 cx,
273 Banner::Warning,
274 SharedString::from("Broker UP, sin provider para el flow"),
275 )),
276 ProbeState::UpWithProvider { .. } => Some(banner_themed(
277 cx,
278 Banner::Success,
279 SharedString::from("Broker UP, provider matcheado"),
280 )),
281 };
282
283 let sessions_items: Vec<String> = self
284 .sessions
285 .as_ref()
286 .map(|list| {
287 let mut entries: Vec<_> = list.entries.iter().collect();
288 // Orden estable por session id (Ulid es ordenable
289 // temporal); útil para que la UI no se reordene
290 // entre ticks aunque el HashMap del server sí.
291 entries.sort_by_key(|e| e.session);
292 entries
293 .iter()
294 .map(|e| {
295 format!(
296 "{} · in:[{}] out:[{}]{}",
297 e.label,
298 e.inputs.join(","),
299 e.outputs.join(","),
300 if e.conscious { " (wit)" } else { "" }
301 )
302 })
303 .collect()
304 })
305 .unwrap_or_default();
306
307 let sessions_count_value = self
308 .sessions
309 .as_ref()
310 .map(|l| l.entries.len().to_string())
311 .unwrap_or_else(|| "—".into());
312 let sessions_descr = match &self.sessions {
313 None => "lista no disponible (broker DOWN o pendiente)".to_string(),
314 Some(l) if l.entries.is_empty() => "sin sesiones registradas en el broker".into(),
315 Some(_) => "labels visibles + flows in/out · (wit) = consciente".into(),
316 };
317
318 let timeline_items: Vec<String> = self
319 .timeline
320 .iter()
321 .take(20)
322 .map(|e| format_timeline_entry(e))
323 .collect();
324 let timeline_value = self.timeline.len().to_string();
325 let timeline_descr = if self.timeline.is_empty() {
326 "esperando primer match…".to_string()
327 } else {
328 "↑ más reciente · ↓ más viejo · cap 50 entries".to_string()
329 };
330
331 let body = div()
332 .flex()
333 .flex_col()
334 .gap(px(8.))
335 .px(px(16.))
336 .py(px(16.))
337 .child(state_card(cx, &self.state, text, text_dim, accent_up,
338 accent_partial, accent_down, accent_pending))
339 .child(stat_card(
340 cx,
341 "Sesiones activas",
342 sessions_count_value,
343 &sessions_descr,
344 accent_up,
345 text,
346 text_dim,
347 &sessions_items,
348 ))
349 .child(stat_card(
350 cx,
351 "Timeline de matches",
352 timeline_value,
353 &timeline_descr,
354 accent_partial,
355 text,
356 text_dim,
357 &timeline_items,
358 ));
359
360 div()
361 .flex()
362 .flex_col()
363 .size_full()
364 .bg(bg)
365 .child(header)
366 .when_some(status_banner, |d, b| d.child(b))
367 .child(body)
368 }
369}
370
371/// Wrap del `stat_card` compartido con el mapeo de
372/// `ProbeState` → (label/accent/value/description). Mantenemos
373/// este helper local porque la traducción del enum a strings es
374/// específica del explorer (no es un patrón cross-app).
375#[allow(clippy::too_many_arguments)]
376fn state_card(
377 cx: &mut Context<Explorer>,
378 state: &ProbeState,
379 text: gpui::Hsla,
380 text_dim: gpui::Hsla,
381 accent_up: gpui::Rgba,
382 accent_partial: gpui::Rgba,
383 accent_down: gpui::Rgba,
384 accent_pending: gpui::Rgba,
385) -> impl IntoElement {
386 let (accent, value, description): (gpui::Rgba, String, String) = match state {
387 ProbeState::Pending => (
388 accent_pending,
389 "PENDING".into(),
390 "esperando primer probe…".into(),
391 ),
392 ProbeState::Down { reason } => (
393 accent_down,
394 "DOWN".into(),
395 format!("connect failed: {reason}"),
396 ),
397 ProbeState::UpNoProvider { flow } => (
398 accent_partial,
399 "UP / NO PROVIDER".into(),
400 format!("broker reachable; sin productor para flow `{flow}`"),
401 ),
402 ProbeState::UpWithProvider {
403 flow,
404 producer_socket,
405 } => (
406 accent_up,
407 "UP / PROVIDER".into(),
408 format!(
409 "flow `{flow}` matcheado en producer socket: {}",
410 producer_socket.display()
411 ),
412 ),
413 };
414
415 stat_card(cx, "Estado", value, &description, accent, text, text_dim, &[])
416}
417
418/// Diff puro entre snapshots de matches. Devuelve la lista de
419/// entries nuevas (Available + Lost) en orden Available-primero, y
420/// el set actualizado de keys. Extraído como free fn para que sea
421/// testeable sin instanciar `Explorer`.
422///
423/// El primer tick (last_keys vacío) marca todos los matches como
424/// Available. Esto es deliberado: la UI muestra el estado al boot
425/// como "todo recién apareció" en vez de quedarse vacía.
426fn diff_matches(
427 last_keys: &HashSet<MatchKey>,
428 list: &brahman_handshake::messages::MatchList,
429) -> (Vec<TimelineEntry>, HashSet<MatchKey>) {
430 use brahman_handshake::messages::MatchEventKind;
431 let now = std::time::SystemTime::now();
432 let current_keys: HashSet<MatchKey> = list
433 .matches
434 .iter()
435 .map(|m| {
436 (
437 m.consumer.session,
438 m.consumer.flow_name.clone(),
439 m.producer.session,
440 m.producer.flow_name.clone(),
441 )
442 })
443 .collect();
444
445 let mut entries = Vec::new();
446 for m in &list.matches {
447 let key = (
448 m.consumer.session,
449 m.consumer.flow_name.clone(),
450 m.producer.session,
451 m.producer.flow_name.clone(),
452 );
453 if !last_keys.contains(&key) {
454 entries.push(TimelineEntry {
455 at: now,
456 kind: MatchEventKind::Available,
457 consumer_label: m.consumer_label.clone(),
458 consumer_flow: m.consumer.flow_name.clone(),
459 producer_label: m.producer_label.clone(),
460 producer_flow: m.producer.flow_name.clone(),
461 via: m.via,
462 pinned: m.pinned,
463 });
464 }
465 }
466 for key in last_keys.iter() {
467 if !current_keys.contains(key) {
468 entries.push(TimelineEntry {
469 at: now,
470 kind: MatchEventKind::Lost,
471 consumer_label: String::new(),
472 consumer_flow: key.1.clone(),
473 producer_label: String::new(),
474 producer_flow: key.3.clone(),
475 via: brahman_broker::MatchStrategy::Exact,
476 pinned: false,
477 });
478 }
479 }
480 (entries, current_keys)
481}
482
483/// Renderiza una entry del timeline en una sola línea: `HH:MM:SS
484/// {kind} consumer.flow ← producer.flow [via]`. Compact por diseño
485/// — el panel es vertical y las líneas largas se cortan.
486fn format_timeline_entry(e: &TimelineEntry) -> String {
487 use brahman_handshake::messages::MatchEventKind;
488 // Wall-clock local en HH:MM:SS — sin zoneinfo (chrono es heavy).
489 // Aproximación: total_seconds % 86400 → hora del día UTC.
490 let secs_today = e
491 .at
492 .duration_since(std::time::UNIX_EPOCH)
493 .map(|d| d.as_secs() % 86_400)
494 .unwrap_or(0);
495 let h = secs_today / 3600;
496 let m = (secs_today % 3600) / 60;
497 let s = secs_today % 60;
498 let kind = match e.kind {
499 MatchEventKind::Available => "+",
500 MatchEventKind::Lost => "-",
501 };
502 let pinned = if e.pinned { " (pinned)" } else { "" };
503 match e.kind {
504 MatchEventKind::Available => format!(
505 "{:02}:{:02}:{:02} {} {}.{} ← {}.{} [{:?}]{}",
506 h, m, s, kind,
507 e.consumer_label, e.consumer_flow,
508 e.producer_label, e.producer_flow,
509 e.via, pinned,
510 ),
511 MatchEventKind::Lost => format!(
512 "{:02}:{:02}:{:02} {} ?.{} ← ?.{} (lost)",
513 h, m, s, kind, e.consumer_flow, e.producer_flow,
514 ),
515 }
516}
517
518#[cfg(test)]
519mod tests {
520 use super::*;
521
522 #[test]
523 fn pending_is_default_state_at_boot() {
524 // El estado inicial DEBE ser Pending — sino el banner del
525 // header arrancaría con un Error o Warning sin haber probado
526 // (mensaje engañoso).
527 let s = ProbeState::Pending;
528 assert!(matches!(s, ProbeState::Pending));
529 }
530
531 #[test]
532 fn poll_and_probe_constants_are_sane() {
533 // Sanity: el timeout del probe DEBE ser menor que el
534 // intervalo de polling, sino los probes se solapan.
535 assert!(PROBE_TIMEOUT < POLL_INTERVAL);
536 // El intervalo no debería ser tan corto que sature al broker.
537 assert!(POLL_INTERVAL >= Duration::from_secs(2));
538 }
539
540 fn synth_match(
541 consumer_label: &str,
542 consumer_flow: &str,
543 producer_label: &str,
544 producer_flow: &str,
545 ) -> brahman_broker::Match {
546 use brahman_broker::{Endpoint, Match, MatchStrategy};
547 use brahman_card::TypeRef;
548 Match {
549 consumer: Endpoint {
550 session: Ulid::new(),
551 flow_name: consumer_flow.into(),
552 },
553 consumer_label: consumer_label.into(),
554 producer: Endpoint {
555 session: Ulid::new(),
556 flow_name: producer_flow.into(),
557 },
558 producer_label: producer_label.into(),
559 ty: TypeRef::Primitive { name: "json".into() },
560 via: MatchStrategy::Exact,
561 pinned: false,
562 }
563 }
564
565 #[test]
566 fn diff_matches_first_snapshot_marks_everything_available() {
567 use brahman_handshake::messages::{MatchEventKind, MatchList};
568 let list = MatchList {
569 matches: vec![
570 synth_match("a", "x", "b", "x"),
571 synth_match("c", "y", "d", "y"),
572 ],
573 };
574 let last = HashSet::new();
575 let (entries, keys) = diff_matches(&last, &list);
576 assert_eq!(entries.len(), 2);
577 assert!(entries
578 .iter()
579 .all(|e| matches!(e.kind, MatchEventKind::Available)));
580 assert_eq!(keys.len(), 2);
581 }
582
583 #[test]
584 fn diff_matches_emits_lost_when_match_disappears() {
585 use brahman_handshake::messages::{MatchEventKind, MatchList};
586 let m = synth_match("a", "x", "b", "x");
587 let prev_key = (
588 m.consumer.session,
589 m.consumer.flow_name.clone(),
590 m.producer.session,
591 m.producer.flow_name.clone(),
592 );
593 let last: HashSet<_> = std::iter::once(prev_key.clone()).collect();
594 let list = MatchList { matches: vec![] };
595 let (entries, keys) = diff_matches(&last, &list);
596 assert_eq!(entries.len(), 1);
597 assert!(matches!(entries[0].kind, MatchEventKind::Lost));
598 assert_eq!(entries[0].consumer_flow, "x");
599 assert_eq!(entries[0].producer_flow, "x");
600 assert!(keys.is_empty());
601 }
602
603 #[test]
604 fn diff_matches_no_change_emits_nothing() {
605 use brahman_handshake::messages::MatchList;
606 let m = synth_match("a", "x", "b", "x");
607 let key = (
608 m.consumer.session,
609 m.consumer.flow_name.clone(),
610 m.producer.session,
611 m.producer.flow_name.clone(),
612 );
613 let last: HashSet<_> = std::iter::once(key.clone()).collect();
614 let list = MatchList {
615 matches: vec![m.clone()],
616 };
617 let (entries, keys) = diff_matches(&last, &list);
618 assert!(entries.is_empty(), "match unchanged → no events");
619 assert_eq!(keys.len(), 1);
620 assert!(keys.contains(&key));
621 }
622}