⎇ d3649bfd1a80153652c2a028315369bb4b184024
crates/apps/shipote-shell/src/main.rs
Historia
Quién tocó qué
crudo
feat(shipote): throughput card + rate-limit + snapshot incremental (fase Q)
1//! `shipote-shell` — GUI dashboard del daemon shipote.
2//!
3//! Probe-style: conecta al socket del daemon cada 2s, pide
4//! capabilities + workspace-list y los muestra en cards.
5//! Si el daemon no está corriendo, marca DOWN.
6
7use gpui::{div, prelude::*, px, Context, IntoElement, Render, SharedString, Window};
8use shipote_protocol::{
9 default_socket_path, read_frame, write_frame, CommandInfo, FlowInfo, FlowThroughputInfo,
10 QuotaReportInfo, Request, Response, WorkspaceStatsInfo, WorkspaceSummary,
11};
12use std::path::PathBuf;
13use std::time::Duration;
14use tokio::net::UnixStream;
15use yahweh_launcher::launch_app;
16use yahweh_theme::Theme;
17use yahweh_widget_app_header::app_header;
18use yahweh_widget_banner::{banner_themed, Banner};
19use yahweh_widget_stat_card::stat_card;
20
21const POLL_INTERVAL: Duration = Duration::from_secs(2);
22
23#[derive(Clone, Debug)]
24enum DaemonState {
25 Pending,
26 Down { reason: String },
27 Up,
28}
29
30#[derive(Clone, Debug, Default)]
31struct CapsSummary {
32 kernel_version: (u32, u32, u32),
33 user_ns: String,
34 cgroup_v2: String,
35 cgroup_delegated: bool,
36 has_cap_sys_admin: bool,
37}
38
39struct Shell {
40 socket_path: PathBuf,
41 state: DaemonState,
42 workspaces: Vec<WorkspaceSummary>,
43 /// Comandos por workspace, indexados por workspace id.toString().
44 commands: std::collections::BTreeMap<String, Vec<CommandInfo>>,
45 saved_pipelines: Vec<String>,
46 flows: Vec<FlowInfo>,
47 /// Throughput por flow socket (bytes_total + bytes/s).
48 flow_throughput: Vec<FlowThroughputInfo>,
49 /// History de RSS por workspace (últimas N samples).
50 stats_history: std::collections::BTreeMap<String, std::collections::VecDeque<WorkspaceStatsInfo>>,
51 /// Quota report fresco por workspace.
52 quotas: std::collections::BTreeMap<String, QuotaReportInfo>,
53 caps: Option<CapsSummary>,
54 last_probe_ms: u64,
55 recent_log: Option<(String, String)>,
56}
57
58const STATS_HISTORY_LEN: usize = 24;
59
60fn main() {
61 launch_app("Shipote — Shell", (820., 560.), Shell::new);
62}
63
64impl Shell {
65 fn new(cx: &mut Context<Self>) -> Self {
66 let socket_path = default_socket_path();
67 let socket_for_loop = socket_path.clone();
68 cx.spawn(async move |this, cx| {
69 let timer = cx.background_executor().clone();
70 let bg = cx.background_executor().clone();
71 loop {
72 let path = socket_for_loop.clone();
73 let started = std::time::Instant::now();
74 let result = bg
75 .spawn(async move { probe_blocking(&path) })
76 .await;
77 let elapsed = started.elapsed().as_millis() as u64;
78 let _ = this.update(cx, |me, cx| {
79 match result {
80 Ok(snap) => {
81 me.state = DaemonState::Up;
82 me.workspaces = snap.workspaces;
83 me.commands = snap.commands;
84 me.saved_pipelines = snap.saved_pipelines;
85 me.flows = snap.flows;
86 me.flow_throughput = snap.flow_throughput;
87 me.quotas = snap.quotas;
88 // Hidratar history server-side para workspaces
89 // que no tenían history local (primer probe).
90 for ws in &me.workspaces {
91 let key = ws.id.to_string();
92 if !me.stats_history.contains_key(&key) {
93 if let Some(hydrated) = snap.hydrate_history.get(&key) {
94 me.stats_history.insert(
95 key.clone(),
96 hydrated.iter().cloned().collect(),
97 );
98 }
99 }
100 }
101 // Append fresh sample a la history por workspace.
102 for (ws_id, fresh) in &snap.fresh_stats {
103 let h = me
104 .stats_history
105 .entry(ws_id.clone())
106 .or_default();
107 if h.len() >= STATS_HISTORY_LEN {
108 h.pop_front();
109 }
110 h.push_back(fresh.clone());
111 }
112 // Limpiar history de workspaces que ya no existen.
113 let alive: std::collections::HashSet<String> = me
114 .workspaces
115 .iter()
116 .map(|w| w.id.to_string())
117 .collect();
118 me.stats_history.retain(|k, _| alive.contains(k));
119 me.caps = Some(snap.caps);
120 me.recent_log = snap.recent_log;
121 }
122 Err(reason) => {
123 me.state = DaemonState::Down { reason };
124 me.workspaces.clear();
125 me.commands.clear();
126 me.saved_pipelines.clear();
127 me.flows.clear();
128 me.flow_throughput.clear();
129 me.quotas.clear();
130 me.caps = None;
131 me.recent_log = None;
132 }
133 }
134 me.last_probe_ms = elapsed;
135 cx.notify();
136 });
137 timer.timer(POLL_INTERVAL).await;
138 }
139 })
140 .detach();
141
142 Self {
143 socket_path,
144 state: DaemonState::Pending,
145 workspaces: Vec::new(),
146 commands: std::collections::BTreeMap::new(),
147 saved_pipelines: Vec::new(),
148 flows: Vec::new(),
149 flow_throughput: Vec::new(),
150 stats_history: std::collections::BTreeMap::new(),
151 quotas: std::collections::BTreeMap::new(),
152 caps: None,
153 last_probe_ms: 0,
154 recent_log: None,
155 }
156 }
157}
158
159#[derive(Debug)]
160struct Snapshot {
161 workspaces: Vec<WorkspaceSummary>,
162 commands: std::collections::BTreeMap<String, Vec<CommandInfo>>,
163 saved_pipelines: Vec<String>,
164 flows: Vec<FlowInfo>,
165 flow_throughput: Vec<FlowThroughputInfo>,
166 /// Stats fresco por workspace (id.toString → stats).
167 fresh_stats: std::collections::BTreeMap<String, WorkspaceStatsInfo>,
168 /// Quota report fresco por workspace.
169 quotas: std::collections::BTreeMap<String, QuotaReportInfo>,
170 /// Workspaces nuevos (no en history local): hidratamos history
171 /// server-side al primer probe que los vea.
172 hydrate_history: std::collections::BTreeMap<String, Vec<WorkspaceStatsInfo>>,
173 caps: CapsSummary,
174 /// tail del log del comando más reciente (label + bytes). None si no hay.
175 recent_log: Option<(String, String)>,
176}
177
178fn probe_blocking(path: &std::path::Path) -> Result<Snapshot, String> {
179 // Mini tokio runtime efímero por probe — no compartimos runtime con
180 // GPUI. Costo aceptable cada 2s: setup ≈ <1 ms.
181 let rt = tokio::runtime::Builder::new_current_thread()
182 .enable_io()
183 .enable_time()
184 .build()
185 .map_err(|e| format!("rt: {e}"))?;
186 rt.block_on(async {
187 let mut stream = UnixStream::connect(path)
188 .await
189 .map_err(|e| format!("connect: {e}"))?;
190 write_frame(&mut stream, &Request::WorkspaceList)
191 .await
192 .map_err(|e| format!("write list: {e}"))?;
193 let resp: Response = read_frame(&mut stream).await.map_err(|e| format!("read list: {e}"))?;
194 let workspaces = match resp {
195 Response::WorkspaceList { items } => items,
196 other => return Err(format!("unexpected list resp: {other:?}")),
197 };
198
199 // Batched: stats+quota+commands+flow_sockets en 1 roundtrip por ws.
200 // Para workspaces nuevos, también pedimos history server-side.
201 let mut commands_map = std::collections::BTreeMap::new();
202 let mut fresh_stats = std::collections::BTreeMap::new();
203 let mut quotas = std::collections::BTreeMap::new();
204 let mut hydrate_history = std::collections::BTreeMap::new();
205 for w in &workspaces {
206 write_frame(&mut stream, &Request::WorkspaceFullSummary { workspace: w.id })
207 .await
208 .map_err(|e| format!("write summary: {e}"))?;
209 let resp: Response = read_frame(&mut stream)
210 .await
211 .map_err(|e| format!("read summary: {e}"))?;
212 if let Response::WorkspaceFullSummary { stats, quota, commands, .. } = resp {
213 let key = w.id.to_string();
214 fresh_stats.insert(key.clone(), stats);
215 quotas.insert(key.clone(), quota);
216 if !commands.is_empty() {
217 commands_map.insert(key, commands);
218 }
219 }
220 // History server-side (para hidratar si el shell es nuevo).
221 write_frame(
222 &mut stream,
223 &Request::WorkspaceStatsHistory {
224 workspace: w.id,
225 tail: 24, // mismo cap que STATS_HISTORY_LEN
226 },
227 )
228 .await
229 .map_err(|e| format!("write history: {e}"))?;
230 let resp: Response = read_frame(&mut stream)
231 .await
232 .map_err(|e| format!("read history: {e}"))?;
233 if let Response::WorkspaceStatsHistory { samples } = resp {
234 if !samples.is_empty() {
235 hydrate_history.insert(w.id.to_string(), samples);
236 }
237 }
238 }
239
240 // Saved pipelines.
241 write_frame(&mut stream, &Request::PipelineSavedList)
242 .await
243 .map_err(|e| format!("write saved: {e}"))?;
244 let resp: Response = read_frame(&mut stream)
245 .await
246 .map_err(|e| format!("read saved: {e}"))?;
247 let saved_pipelines = match resp {
248 Response::PipelineSavedList { names } => names,
249 _ => Vec::new(),
250 };
251
252 // Flow channels activos (data plane).
253 write_frame(&mut stream, &Request::FlowList)
254 .await
255 .map_err(|e| format!("write flows: {e}"))?;
256 let resp: Response = read_frame(&mut stream)
257 .await
258 .map_err(|e| format!("read flows: {e}"))?;
259 let flows = match resp {
260 Response::FlowList { items } => items,
261 _ => Vec::new(),
262 };
263 // Throughput per-socket.
264 write_frame(&mut stream, &Request::FlowThroughput)
265 .await
266 .map_err(|e| format!("write throughput: {e}"))?;
267 let resp: Response = read_frame(&mut stream)
268 .await
269 .map_err(|e| format!("read throughput: {e}"))?;
270 let flow_throughput = match resp {
271 Response::FlowThroughput { items } => items,
272 _ => Vec::new(),
273 };
274
275 // Live tail: log del comando más reciente con bytes>0.
276 let recent_log = {
277 // Pick: comando con id más alto que tiene log_bytes>0, en cualquier workspace.
278 let mut best: Option<(&str, &CommandInfo)> = None;
279 for (ws, cmds) in &commands_map {
280 for c in cmds {
281 if c.log_bytes == 0 {
282 continue;
283 }
284 let take = match &best {
285 Some((_, prev)) => c.id > prev.id,
286 None => true,
287 };
288 if take {
289 best = Some((ws.as_str(), c));
290 }
291 }
292 }
293 match best {
294 Some((ws_str, cmd)) => {
295 let ws_id: shipote_card::WorkspaceId = ws_str
296 .parse::<ulid::Ulid>()
297 .map(shipote_card::WorkspaceId)
298 .map_err(|e| format!("ulid parse: {e}"))?;
299 write_frame(
300 &mut stream,
301 &Request::CommandLogs {
302 workspace: ws_id,
303 command: cmd.id,
304 tail_bytes: 512,
305 stream: "both".into(),
306 },
307 )
308 .await
309 .map_err(|e| format!("write logs: {e}"))?;
310 let resp: Response = read_frame(&mut stream)
311 .await
312 .map_err(|e| format!("read logs: {e}"))?;
313 match resp {
314 Response::CommandLogs { bytes } => {
315 let text = String::from_utf8_lossy(&bytes).to_string();
316 Some((cmd.label.clone(), text))
317 }
318 _ => None,
319 }
320 }
321 None => None,
322 }
323 };
324
325 write_frame(&mut stream, &Request::Capabilities)
326 .await
327 .map_err(|e| format!("write caps: {e}"))?;
328 let resp: Response = read_frame(&mut stream).await.map_err(|e| format!("read caps: {e}"))?;
329 let caps = match resp {
330 Response::Capabilities {
331 kernel_version,
332 user_ns,
333 cgroup_v2,
334 cgroup_delegated,
335 has_cap_sys_admin,
336 } => CapsSummary {
337 kernel_version,
338 user_ns,
339 cgroup_v2,
340 cgroup_delegated,
341 has_cap_sys_admin,
342 },
343 other => return Err(format!("unexpected caps resp: {other:?}")),
344 };
345 Ok(Snapshot {
346 workspaces,
347 commands: commands_map,
348 saved_pipelines,
349 flows,
350 flow_throughput,
351 fresh_stats,
352 quotas,
353 hydrate_history,
354 caps,
355 recent_log,
356 })
357 })
358}
359
360/// Render ASCII de sparkline para una serie de valores. `chars` define los
361/// glifos (orden ascendente). Auto-scales al máximo de la serie.
362fn sparkline(values: &[u64], width: usize) -> String {
363 if values.is_empty() {
364 return String::new();
365 }
366 const CHARS: &[char] = &['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█'];
367 let take = values.len().min(width);
368 let series = &values[values.len() - take..];
369 let max = *series.iter().max().unwrap_or(&1);
370 if max == 0 {
371 return "▁".repeat(take);
372 }
373 series
374 .iter()
375 .map(|v| {
376 let idx = ((*v as f64 / max as f64) * (CHARS.len() as f64 - 1.0)).round() as usize;
377 CHARS[idx.min(CHARS.len() - 1)]
378 })
379 .collect()
380}
381
382impl Render for Shell {
383 fn render(&mut self, _w: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
384 let theme = Theme::global(cx).clone();
385 let bg = theme.bg_app.clone();
386 let text = theme.fg_text;
387 let text_dim = theme.fg_muted;
388
389 let accent_up = gpui::rgb(0xa3be8c);
390 let accent_down = gpui::rgb(0xbf616a);
391 let accent_pending = gpui::rgb(0x6a7280);
392 let accent_info = gpui::rgb(0x88c0d0);
393
394 let header_text = format!(
395 "Daemon: {} · reload {} ms",
396 self.socket_path.display(),
397 self.last_probe_ms
398 );
399 let header = app_header(cx, header_text);
400
401 let status_banner = match &self.state {
402 DaemonState::Pending => None,
403 DaemonState::Down { reason } => Some(banner_themed(
404 cx,
405 Banner::Error,
406 SharedString::from(format!("Daemon DOWN — {reason}")),
407 )),
408 DaemonState::Up => Some(banner_themed(
409 cx,
410 Banner::Success,
411 SharedString::from("Daemon UP"),
412 )),
413 };
414
415 let (status_value, status_descr, status_accent) = match &self.state {
416 DaemonState::Pending => ("PENDING".to_string(), "primer probe…".to_string(), accent_pending),
417 DaemonState::Down { reason } => ("DOWN".to_string(), reason.clone(), accent_down),
418 DaemonState::Up => ("UP".to_string(), "shipote-daemon respondiendo".to_string(), accent_up),
419 };
420
421 let caps_items: Vec<String> = self
422 .caps
423 .as_ref()
424 .map(|c| {
425 vec![
426 format!(
427 "kernel: {}.{}.{}",
428 c.kernel_version.0, c.kernel_version.1, c.kernel_version.2
429 ),
430 format!("user_ns: {}", c.user_ns),
431 format!("cgroup_v2: {}", c.cgroup_v2),
432 format!("cgroup_delegated: {}", c.cgroup_delegated),
433 format!("cap_sys_admin: {}", c.has_cap_sys_admin),
434 ]
435 })
436 .unwrap_or_default();
437 let caps_value = if self.caps.is_some() { "OK".to_string() } else { "—".to_string() };
438
439 let ws_items: Vec<String> = self
440 .workspaces
441 .iter()
442 .map(|w| {
443 let key = w.id.to_string();
444 let history = self.stats_history.get(&key);
445 let rss_series: Vec<u64> = history
446 .map(|h| h.iter().map(|s| s.rss_bytes.unwrap_or(0)).collect())
447 .unwrap_or_default();
448 let spark = sparkline(&rss_series, STATS_HISTORY_LEN);
449 let latest = history.and_then(|h| h.back());
450 let (rss_now, peak, cpu_pct) = latest
451 .map(|s| (
452 s.rss_bytes.unwrap_or(0),
453 s.rss_peak_bytes.unwrap_or(0),
454 s.cpu_percent.unwrap_or(0.0),
455 ))
456 .unwrap_or((0, 0, 0.0));
457 let rss_mb = rss_now as f64 / 1024.0 / 1024.0;
458 let peak_mb = peak as f64 / 1024.0 / 1024.0;
459 format!(
460 "{:<14} {:<14} {} {:>6.1}M peak {:>6.1}M {:>5.1}%cpu",
461 &w.id.to_string()[20..],
462 w.label,
463 spark,
464 rss_mb,
465 peak_mb,
466 cpu_pct,
467 )
468 })
469 .collect();
470 let ws_count = self.workspaces.len().to_string();
471 let ws_descr = if self.workspaces.is_empty() {
472 "no hay workspaces vivos".to_string()
473 } else {
474 "id · label · cmds · uptime".to_string()
475 };
476
477 // Comandos: aplanamos por workspace, tomamos los más recientes (orden ULID ya temporal).
478 let mut cmd_items: Vec<String> = Vec::new();
479 let mut cmd_total = 0usize;
480 for (ws_id, cmds) in &self.commands {
481 cmd_total += cmds.len();
482 for c in cmds.iter().rev().take(8) {
483 let alive = if c.alive { "▶" } else { "✓" };
484 let exit = c
485 .exit_status
486 .map(|s| format!(" exit={s}"))
487 .unwrap_or_default();
488 cmd_items.push(format!(
489 "{} {} {:<20} pid={} logs={}B{}",
490 alive,
491 &ws_id[..6.min(ws_id.len())],
492 c.label,
493 c.pid,
494 c.log_bytes,
495 exit
496 ));
497 }
498 }
499 let cmd_count = cmd_total.to_string();
500 let cmd_descr = if cmd_total == 0 {
501 "no hay comandos lanzados".to_string()
502 } else {
503 "▶=alive ✓=exited · ws_prefix · label · pid · logs".to_string()
504 };
505
506 // Saved pipelines.
507 let saved_count = self.saved_pipelines.len().to_string();
508 let saved_items: Vec<String> = self.saved_pipelines.clone();
509 let saved_descr = if saved_items.is_empty() {
510 "shipote pipeline save <name> <file> para persistir".to_string()
511 } else {
512 "definiciones reusables vía run-saved".to_string()
513 };
514
515 // Quota breaches por workspace.
516 let mut breach_items: Vec<String> = Vec::new();
517 for (ws_id, q) in &self.quotas {
518 for b in &q.breaches {
519 let short = &ws_id[20..];
520 breach_items.push(format!("{short} {b}"));
521 }
522 }
523 let breach_count = breach_items.len().to_string();
524 let breach_descr = if breach_items.is_empty() {
525 "todos los workspaces dentro de quota".to_string()
526 } else {
527 "ws_suffix · recurso · uso > limit".to_string()
528 };
529
530 // Flow channels (data plane) con throughput.
531 let flow_count: usize = self.flows.iter().map(|f| f.sockets.len()).sum();
532 // Lookup helper que NO captura por ref (evita issue de borrow
533 // en el closure de flat_map).
534 let find_tp = |s: &std::path::PathBuf| -> (f64, f64) {
535 for t in &self.flow_throughput {
536 if t.socket == *s {
537 return (t.bytes_total as f64 / 1024.0, t.bytes_per_sec / 1024.0);
538 }
539 }
540 (0.0, 0.0)
541 };
542 let mut flow_items: Vec<String> = Vec::new();
543 for f in &self.flows {
544 let pipe = f.pipeline.to_string();
545 let short_pipe = &pipe[pipe.len() - 6..];
546 for s in &f.sockets {
547 let name = s
548 .file_name()
549 .map(|n| n.to_string_lossy().to_string())
550 .unwrap_or_else(|| s.display().to_string());
551 let (total_kib, rate_kib) = find_tp(s);
552 flow_items.push(format!(
553 "{short_pipe} {:<48} {:>7.1} KiB {:>6.2} KiB/s",
554 name, total_kib, rate_kib
555 ));
556 }
557 }
558 let flow_descr = if flow_count == 0 {
559 "pipelines con --tap exponen sockets aquí".to_string()
560 } else {
561 "pipe6 · socket · total · rate".to_string()
562 };
563
564 let body = div()
565 .flex()
566 .flex_col()
567 .gap(px(8.))
568 .px(px(16.))
569 .py(px(16.))
570 .child(stat_card(
571 cx,
572 "Estado",
573 status_value,
574 &status_descr,
575 status_accent,
576 text,
577 text_dim,
578 &[],
579 ))
580 .child(stat_card(
581 cx,
582 "Capabilities",
583 caps_value,
584 "kernel + namespaces + cgroup delegation",
585 accent_info,
586 text,
587 text_dim,
588 &caps_items,
589 ))
590 .child(stat_card(
591 cx,
592 "Workspaces",
593 ws_count,
594 &ws_descr,
595 accent_info,
596 text,
597 text_dim,
598 &ws_items,
599 ))
600 .child(stat_card(
601 cx,
602 "Comandos",
603 cmd_count,
604 &cmd_descr,
605 accent_info,
606 text,
607 text_dim,
608 &cmd_items,
609 ))
610 .child(stat_card(
611 cx,
612 "Saved pipelines",
613 saved_count,
614 &saved_descr,
615 accent_info,
616 text,
617 text_dim,
618 &saved_items,
619 ))
620 .child(stat_card(
621 cx,
622 "Flow channels",
623 flow_count.to_string(),
624 &flow_descr,
625 accent_up,
626 text,
627 text_dim,
628 &flow_items,
629 ))
630 .child(stat_card(
631 cx,
632 "Quota breaches",
633 breach_count,
634 &breach_descr,
635 if breach_items.is_empty() { accent_up } else { accent_down },
636 text,
637 text_dim,
638 &breach_items,
639 ));
640
641 // Live tail del comando más reciente con output.
642 let live_card = self.recent_log.as_ref().map(|(label, content)| {
643 // Cortamos a las últimas ~12 líneas para no inflar el panel.
644 let lines: Vec<String> = content
645 .lines()
646 .rev()
647 .take(12)
648 .map(|l| l.to_string())
649 .collect::<Vec<_>>()
650 .into_iter()
651 .rev()
652 .collect();
653 stat_card(
654 cx,
655 "Live tail",
656 label.clone(),
657 "últimas líneas del comando más reciente",
658 accent_up,
659 text,
660 text_dim,
661 &lines,
662 )
663 });
664
665 let body = body.when_some(live_card, |d, c| d.child(c));
666
667 div()
668 .flex()
669 .flex_col()
670 .size_full()
671 .bg(bg)
672 .child(header)
673 .when_some(status_banner, |d, b| d.child(b))
674 .child(body)
675 }
676}