refactor(naming): A1 — ente→arje, vista→revista, pluma→fana
1//! `shuma-daemon` — punto de entrada del runtime de shuma.
2//!
3//! Responsabilidades:
4//! - Escuchar el Unix socket admin (default: `$XDG_RUNTIME_DIR/shuma.sock`).
5//! - Despachar mensajes del [`shuma_protocol`] al [`WorkspaceManager`].
6//! - Reapear hijos periódicamente.
7//!
8//! Lo que NO hace en v1:
9//! - Sidecar al broker / handshake con Init (futuro: cuando un workspace
10//! exponga `service_socket`, anunciar al broker).
11//! - GUI (futuro `shuma-shell` con nahual_launcher).
12
13use anyhow::Context;
14use brahman_card::{Card, CardKind, Flow, Flows, Lifecycle, Payload, Supervision, TypeRef};
15use arje_incarnate::IncarnatorConfig;
16use shuma_core::WorkspaceManager;
17use shuma_discern::{DiscernPipeline, Hint};
18use shuma_protocol::{
19 default_socket_path, read_frame, write_frame, CommandInfo as ProtoCommandInfo,
20 EdgeDiscernmentInfo, FlowInfo, FlowThroughputInfo, QuotaReportInfo, Request, Response,
21 WorkspaceStatsInfo, WorkspaceSummary,
22};
23use std::sync::Arc;
24use tokio::net::{UnixListener, UnixStream};
25use tracing::{error, info, warn};
26
27#[tokio::main]
28async fn main() -> anyhow::Result<()> {
29 init_tracing();
30 let sock = default_socket_path();
31 if sock.exists() {
32 // Si ya existe, asumimos restart limpio. Si hubiera otro daemon vivo,
33 // bind fallaría con EADDRINUSE — más adelante: lockfile + check de PID.
34 let _ = std::fs::remove_file(&sock);
35 }
36 if let Some(parent) = sock.parent() {
37 let _ = std::fs::create_dir_all(parent);
38 }
39 let listener = UnixListener::bind(&sock).with_context(|| format!("bind {}", sock.display()))?;
40 info!(socket = %sock.display(), "shuma-daemon listening");
41 let daemon_started = std::time::Instant::now();
42
43 // Sidecar pool: una sesión global del daemon + N sesiones efímeras
44 // por edge enriquecido tras cada pipeline tap.
45 let sidecar_pool = match brahman_sidecar::SidecarPool::new() {
46 Ok(p) => Some(Arc::new(p)),
47 Err(e) => {
48 warn!(?e, "SidecarPool falló — broker integration disabled");
49 None
50 }
51 };
52 if let Some(pool) = &sidecar_pool {
53 pool.spawn(build_daemon_card(&sock));
54 }
55
56 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig {
57 // El daemon aún no se conecta al broker; cuando lo haga, este path
58 // se llenará desde el handshake.
59 bus_sock: None,
60 notify_socket: None,
61 extra_env: vec![("SHIPOTE_DAEMON".into(), "1".into())],
62 // strict_caps=false en v1: queremos UX permisiva (correr en non-root
63 // sin user_ns y avisar via warnings, no abortar).
64 strict_caps: false,
65 }));
66
67 // Restaurar snapshot previo si existe. Workspaces se recrean; los
68 // pids de comandos viejos NO se recuperan (kernel los mató). Los
69 // pipelines vivos (con supervisor) se relanzan desde cero.
70 let snapshot_path = shuma_core::persist::default_snapshot_path();
71 let restore = match mgr.restore_snapshot(&snapshot_path).await {
72 Ok(r) => r,
73 Err(e) => {
74 warn!(?e, "restore_snapshot falló — start fresh");
75 shuma_core::persist::RestoreOutcome::default()
76 }
77 };
78 // Relauncher de live_pipelines: como necesita inc+disc del daemon,
79 // lo hacemos acá tras el restore. Cada uno mismo flujo que un run
80 // normal — register_pipeline_commands + register_pipeline_supervisor.
81 for entry in restore.live_pipelines {
82 let inc = mgr.incarnator_handle();
83 let disc = Arc::new(DiscernPipeline::default_pipeline());
84 let workspace = entry.workspace;
85 let ws_label = mgr.workspace_label(workspace).await.unwrap_or_default();
86 let tap = entry.tap;
87 let spec = entry.spec;
88 match shuma_core::pipeline::run_pipeline(
89 &spec, &ws_label, tap, disc, inc, Some(mgr.clone()),
90 )
91 .await
92 {
93 Ok(launch) => {
94 mgr.register_pipeline_commands(workspace, launch.pipeline, launch.command_pids.clone()).await;
95 mgr.register_pipeline_supervisor(launch.pipeline, workspace, spec, tap).await;
96 info!(label = %launch.pipeline, "live pipeline relaunched from snapshot");
97 }
98 Err(e) => warn!(?e, "live pipeline relaunch failed"),
99 }
100 }
101
102 // Shutdown handler: SIGTERM/SIGINT → drain (stop_with_grace de todos
103 // los workspaces) → snapshot → exit. El drain usa grace=1s para dar
104 // chance a los comandos a terminar limpio antes del SIGKILL.
105 {
106 let mgr = mgr.clone();
107 let path = snapshot_path.clone();
108 tokio::spawn(async move {
109 let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
110 .expect("SIGTERM handler");
111 let mut int = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
112 .expect("SIGINT handler");
113 let sig_name = tokio::select! {
114 _ = term.recv() => "SIGTERM",
115 _ = int.recv() => "SIGINT",
116 };
117 info!(signal = sig_name, "shuma-daemon shutdown: draining workspaces");
118
119 // 1) Snapshot ANTES del drain — preserva intención declarada
120 // (los workspace specs siguen vivos en el snapshot aunque
121 // matemos los procesos hijos).
122 if let Err(e) = mgr.save_snapshot(&path).await {
123 warn!(?e, "save_snapshot falló");
124 }
125
126 // 2) Drain: stop_with_grace de todos los workspaces vivos.
127 // Grace 1s da chance a apps Type=notify de hacer cleanup.
128 let workspaces = mgr.list().await;
129 let n = workspaces.len();
130 for ws in workspaces {
131 if let Err(e) = mgr
132 .stop_with_grace(ws.id, std::time::Duration::from_millis(1000))
133 .await
134 {
135 warn!(?e, %ws.id, "stop_with_grace falló en drain");
136 }
137 }
138 info!(drained = n, "drain complete");
139 std::process::exit(0);
140 });
141 }
142
143 let discerner = Arc::new(DiscernPipeline::default_pipeline());
144
145 // Reaper periódico cada 500 ms. Además drena pipelines pendientes
146 // de restart (supervisión a nivel pipeline).
147 {
148 let mgr = mgr.clone();
149 tokio::spawn(async move {
150 let mut tick = tokio::time::interval(std::time::Duration::from_millis(500));
151 loop {
152 tick.tick().await;
153 mgr.reap_dead().await;
154 let pending = mgr.take_pending_restarts().await;
155 for sup in pending {
156 let backoff = std::time::Duration::from_millis(sup.current_backoff_ms);
157 info!(
158 label = %sup.spec.label,
159 restart_count = sup.restart_count,
160 backoff_ms = sup.current_backoff_ms,
161 "pipeline restart: relaunching after backoff"
162 );
163 // Backoff antes del relaunch — anti-thrash.
164 tokio::time::sleep(backoff).await;
165 let inc = mgr.incarnator_handle();
166 let disc = std::sync::Arc::new(DiscernPipeline::default_pipeline());
167 let workspace = sup.spec.workspace;
168 let ws_label = mgr.workspace_label(workspace).await.unwrap_or_default();
169 let tap = sup.tap;
170 let mut new_spec = sup.spec.clone();
171 new_spec.restart_on_failure = true;
172 // Escalar el backoff para la PRÓXIMA falla.
173 let next_backoff = (sup.current_backoff_ms * 2)
174 .min(new_spec.restart_max_backoff_ms);
175 match shuma_core::pipeline::run_pipeline(
176 &new_spec,
177 &ws_label,
178 tap,
179 disc,
180 inc,
181 Some(mgr.clone()),
182 )
183 .await
184 {
185 Ok(launch) => {
186 mgr.register_pipeline_commands(
187 workspace,
188 launch.pipeline,
189 launch.command_pids.clone(),
190 )
191 .await;
192 // Re-registrar supervisor con backoff escalado +
193 // restart_count preservado.
194 mgr.register_pipeline_supervisor_with_state(
195 launch.pipeline,
196 workspace,
197 new_spec,
198 tap,
199 sup.restart_count,
200 next_backoff,
201 )
202 .await;
203 }
204 Err(e) => {
205 warn!(?e, "pipeline restart failed");
206 }
207 }
208 }
209 }
210 });
211 }
212
213 // UID propio (para auth). SHIPOTE_TRUST_ANYONE=1 deshabilita.
214 let own_uid = nix::unistd::getuid().as_raw();
215 let trust_anyone = std::env::var("SHIPOTE_TRUST_ANYONE").as_deref() == Ok("1");
216 if trust_anyone {
217 warn!("SHIPOTE_TRUST_ANYONE=1 — accepting any peer uid");
218 }
219
220 loop {
221 match listener.accept().await {
222 Ok((stream, _)) => {
223 // Auth: SO_PEERCRED es automático en Unix sockets. Si
224 // el uid del peer no coincide con el nuestro, rechazo
225 // antes de procesar nada (a menos que esté permitido).
226 if !trust_anyone {
227 match peer_uid(&stream) {
228 Ok(peer) if peer == own_uid => {}
229 Ok(peer) => {
230 warn!(peer, own = own_uid, "rejecting peer with different uid");
231 drop(stream);
232 continue;
233 }
234 Err(e) => {
235 warn!(?e, "could not read peer uid — rejecting");
236 drop(stream);
237 continue;
238 }
239 }
240 }
241 let mgr = mgr.clone();
242 let disc = discerner.clone();
243 let pool = sidecar_pool.clone();
244 tokio::spawn(async move {
245 if let Err(e) = handle_client(stream, mgr, disc, pool, daemon_started).await {
246 warn!(?e, "client handler error");
247 }
248 });
249 }
250 Err(e) => {
251 error!(?e, "accept failed");
252 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
253 }
254 }
255 }
256}
257
258/// Lee SO_PEERCRED del Unix socket conectado. Devuelve el uid del peer.
259fn peer_uid(stream: &tokio::net::UnixStream) -> std::io::Result<u32> {
260 use std::os::fd::AsRawFd;
261 let fd = stream.as_raw_fd();
262 let mut ucred: libc::ucred = unsafe { std::mem::zeroed() };
263 let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
264 let r = unsafe {
265 libc::getsockopt(
266 fd,
267 libc::SOL_SOCKET,
268 libc::SO_PEERCRED,
269 &mut ucred as *mut _ as *mut _,
270 &mut len,
271 )
272 };
273 if r != 0 {
274 return Err(std::io::Error::last_os_error());
275 }
276 Ok(ucred.uid)
277}
278
279async fn handle_client(
280 mut stream: UnixStream,
281 mgr: Arc<WorkspaceManager>,
282 disc: Arc<DiscernPipeline>,
283 pool: Option<Arc<brahman_sidecar::SidecarPool>>,
284 daemon_started: std::time::Instant,
285) -> anyhow::Result<()> {
286 // Audit: peer uid lo leemos una vez aquí (no cambia durante la conexión).
287 let peer = peer_uid(&stream).unwrap_or(u32::MAX);
288 loop {
289 let req: Request = match read_frame(&mut stream).await {
290 Ok(r) => r,
291 Err(shuma_protocol::ProtocolError::Closed) => return Ok(()),
292 Err(e) => return Err(e.into()),
293 };
294 audit_request(peer, &req);
295 let resp = dispatch(&mgr, &disc, &pool, daemon_started, req).await;
296 write_frame(&mut stream, &resp).await?;
297 }
298}
299
300/// Path canónico del audit log: `$XDG_STATE_HOME/shuma/audit.log` o
301/// fallback `$HOME/.local/state/shuma/audit.log`.
302fn default_audit_log_path() -> std::path::PathBuf {
303 if let Ok(state) = std::env::var("XDG_STATE_HOME") {
304 return std::path::PathBuf::from(state).join("shuma/audit.log");
305 }
306 if let Ok(home) = std::env::var("HOME") {
307 return std::path::PathBuf::from(home).join(".local/state/shuma/audit.log");
308 }
309 std::path::PathBuf::from("/tmp/shuma-audit.log")
310}
311
312/// Cap del audit log antes de rotar a `audit.log.1`. 1 MiB.
313const AUDIT_LOG_MAX_BYTES: u64 = 1 << 20;
314
315/// Append + rotate (mueve a `.1` si supera el cap). Append-only, sin
316/// reordenar. Sync: cada line, fsync no — el log es defensive, no
317/// transactional.
318fn append_audit_line(path: &std::path::Path, line: &str) -> std::io::Result<()> {
319 use std::io::Write;
320 // Rotar si pasa el cap.
321 if let Ok(meta) = std::fs::metadata(path) {
322 if meta.len() >= AUDIT_LOG_MAX_BYTES {
323 let rotated = path.with_extension("log.1");
324 let _ = std::fs::rename(path, &rotated);
325 }
326 }
327 if let Some(parent) = path.parent() {
328 let _ = std::fs::create_dir_all(parent);
329 }
330 let mut f = std::fs::OpenOptions::new()
331 .create(true)
332 .append(true)
333 .open(path)?;
334 writeln!(f, "{line}")?;
335 Ok(())
336}
337
338/// Loguea cada mutación con target="audit" y el peer uid. Reads (ping,
339/// list, stats) se omiten para no inundar el log.
340fn audit_request(peer_uid: u32, req: &Request) {
341 let (action, detail) = match req {
342 Request::WorkspaceCreate { spec } => ("workspace.create", format!("label={}", spec.label)),
343 Request::WorkspaceStop { id, grace_ms } => ("workspace.stop", format!("id={id} grace_ms={grace_ms}")),
344 Request::Run { workspace, exec, restart_on_failure, .. } => (
345 "run",
346 format!("ws={workspace} exec={exec} restart={restart_on_failure}"),
347 ),
348 Request::PipelineRun { spec, tap, .. } => ("pipeline.run", format!("label={} tap={tap}", spec.label)),
349 Request::PipelineRunSaved { name, tap, .. } => ("pipeline.run-saved", format!("name={name} tap={tap}")),
350 Request::PipelineStop { pipeline, grace_ms } => ("pipeline.stop", format!("id={pipeline} grace_ms={grace_ms}")),
351 Request::PipelineSave { name, .. } => ("pipeline.save", format!("name={name}")),
352 Request::PipelineDrop { name } => ("pipeline.drop", format!("name={name}")),
353 Request::FlowDrop { pipeline } => ("flow.drop", format!("pipeline={pipeline}")),
354 // Reads (no audit):
355 Request::Ping
356 | Request::Health
357 | Request::WorkspaceList
358 | Request::WorkspaceStats { .. }
359 | Request::WorkspaceQuota { .. }
360 | Request::WorkspaceStatsHistory { .. }
361 | Request::WorkspaceFullSummary { .. }
362 | Request::CommandList { .. }
363 | Request::CommandLogs { .. }
364 | Request::PipelineSavedList
365 | Request::FlowList
366 | Request::FlowThroughput
367 | Request::Discern { .. }
368 | Request::Capabilities => return,
369 };
370 info!(target: "audit", uid = peer_uid, action, detail = %detail, "audit");
371 // Append a file. Failure no es fatal — sólo se pierde la entry.
372 let ts = std::time::SystemTime::now()
373 .duration_since(std::time::UNIX_EPOCH)
374 .map(|d| d.as_millis())
375 .unwrap_or(0);
376 let line = format!("ts={ts} uid={peer_uid} action={action} {detail}");
377 let path = default_audit_log_path();
378 if let Err(e) = append_audit_line(&path, &line) {
379 // Sólo loguear si el filesystem está roto. No reportar al cliente.
380 tracing::debug!(?e, path = %path.display(), "audit file write failed");
381 }
382}
383
384async fn dispatch(
385 mgr: &Arc<WorkspaceManager>,
386 disc: &DiscernPipeline,
387 pool: &Option<Arc<brahman_sidecar::SidecarPool>>,
388 daemon_started: std::time::Instant,
389 req: Request,
390) -> Response {
391 match req {
392 Request::Ping => Response::Pong,
393
394 Request::Health => {
395 let counts = mgr.health_counts().await;
396 Response::Health {
397 version: env!("CARGO_PKG_VERSION").to_string(),
398 uptime_ms: daemon_started.elapsed().as_millis() as u64,
399 alive_workspaces: counts.alive_workspaces,
400 alive_commands: counts.alive_commands,
401 alive_pipelines: counts.alive_pipelines,
402 active_flows: counts.active_flows,
403 dirty: mgr.is_dirty(),
404 }
405 }
406
407 Request::WorkspaceCreate { spec } => match mgr.create(spec).await {
408 Ok((id, warnings)) => Response::WorkspaceCreated { id, warnings },
409 Err(e) => Response::Error { message: format!("{e}") },
410 },
411
412 Request::WorkspaceList => {
413 let items = mgr
414 .list()
415 .await
416 .into_iter()
417 .map(|s| WorkspaceSummary {
418 id: s.id,
419 label: s.label,
420 commands: s.commands,
421 uptime_ms: s.uptime_ms,
422 })
423 .collect();
424 Response::WorkspaceList { items }
425 }
426
427 Request::WorkspaceStop { id, grace_ms } => {
428 match mgr
429 .stop_with_grace(id, std::time::Duration::from_millis(grace_ms))
430 .await
431 {
432 Ok(reaped) => Response::WorkspaceStopped { id, reaped },
433 Err(e) => Response::Error { message: format!("{e}") },
434 }
435 }
436
437 Request::Run { workspace, exec, argv, envp, restart_on_failure } => {
438 match mgr
439 .run_with_options(workspace, exec, argv, envp, restart_on_failure)
440 .await
441 {
442 Ok(s) => Response::RunStarted {
443 workspace,
444 command_id: s.id,
445 pid: s.pid,
446 },
447 Err(e) => Response::Error { message: format!("{e}") },
448 }
449 }
450
451 Request::PipelineRun { spec, tap, vars } => {
452 let vars_map: std::collections::HashMap<String, String> = vars.into_iter().collect();
453 let spec = match shuma_card::substitute_vars(&spec, &vars_map) {
454 Ok(s) => s,
455 Err(e) => return Response::Error { message: format!("template: {e}") },
456 };
457 let disc = DiscernPipeline::default_pipeline();
458 let inc = mgr.incarnator_handle();
459 let ws_label = mgr.workspace_label(spec.workspace).await.unwrap_or_default();
460 match shuma_core::pipeline::run_pipeline(
461 &spec,
462 &ws_label,
463 tap,
464 std::sync::Arc::new(disc),
465 inc,
466 Some(mgr.clone()),
467 )
468 .await
469 {
470 Ok(launch) => {
471 let pipeline_id = launch.pipeline;
472 announce_edges_to_broker(pool.as_deref(), &pipeline_id, &launch.edge_discernments);
473 let cmds = launch.command_pids;
474 mgr.register_pipeline_commands(spec.workspace, pipeline_id, cmds.clone()).await;
475 mgr.register_pipeline_supervisor(pipeline_id, spec.workspace, spec.clone(), tap).await;
476 let edges = launch.edge_discernments.into_iter().map(map_edge_to_info).collect();
477 Response::PipelineStarted {
478 pipeline: pipeline_id,
479 command_pids: cmds,
480 edges,
481 }
482 }
483 Err(e) => Response::Error { message: format!("{e}") },
484 }
485 }
486
487 Request::Discern { sample, hint_path } => {
488 let path_str = hint_path.as_ref().and_then(|p| p.to_str());
489 let hint = Hint {
490 path: path_str,
491 size_total: None,
492 };
493 match disc.discern(&sample, &hint) {
494 Some(d) => Response::Discernment {
495 ty: format!("{:?}", d.ty),
496 confidence: d.confidence,
497 mime: d.mime,
498 lens: d.lens,
499 },
500 None => Response::Error { message: "no discernment".into() },
501 }
502 }
503
504 Request::CommandList { workspace } => {
505 let items: Vec<ProtoCommandInfo> = mgr
506 .list_commands(workspace)
507 .await
508 .into_iter()
509 .map(|c| ProtoCommandInfo {
510 id: c.id,
511 label: c.label,
512 pid: c.pid,
513 alive: c.alive,
514 exit_status: c.exit_status,
515 log_bytes: c.log_bytes,
516 })
517 .collect();
518 Response::CommandList { items }
519 }
520
521 Request::CommandLogs { workspace, command, tail_bytes, stream } => {
522 let s = match stream.as_str() {
523 "stdout" => shuma_core::LogStream::Stdout,
524 "stderr" => shuma_core::LogStream::Stderr,
525 _ => shuma_core::LogStream::Both,
526 };
527 match mgr.get_command_logs(workspace, command, tail_bytes, s).await {
528 Some(bytes) => Response::CommandLogs { bytes },
529 None => Response::Error {
530 message: format!("no logs for command {command} in workspace {workspace}"),
531 },
532 }
533 }
534
535 Request::PipelineSave { name, spec } => {
536 mgr.save_pipeline(name.clone(), spec).await;
537 Response::PipelineSaved { name }
538 }
539
540 Request::PipelineSavedList => {
541 let names = mgr.list_saved_pipelines().await;
542 Response::PipelineSavedList { names }
543 }
544
545 Request::PipelineDrop { name } => {
546 let existed = mgr.drop_saved_pipeline(&name).await;
547 Response::PipelineDropped { name, existed }
548 }
549
550 Request::PipelineRunSaved { name, tap, vars } => match mgr.get_saved_pipeline(&name).await {
551 Some(spec) => {
552 let vars_map: std::collections::HashMap<String, String> = vars.into_iter().collect();
553 let spec = match shuma_card::substitute_vars(&spec, &vars_map) {
554 Ok(s) => s,
555 Err(e) => return Response::Error { message: format!("template: {e}") },
556 };
557 let disc = DiscernPipeline::default_pipeline();
558 let inc = mgr.incarnator_handle();
559 let ws_label = mgr.workspace_label(spec.workspace).await.unwrap_or_default();
560 match shuma_core::pipeline::run_pipeline(
561 &spec,
562 &ws_label,
563 tap,
564 std::sync::Arc::new(disc),
565 inc,
566 Some(mgr.clone()),
567 )
568 .await
569 {
570 Ok(launch) => {
571 let pipeline_id = launch.pipeline;
572 announce_edges_to_broker(pool.as_deref(), &pipeline_id, &launch.edge_discernments);
573 let cmds = launch.command_pids;
574 mgr.register_pipeline_commands(spec.workspace, pipeline_id, cmds.clone()).await;
575 mgr.register_pipeline_supervisor(pipeline_id, spec.workspace, spec.clone(), tap).await;
576 let edges = launch.edge_discernments.into_iter().map(map_edge_to_info).collect();
577 Response::PipelineStarted {
578 pipeline: pipeline_id,
579 command_pids: cmds,
580 edges,
581 }
582 }
583 Err(e) => Response::Error { message: format!("{e}") },
584 }
585 }
586 None => Response::Error {
587 message: format!("pipeline `{name}` no encontrado"),
588 },
589 },
590
591 Request::PipelineStop { pipeline, grace_ms } => {
592 let reaped = mgr
593 .stop_pipeline(pipeline, std::time::Duration::from_millis(grace_ms))
594 .await;
595 Response::PipelineStopped { pipeline, reaped }
596 }
597
598 Request::WorkspaceStats { workspace } => match mgr.workspace_stats(workspace).await {
599 Some(s) => Response::WorkspaceStats {
600 info: WorkspaceStatsInfo {
601 commands_alive: s.commands_alive,
602 commands_total: s.commands_total,
603 rss_bytes: s.rss_bytes,
604 rss_peak_bytes: s.rss_peak_bytes,
605 cpu_usec: s.cpu_usec,
606 cpu_percent: s.cpu_percent,
607 cpu_cores: s.cpu_cores,
608 source: s.source,
609 uptime_ms: s.uptime_ms,
610 },
611 },
612 None => Response::Error {
613 message: format!("workspace {workspace} not found"),
614 },
615 },
616
617 Request::WorkspaceStatsHistory { workspace, tail } => {
618 match mgr.workspace_stats_history(workspace, tail).await {
619 Some(samples) => {
620 let mapped: Vec<WorkspaceStatsInfo> = samples
621 .into_iter()
622 .map(|s| WorkspaceStatsInfo {
623 commands_alive: s.commands_alive,
624 commands_total: s.commands_total,
625 rss_bytes: s.rss_bytes,
626 rss_peak_bytes: s.rss_peak_bytes,
627 cpu_usec: s.cpu_usec,
628 cpu_percent: s.cpu_percent,
629 cpu_cores: s.cpu_cores,
630 source: s.source,
631 uptime_ms: s.uptime_ms,
632 })
633 .collect();
634 Response::WorkspaceStatsHistory { samples: mapped }
635 }
636 None => Response::Error {
637 message: format!("workspace {workspace} not found"),
638 },
639 }
640 }
641
642 Request::WorkspaceFullSummary { workspace } => {
643 let stats = match mgr.workspace_stats(workspace).await {
644 Some(s) => WorkspaceStatsInfo {
645 commands_alive: s.commands_alive,
646 commands_total: s.commands_total,
647 rss_bytes: s.rss_bytes,
648 rss_peak_bytes: s.rss_peak_bytes,
649 cpu_usec: s.cpu_usec,
650 cpu_percent: s.cpu_percent,
651 cpu_cores: s.cpu_cores,
652 source: s.source,
653 uptime_ms: s.uptime_ms,
654 },
655 None => return Response::Error { message: format!("workspace {workspace} not found") },
656 };
657 let quota = match mgr.workspace_quota(workspace).await {
658 Some(q) => QuotaReportInfo {
659 mem_limit: q.mem_limit,
660 nproc_limit: q.nproc_limit,
661 breaches: q.breaches,
662 },
663 None => QuotaReportInfo { mem_limit: None, nproc_limit: None, breaches: Vec::new() },
664 };
665 let commands = mgr
666 .list_commands(workspace)
667 .await
668 .into_iter()
669 .map(|c| ProtoCommandInfo {
670 id: c.id,
671 label: c.label,
672 pid: c.pid,
673 alive: c.alive,
674 exit_status: c.exit_status,
675 log_bytes: c.log_bytes,
676 })
677 .collect();
678 // Flow sockets de pipelines whose workspace == este.
679 let flow_sockets = mgr
680 .list_flow_pipelines()
681 .await
682 .into_iter()
683 .flat_map(|(_, sockets)| sockets)
684 .collect();
685 Response::WorkspaceFullSummary { stats, quota, commands, flow_sockets }
686 }
687
688 Request::WorkspaceQuota { workspace } => match mgr.workspace_quota(workspace).await {
689 Some(q) => Response::WorkspaceQuota {
690 info: QuotaReportInfo {
691 mem_limit: q.mem_limit,
692 nproc_limit: q.nproc_limit,
693 breaches: q.breaches,
694 },
695 },
696 None => Response::Error {
697 message: format!("workspace {workspace} not found"),
698 },
699 },
700
701 Request::FlowList => {
702 let items = mgr
703 .list_flow_pipelines()
704 .await
705 .into_iter()
706 .map(|(pipeline, sockets)| FlowInfo { pipeline, sockets })
707 .collect();
708 Response::FlowList { items }
709 }
710
711 Request::FlowThroughput => {
712 let items = mgr
713 .flow_throughput()
714 .await
715 .into_iter()
716 .map(|(socket, bytes_total, bytes_per_sec)| FlowThroughputInfo {
717 socket,
718 bytes_total,
719 bytes_per_sec,
720 })
721 .collect();
722 Response::FlowThroughput { items }
723 }
724
725 Request::FlowDrop { pipeline } => {
726 let existed = mgr.drop_pipeline_flows(pipeline).await;
727 Response::FlowDropped { pipeline, existed }
728 }
729
730 Request::Capabilities => {
731 let c = mgr.incarnator().capabilities();
732 Response::Capabilities {
733 kernel_version: c.kernel_version,
734 user_ns: format!("{:?}", c.user_ns),
735 cgroup_v2: format!("{:?}", c.cgroup_v2),
736 cgroup_delegated: c.cgroup_delegated,
737 has_cap_sys_admin: c.has_cap_sys_admin,
738 }
739 }
740 }
741}
742
743fn map_edge_to_info(e: shuma_core::pipeline::EdgeDiscernment) -> EdgeDiscernmentInfo {
744 EdgeDiscernmentInfo {
745 from_label: e.from_label,
746 from_output: e.from_output,
747 to_label: e.to_label,
748 to_input: e.to_input,
749 ty: e.discernment.as_ref().map(|d| format!("{:?}", d.ty)),
750 mime: e.discernment.as_ref().and_then(|d| d.mime.clone()),
751 lens: e.discernment.as_ref().and_then(|d| d.lens.clone()),
752 confidence: e.discernment.as_ref().map(|d| d.confidence).unwrap_or(0.0),
753 flow_socket: e.flow_socket,
754 }
755}
756
757/// Por cada edge con TypeRef detectado, spawneamos una Card efímera en el
758/// SidecarPool que se anuncia al broker como producer del TypeRef
759/// enriquecido. Esto permite a otros explorers (broker-explorer, etc.)
760/// ver que shuma vio JSON/text/wasm/etc. saliendo de un pipeline.
761fn announce_edges_to_broker(
762 pool: Option<&brahman_sidecar::SidecarPool>,
763 pipeline: &ulid::Ulid,
764 edges: &[shuma_core::pipeline::EdgeDiscernment],
765) {
766 let Some(pool) = pool else { return };
767 for e in edges {
768 let Some(d) = &e.discernment else { continue };
769 let label = format!(
770 "shuma.flow.{}.{}.{}.{}",
771 short_ulid(pipeline),
772 e.from_label,
773 e.from_output,
774 type_label(&d.ty)
775 );
776 let mut card = Card::new(label);
777 card.kind = CardKind::Data;
778 card.lifecycle = Lifecycle::Oneshot;
779 card.payload = Payload::Virtual;
780 card.supervision = Supervision::OneShot;
781 card.flow = Flows {
782 input: Vec::new(),
783 output: vec![Flow {
784 name: e.from_output.clone(),
785 ty: d.ty.clone(),
786 pin_to: None,
787 }],
788 };
789 pool.spawn(card);
790 info!(pipeline = %pipeline, from = %e.from_label, ty = ?d.ty, "edge announced to broker");
791 }
792}
793
794fn short_ulid(u: &ulid::Ulid) -> String {
795 let s = u.to_string();
796 s[s.len() - 6..].to_string()
797}
798
799fn type_label(t: &TypeRef) -> String {
800 match t {
801 TypeRef::Primitive { name } => name.clone(),
802 TypeRef::Wit { package, name, .. } => format!("{package}.{name}"),
803 }
804}
805
806/// Card del daemon. La presentamos al broker así otras sesiones pueden
807/// descubrir que shuma está corriendo y, eventualmente, conectarse
808/// como consumidoras del flow `workspaces` (futuro: que la GUI o el
809/// broker-explorer los listen vía broker en lugar de socket directo).
810fn build_daemon_card(service_socket: &std::path::Path) -> Card {
811 let mut card = Card::new("shuma.daemon");
812 card.kind = CardKind::Ente;
813 card.lifecycle = Lifecycle::Daemon;
814 card.payload = Payload::Virtual; // el daemon ya está corriendo (no es PID 1 quien lo encarna)
815 card.supervision = Supervision::Delegate;
816 card.service_socket = Some(service_socket.to_path_buf());
817 card.flow = Flows {
818 input: Vec::new(),
819 output: vec![
820 Flow {
821 name: "workspaces".into(),
822 ty: TypeRef::Wit {
823 package: "shuma:admin".into(),
824 interface: None,
825 name: "workspace-list".into(),
826 },
827 pin_to: None,
828 },
829 Flow {
830 name: "discern".into(),
831 ty: TypeRef::Wit {
832 package: "shuma:admin".into(),
833 interface: None,
834 name: "discernment".into(),
835 },
836 pin_to: None,
837 },
838 ],
839 };
840 card
841}
842
843fn init_tracing() {
844 use tracing_subscriber::{fmt, EnvFilter};
845 let filter = EnvFilter::try_from_env("SHIPOTE_LOG").unwrap_or_else(|_| EnvFilter::new("info"));
846 fmt().with_env_filter(filter).init();
847}