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