⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/modules/shipote/shipote-core/src/lib.rs
Historia
Quién tocó qué
crudo
feat(shipote): health endpoint + audit log + token-bucket real (fase R)
1//! `shipote-core` — runtime in-memory de Workspaces y comandos.
2//!
3//! Mantiene un estado tokio-friendly (Mutex sobre HashMap) con:
4//! - Workspaces vivos (id → state).
5//! - PIDs de comandos lanzados, indexados por workspace.
6//! - Reaping cooperativo: `reap_dead()` cosecha hijos terminados.
7
8// `pipeline` necesita `unsafe` puntual para `libc::close` y construir
9// `OwnedFd` desde fds que armamos con `pipe2(2)`. El resto del crate
10// permanece safe — el cargo lint `unsafe_code` queda permitido sólo en
11// el módulo concreto.
12#![deny(unsafe_op_in_unsafe_fn)]
13
14pub mod flow_channel;
15pub mod logbuf;
16pub mod persist;
17pub mod pipeline;
18pub mod stats;
19
20use brahman_card::{Card, Payload, Supervision};
21use ente_incarnate::{Incarnator, IncarnatorConfig};
22use nix::sys::signal::{kill, Signal};
23use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
24use nix::unistd::Pid;
25use shipote_card::{CommandRef, PipelineSpec, WorkspaceId, WorkspaceSpec};
26use std::collections::HashMap;
27use std::sync::Arc;
28use std::time::Instant;
29use thiserror::Error;
30use tokio::sync::Mutex;
31use tracing::{info, warn};
32use ulid::Ulid;
33
34#[derive(Debug, Error)]
35pub enum CoreError {
36 #[error("workspace {0} not found")]
37 WorkspaceNotFound(WorkspaceId),
38 #[error("compile: {0}")]
39 Compile(#[from] shipote_card::CompileError),
40 #[error("incarnate: {0}")]
41 Incarnate(#[from] ente_incarnate::IncarnateError),
42}
43
44#[derive(Debug)]
45pub struct WorkspaceState {
46 pub id: WorkspaceId,
47 pub spec: WorkspaceSpec,
48 pub root_card: Card,
49 pub commands: HashMap<Ulid, CommandState>,
50 pub started: Instant,
51 /// Última muestra de `(wall_instant, cpu_usec)` usada para calcular
52 /// `cpu_percent` en la próxima medición. None hasta el primer measure.
53 pub last_cpu_sample: Option<(Instant, u64)>,
54 /// Ring buffer de samples recientes para sparklines. Se popula cada
55 /// vez que `workspace_stats` se llama (típicamente desde el shell).
56 /// Cap 64 samples = ~2 minutos a 2s/sample.
57 pub stats_history: std::collections::VecDeque<stats::WorkspaceStats>,
58}
59
60const STATS_HISTORY_CAP: usize = 64;
61
62#[derive(Debug, Clone)]
63pub struct CommandState {
64 pub id: Ulid,
65 pub label: String,
66 pub pid: Pid,
67 pub alive: bool,
68 pub exit_status: Option<i32>,
69 /// Ring buffer del stdout. `None` para comandos sin captura.
70 pub stdout: Option<logbuf::LogBuf>,
71 /// Ring buffer del stderr. Separado de `stdout` para que el CLI
72 /// pueda filtrarlos. `None` para comandos sin captura.
73 pub stderr: Option<logbuf::LogBuf>,
74 /// Si el comando fue lanzado como parte de un Pipeline, su ULID.
75 pub pipeline_id: Option<Ulid>,
76}
77
78/// Stream a leer en `get_command_logs`. `Both` concatena stderr-después-stdout
79/// para una vista combinada (orden temporal aproximado).
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum LogStream {
82 Stdout,
83 Stderr,
84 Both,
85}
86
87pub struct WorkspaceManager {
88 inner: Arc<Mutex<Inner>>,
89 incarnator: Arc<Incarnator>,
90 /// True si hubo alguna mutación desde el último `save_snapshot`.
91 /// `save_snapshot` skip si false (snapshot incremental — evita
92 /// re-serialize cuando nada cambió, ej. SIGTERM tras un período idle).
93 dirty: std::sync::atomic::AtomicBool,
94}
95
96struct Inner {
97 workspaces: HashMap<WorkspaceId, WorkspaceState>,
98 /// Definiciones nombradas de pipelines persistidas. NO es lo mismo
99 /// que "pipelines vivos" — son specs guardados para reusar con
100 /// `run-saved`. Sobreviven restart vía snapshot.
101 saved_pipelines: HashMap<String, PipelineSpec>,
102 /// Flow channels vivos por pipeline. Se retienen hasta que el
103 /// pipeline termine — cuando todos los hijos del pipeline murieron,
104 /// el reaper los borra (futuro). v1: viven hasta `stop_pipeline_flows`
105 /// explícito o hasta shutdown.
106 pipeline_flows: HashMap<Ulid, Vec<crate::flow_channel::FlowChannel>>,
107 /// Specs de comandos `run()` con `restart_on_failure=true`. Indexed
108 /// por command_id. Cuando `reap_dead` detecta exit!=0, se relauncha
109 /// con la misma spec (nuevo pid y nuevo command_id se asigna por
110 /// el nuevo state pero el restart_spec sigue ligado al original).
111 restart_specs: HashMap<Ulid, RestartSpec>,
112 /// Supervisores de pipelines con `restart_on_failure`. Indexed por
113 /// pipeline_id. Cuando `reap_dead` detecta que el pipeline tuvo
114 /// algún command failure, agrega un entry a `pending_pipeline_restarts`.
115 pipeline_supervisors: HashMap<Ulid, PipelineSupervisor>,
116 /// Cola de pipelines pendientes de restart. El daemon la drena en
117 /// cada loop del reaper, hace stop + run_pipeline.
118 pending_pipeline_restarts: Vec<Ulid>,
119}
120
121#[derive(Debug, Clone)]
122pub struct PipelineSupervisor {
123 pub workspace: WorkspaceId,
124 pub spec: PipelineSpec,
125 pub tap: bool,
126 pub restart_count: u32,
127 /// Backoff actual (ms) — escala exponencialmente con cada restart.
128 pub current_backoff_ms: u64,
129}
130
131#[derive(Debug, Clone)]
132struct RestartSpec {
133 workspace: WorkspaceId,
134 exec: String,
135 argv: Vec<String>,
136 envp: Vec<(String, String)>,
137 /// Backoff inicial (ms). Crece exponencialmente hasta max_backoff_ms.
138 backoff_ms: u64,
139 max_backoff_ms: u64,
140 /// Cantidad de restarts ya ejecutados (para tracking).
141 restart_count: u32,
142}
143
144#[derive(Debug, Clone)]
145pub struct CommandSummary {
146 pub id: Ulid,
147 pub label: String,
148 pub pid: i32,
149}
150
151#[derive(Debug, Clone, Default)]
152pub struct HealthCounts {
153 pub alive_workspaces: u32,
154 pub alive_commands: u32,
155 pub alive_pipelines: u32,
156 pub active_flows: u32,
157}
158
159#[derive(Debug, Clone)]
160pub struct CommandInfo {
161 pub id: Ulid,
162 pub label: String,
163 pub pid: i32,
164 pub alive: bool,
165 pub exit_status: Option<i32>,
166 pub log_bytes: u64,
167}
168
169/// Lee VmRSS (bytes) de `/proc/<pid>/status`. Helper local para
170/// reap_dead que no necesita el full stats. Devuelve 0 si el proc no
171/// existe o el campo no aparece.
172fn read_proc_rss(pid: i32) -> Option<u64> {
173 let status = std::fs::read_to_string(format!("/proc/{pid}/status")).ok()?;
174 status
175 .lines()
176 .find_map(|l| l.strip_prefix("VmRSS:").map(str::trim))
177 .and_then(|s| s.split_whitespace().next())
178 .and_then(|s| s.parse::<u64>().ok())
179 .map(|kb| kb * 1024)
180}
181
182fn spawn_log_drainer(read_fd: std::os::fd::RawFd, logs: logbuf::LogBuf) {
183 // Marcar non-blocking + envolver en AsyncFd; igual patrón que el tap.
184 // SAFETY: F_SETFL sobre fd válido.
185 unsafe {
186 let flags = libc::fcntl(read_fd, libc::F_GETFL, 0);
187 if flags >= 0 {
188 libc::fcntl(read_fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
189 }
190 }
191 tokio::spawn(async move {
192 // SAFETY: ownership del fd transferido al drainer task.
193 let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(read_fd) };
194 let afd = match tokio::io::unix::AsyncFd::with_interest(owned, tokio::io::Interest::READABLE) {
195 Ok(a) => a,
196 Err(e) => {
197 tracing::warn!(?e, "log drainer AsyncFd failed");
198 return;
199 }
200 };
201 let mut buf = [0u8; 4096];
202 loop {
203 let mut guard = match afd.readable().await {
204 Ok(g) => g,
205 Err(_) => break,
206 };
207 use std::os::fd::AsRawFd;
208 let fd = afd.as_raw_fd();
209 // SAFETY: read sobre fd válido.
210 let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
211 if r > 0 {
212 logs.append(&buf[..r as usize]);
213 continue;
214 }
215 if r == 0 {
216 break; // EOF
217 }
218 let err = std::io::Error::last_os_error();
219 if err.kind() == std::io::ErrorKind::WouldBlock {
220 guard.clear_ready();
221 continue;
222 }
223 tracing::warn!(?err, "log drainer read err");
224 break;
225 }
226 });
227}
228
229trait OwnedFdFromRawCompat: Sized {
230 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self;
231}
232
233impl OwnedFdFromRawCompat for std::os::fd::OwnedFd {
234 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self {
235 use std::os::fd::FromRawFd;
236 // SAFETY: el caller transfiere ownership de fd a OwnedFd.
237 unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }
238 }
239}
240
241impl WorkspaceManager {
242 pub fn new(cfg: IncarnatorConfig) -> Self {
243 Self {
244 inner: Arc::new(Mutex::new(Inner {
245 workspaces: HashMap::new(),
246 saved_pipelines: HashMap::new(),
247 pipeline_flows: HashMap::new(),
248 restart_specs: HashMap::new(),
249 pipeline_supervisors: HashMap::new(),
250 pending_pipeline_restarts: Vec::new(),
251 })),
252 incarnator: Arc::new(Incarnator::new(cfg)),
253 dirty: std::sync::atomic::AtomicBool::new(false),
254 }
255 }
256
257 /// Marca el manager como dirty. Cualquier mutación que afecta al
258 /// snapshot debería llamar esto.
259 #[inline]
260 fn mark_dirty(&self) {
261 self.dirty.store(true, std::sync::atomic::Ordering::Relaxed);
262 }
263
264 /// True si hubo cambios desde el último `save_snapshot`. Útil para
265 /// chequeos cooperativos (ej. monitoring que pollea cada N).
266 pub fn is_dirty(&self) -> bool {
267 self.dirty.load(std::sync::atomic::Ordering::Relaxed)
268 }
269
270 /// Registra un supervisor para un pipeline con `restart_on_failure=true`.
271 /// El daemon llama esto tras `run_pipeline` para que `reap_dead` agregue
272 /// el pipeline a la cola de restart cuando algún command falle.
273 pub async fn register_pipeline_supervisor(
274 &self,
275 pipeline_id: Ulid,
276 workspace: WorkspaceId,
277 spec: PipelineSpec,
278 tap: bool,
279 ) {
280 if !spec.restart_on_failure {
281 return;
282 }
283 tracing::debug!(%pipeline_id, label = %spec.label, "pipeline supervisor registered");
284 let mut g = self.inner.lock().await;
285 let initial_backoff = spec.restart_backoff_ms.max(50);
286 g.pipeline_supervisors.insert(
287 pipeline_id,
288 PipelineSupervisor {
289 workspace,
290 spec,
291 tap,
292 restart_count: 0,
293 current_backoff_ms: initial_backoff,
294 },
295 );
296 drop(g);
297 self.mark_dirty();
298 }
299
300 /// Variante que preserva backoff/count del supervisor anterior (para
301 /// re-registrar tras un restart sin perder el throttle acumulado).
302 pub async fn register_pipeline_supervisor_with_state(
303 &self,
304 pipeline_id: Ulid,
305 workspace: WorkspaceId,
306 spec: PipelineSpec,
307 tap: bool,
308 restart_count: u32,
309 current_backoff_ms: u64,
310 ) {
311 if !spec.restart_on_failure {
312 return;
313 }
314 let mut g = self.inner.lock().await;
315 g.pipeline_supervisors.insert(
316 pipeline_id,
317 PipelineSupervisor {
318 workspace,
319 spec,
320 tap,
321 restart_count,
322 current_backoff_ms,
323 },
324 );
325 }
326
327 /// Drena la cola de pipelines pendientes de restart y retorna las
328 /// specs a relaunch. El daemon lo llama tras cada `reap_dead`.
329 ///
330 /// Aplica `restart_max`: si el supervisor ya pasó el límite, no se
331 /// retorna y el supervisor se elimina (give-up). El backoff
332 /// preserva el valor actual; el daemon decide cuándo aplicar el
333 /// sleep antes del relaunch.
334 pub async fn take_pending_restarts(&self) -> Vec<PipelineSupervisor> {
335 let mut g = self.inner.lock().await;
336 let pending = std::mem::take(&mut g.pending_pipeline_restarts);
337 let mut out = Vec::with_capacity(pending.len());
338 for old_id in pending {
339 if let Some(mut sup) = g.pipeline_supervisors.remove(&old_id) {
340 if sup.spec.restart_max > 0 && sup.restart_count >= sup.spec.restart_max {
341 tracing::warn!(
342 label = %sup.spec.label,
343 restart_count = sup.restart_count,
344 max = sup.spec.restart_max,
345 "pipeline restart_max reached — giving up"
346 );
347 continue; // no relaunch, supervisor discarded.
348 }
349 sup.restart_count += 1;
350 out.push(sup);
351 }
352 }
353 out
354 }
355
356 /// Registra los comandos lanzados por un pipeline en el workspace.
357 /// Esto permite `pipeline_stop` (matar selectivamente sólo los pids
358 /// de un pipeline). `pipeline_id` se setea en cada CommandState.
359 pub async fn register_pipeline_commands(
360 &self,
361 workspace: WorkspaceId,
362 pipeline_id: Ulid,
363 commands: Vec<(String, i32)>,
364 ) {
365 let mut g = self.inner.lock().await;
366 let Some(ws) = g.workspaces.get_mut(&workspace) else { return };
367 for (label, pid) in commands {
368 let cmd_id = Ulid::new();
369 ws.commands.insert(
370 cmd_id,
371 CommandState {
372 id: cmd_id,
373 label,
374 pid: Pid::from_raw(pid),
375 alive: true,
376 exit_status: None,
377 stdout: None,
378 stderr: None,
379 pipeline_id: Some(pipeline_id),
380 },
381 );
382 }
383 }
384
385 /// Detiene selectivamente los comandos de un pipeline. SIGTERM →
386 /// `grace` → SIGKILL. Devuelve cantidad reapeada. Si no hay comandos
387 /// del pipeline en ningún workspace, retorna 0.
388 pub async fn stop_pipeline(
389 &self,
390 pipeline_id: Ulid,
391 grace: std::time::Duration,
392 ) -> u32 {
393 // 1) Recolectamos pids de ese pipeline en todos los workspaces.
394 let mut targets: Vec<Pid> = Vec::new();
395 {
396 let g = self.inner.lock().await;
397 for ws in g.workspaces.values() {
398 for cmd in ws.commands.values() {
399 if cmd.alive && cmd.pipeline_id == Some(pipeline_id) {
400 targets.push(cmd.pid);
401 }
402 }
403 }
404 }
405 if targets.is_empty() {
406 return 0;
407 }
408 let initial = if grace.is_zero() { Signal::SIGKILL } else { Signal::SIGTERM };
409 for pid in &targets {
410 let _ = kill(*pid, initial);
411 }
412 let mut reaped = 0u32;
413 let mut still = targets.clone();
414 let deadline = std::time::Instant::now() + grace;
415 let poll = std::time::Duration::from_millis(20);
416 while !still.is_empty() && std::time::Instant::now() < deadline {
417 still.retain(|pid| match waitpid(*pid, Some(WaitPidFlag::WNOHANG)) {
418 Ok(WaitStatus::StillAlive) => true,
419 Ok(_) => {
420 reaped += 1;
421 false
422 }
423 Err(_) => false,
424 });
425 if !still.is_empty() {
426 tokio::time::sleep(poll).await;
427 }
428 }
429 for pid in &still {
430 let _ = kill(*pid, Signal::SIGKILL);
431 let _ = waitpid(*pid, None);
432 reaped += 1;
433 }
434 // Marcar como dead en estado in-memory.
435 let mut g = self.inner.lock().await;
436 for ws in g.workspaces.values_mut() {
437 for cmd in ws.commands.values_mut() {
438 if cmd.pipeline_id == Some(pipeline_id) && cmd.alive {
439 cmd.alive = false;
440 }
441 }
442 }
443 // Drop flows del pipeline.
444 g.pipeline_flows.remove(&pipeline_id);
445 info!(%pipeline_id, reaped, "pipeline stopped");
446 reaped
447 }
448
449 /// Retiene los FlowChannels de un pipeline para que sobrevivan al
450 /// fin del request. Drop = cierre del data plane.
451 pub async fn retain_pipeline_flows(
452 &self,
453 pipeline: Ulid,
454 flows: Vec<crate::flow_channel::FlowChannel>,
455 ) {
456 self.inner.lock().await.pipeline_flows.insert(pipeline, flows);
457 }
458
459 /// Snapshot de counts agregados para health endpoint.
460 pub async fn health_counts(&self) -> HealthCounts {
461 let g = self.inner.lock().await;
462 let alive_workspaces = g.workspaces.len() as u32;
463 let alive_commands: u32 = g
464 .workspaces
465 .values()
466 .map(|ws| ws.commands.values().filter(|c| c.alive).count() as u32)
467 .sum();
468 let alive_pipelines = g.pipeline_supervisors.len() as u32;
469 let active_flows: u32 = g.pipeline_flows.values().map(|v| v.len() as u32).sum();
470 HealthCounts {
471 alive_workspaces,
472 alive_commands,
473 alive_pipelines,
474 active_flows,
475 }
476 }
477
478 /// Lista pipelines vivos con sus sockets activos.
479 pub async fn list_flow_pipelines(&self) -> Vec<(Ulid, Vec<std::path::PathBuf>)> {
480 let g = self.inner.lock().await;
481 g.pipeline_flows
482 .iter()
483 .map(|(id, flows)| {
484 (
485 *id,
486 flows.iter().map(|f| f.socket_path().to_path_buf()).collect(),
487 )
488 })
489 .collect()
490 }
491
492 /// Throughput per-socket: bytes_total + bytes_per_sec por flow socket.
493 pub async fn flow_throughput(&self) -> Vec<(std::path::PathBuf, u64, f64)> {
494 let g = self.inner.lock().await;
495 let mut out = Vec::new();
496 for flows in g.pipeline_flows.values() {
497 for fc in flows {
498 out.push((
499 fc.socket_path().to_path_buf(),
500 fc.meter().total_bytes(),
501 fc.meter().bytes_per_sec(),
502 ));
503 }
504 }
505 out
506 }
507
508 /// Cierra el data plane de un pipeline (drop = remove_file de sockets).
509 pub async fn drop_pipeline_flows(&self, pipeline: Ulid) -> bool {
510 self.inner.lock().await.pipeline_flows.remove(&pipeline).is_some()
511 }
512
513 pub fn incarnator(&self) -> &Incarnator {
514 &self.incarnator
515 }
516
517 /// Handle Arc-clonable del Incarnator, para que el pipeline lo pueda
518 /// usar fuera del manager.
519 pub fn incarnator_handle(&self) -> Arc<Incarnator> {
520 self.incarnator.clone()
521 }
522
523 // -----------------------------------------------------------------
524 // Saved pipelines (definiciones nombradas, no runs)
525 // -----------------------------------------------------------------
526
527 /// Guarda (o reemplaza) un PipelineSpec bajo `name`.
528 pub async fn save_pipeline(&self, name: String, spec: PipelineSpec) {
529 self.inner.lock().await.saved_pipelines.insert(name, spec);
530 self.mark_dirty();
531 }
532
533 /// Devuelve los nombres de los pipelines guardados.
534 pub async fn list_saved_pipelines(&self) -> Vec<String> {
535 let g = self.inner.lock().await;
536 let mut v: Vec<String> = g.saved_pipelines.keys().cloned().collect();
537 v.sort();
538 v
539 }
540
541 /// Recupera el PipelineSpec guardado bajo `name`.
542 pub async fn get_saved_pipeline(&self, name: &str) -> Option<PipelineSpec> {
543 self.inner.lock().await.saved_pipelines.get(name).cloned()
544 }
545
546 /// Elimina un saved pipeline.
547 pub async fn drop_saved_pipeline(&self, name: &str) -> bool {
548 let existed = self.inner.lock().await.saved_pipelines.remove(name).is_some();
549 if existed {
550 self.mark_dirty();
551 }
552 existed
553 }
554
555 /// Label del workspace, si existe.
556 pub async fn workspace_label(&self, id: WorkspaceId) -> Option<String> {
557 self.inner
558 .lock()
559 .await
560 .workspaces
561 .get(&id)
562 .map(|w| w.spec.label.clone())
563 }
564
565 /// Compara accounting real (RSS, commands_alive) contra los rlimits
566 /// declarados en `SomaSpec`. Devuelve violaciones humanizadas. NO
567 /// hace enforcement automático.
568 pub async fn workspace_quota(&self, id: WorkspaceId) -> Option<stats::QuotaReport> {
569 let stats_now = self.workspace_stats(id).await?;
570 let g = self.inner.lock().await;
571 let ws = g.workspaces.get(&id)?;
572 let rl = &ws.spec.soma.rlimits;
573 let mut report = stats::QuotaReport {
574 mem_limit: rl.mem_bytes,
575 nproc_limit: rl.nproc,
576 breaches: Vec::new(),
577 };
578 if let (Some(limit), Some(used)) = (rl.mem_bytes, stats_now.rss_bytes) {
579 if used > limit {
580 report.breaches.push(format!(
581 "memory: {:.2} MiB > {:.2} MiB limit",
582 used as f64 / 1024.0 / 1024.0,
583 limit as f64 / 1024.0 / 1024.0,
584 ));
585 }
586 }
587 if let Some(limit) = rl.nproc {
588 if stats_now.commands_alive > limit {
589 report.breaches.push(format!(
590 "nproc: {} alive > {} limit",
591 stats_now.commands_alive, limit
592 ));
593 }
594 }
595 Some(report)
596 }
597
598 /// Estadísticas de recursos del workspace: RSS + CPU agregado de sus
599 /// comandos vivos. Lee `/proc/<pid>/` directamente; si el spec declara
600 /// `soma.cgroup.path`, también intenta el cgroup (más preciso, incluye
601 /// descendants).
602 ///
603 /// `cpu_percent` se calcula entre samples consecutivos. Necesita ≥2
604 /// llamadas para tener un valor (la primera siempre retorna `None`).
605 pub async fn workspace_stats(&self, id: WorkspaceId) -> Option<stats::WorkspaceStats> {
606 let mut g = self.inner.lock().await;
607 let ws = g.workspaces.get_mut(&id)?;
608 let alive: Vec<i32> = ws
609 .commands
610 .values()
611 .filter(|c| c.alive)
612 .map(|c| c.pid.as_raw())
613 .collect();
614 let total = ws.commands.len() as u32;
615 let cgroup_path = if ws.spec.soma.cgroup.path.is_empty() {
616 None
617 } else {
618 Some(std::path::PathBuf::from(format!(
619 "/sys/fs/cgroup{}",
620 ws.spec.soma.cgroup.path
621 )))
622 };
623 let mut s = stats::measure(&alive, cgroup_path.as_deref(), ws.started);
624 s.commands_total = total;
625
626 // CPU%: diff entre el sample actual y el previo, dividido por
627 // wall time. 100% = 1 core saturado. >100% = varios cores.
628 let now = Instant::now();
629 if let Some(cpu_now) = s.cpu_usec {
630 if let Some((prev_t, prev_cpu)) = ws.last_cpu_sample {
631 let dt_us = now.duration_since(prev_t).as_micros() as u64;
632 let d_cpu = cpu_now.saturating_sub(prev_cpu);
633 if dt_us > 0 {
634 s.cpu_percent = Some(100.0 * d_cpu as f32 / dt_us as f32);
635 }
636 }
637 ws.last_cpu_sample = Some((now, cpu_now));
638 }
639 // Append a history (ring buffer cap).
640 if ws.stats_history.len() >= STATS_HISTORY_CAP {
641 ws.stats_history.pop_front();
642 }
643 ws.stats_history.push_back(s.clone());
644 Some(s)
645 }
646
647 /// Retorna las últimas N samples de stats (servidas desde el ring
648 /// buffer interno). Sobrevive restart del shell.
649 pub async fn workspace_stats_history(
650 &self,
651 id: WorkspaceId,
652 tail: usize,
653 ) -> Option<Vec<stats::WorkspaceStats>> {
654 let g = self.inner.lock().await;
655 let ws = g.workspaces.get(&id)?;
656 let take = if tail == 0 { ws.stats_history.len() } else { tail };
657 let skip = ws.stats_history.len().saturating_sub(take);
658 Some(ws.stats_history.iter().skip(skip).cloned().collect())
659 }
660
661 pub async fn create(
662 self: &Arc<Self>,
663 spec: WorkspaceSpec,
664 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
665 self.create_with_id(WorkspaceId::new(), spec).await
666 }
667
668 /// Variante que acepta el ID. Útil para restore_snapshot: preserva
669 /// ULIDs entre restarts, así clients que tracking workspace_id no se
670 /// rompen.
671 pub async fn create_with_id(
672 self: &Arc<Self>,
673 id: WorkspaceId,
674 spec: WorkspaceSpec,
675 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
676 let card = spec.to_card(id)?;
677 let mut warnings = self.incarnator.dry_run(&card).warnings;
678 let ttl = spec.ttl;
679
680 // Si el workspace declara cgroup path Y rlimits, intentamos
681 // crear el cgroup y escribir memory.max/pids.max. El kernel
682 // hace OOM kill al exceder memory.max — enforcement automático
683 // sin policy adicional. Falla silenciosa si no hay delegation.
684 if !spec.soma.cgroup.path.is_empty() {
685 if let Ok(abs) = ente_incarnate::cgroup::ensure_cgroup(&spec.soma.cgroup) {
686 let applied =
687 ente_incarnate::cgroup::apply_rlimits_to_cgroup(&abs, &spec.soma.rlimits);
688 if !applied.is_empty() {
689 warnings.push(format!("cgroup limits applied: {}", applied.join(", ")));
690 }
691 }
692 }
693 let state = WorkspaceState {
694 id,
695 spec,
696 root_card: card,
697 commands: HashMap::new(),
698 started: Instant::now(),
699 last_cpu_sample: None,
700 stats_history: std::collections::VecDeque::with_capacity(STATS_HISTORY_CAP),
701 };
702 self.inner.lock().await.workspaces.insert(id, state);
703 self.mark_dirty();
704 info!(%id, ?ttl, "workspace created");
705
706 // Si tiene TTL, programar auto-stop. El task captura un weak ref
707 // al manager para no impedir que se dropée si el daemon termina.
708 if let Some(duration) = ttl {
709 let mgr_weak = Arc::downgrade(self);
710 tokio::spawn(async move {
711 tokio::time::sleep(duration).await;
712 if let Some(mgr) = mgr_weak.upgrade() {
713 let exists = mgr.inner.lock().await.workspaces.contains_key(&id);
714 if exists {
715 info!(%id, "workspace TTL expired — auto-stop");
716 let _ = mgr.stop(id).await;
717 }
718 }
719 });
720 }
721
722 Ok((id, warnings))
723 }
724
725 pub async fn list(&self) -> Vec<WorkspaceSnapshot> {
726 let g = self.inner.lock().await;
727 g.workspaces
728 .values()
729 .map(|w| WorkspaceSnapshot {
730 id: w.id,
731 label: w.spec.label.clone(),
732 commands: w.commands.len() as u32,
733 uptime_ms: w.started.elapsed().as_millis() as u64,
734 })
735 .collect()
736 }
737
738 pub async fn stop(&self, id: WorkspaceId) -> Result<u32, CoreError> {
739 self.stop_with_grace(id, std::time::Duration::from_millis(1000)).await
740 }
741
742 /// Variante con tiempo de gracia configurable. SIGTERM → espera `grace`
743 /// → SIGKILL si quedan vivos. `grace=0` = SIGKILL inmediato.
744 pub async fn stop_with_grace(
745 &self,
746 id: WorkspaceId,
747 grace: std::time::Duration,
748 ) -> Result<u32, CoreError> {
749 let mut g = self.inner.lock().await;
750 let ws = g.workspaces.remove(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
751 // También limpiamos flow_channels del workspace si los hubiera —
752 // por workspace lo retenemos por pipeline, no por workspace.
753 drop(g);
754 self.mark_dirty();
755
756 // 1) SIGTERM (o SIGKILL si grace=0) a todos vivos.
757 let initial_signal = if grace.is_zero() { Signal::SIGKILL } else { Signal::SIGTERM };
758 let alive_pids: Vec<Pid> = ws
759 .commands
760 .values()
761 .filter(|c| c.alive)
762 .map(|c| c.pid)
763 .collect();
764 for pid in &alive_pids {
765 let _ = kill(*pid, initial_signal);
766 }
767
768 // 2) Esperar hasta `grace` haciendo polling WNOHANG.
769 let mut reaped = 0u32;
770 let mut still_alive: Vec<Pid> = alive_pids.clone();
771 let deadline = std::time::Instant::now() + grace;
772 let poll_interval = std::time::Duration::from_millis(20);
773 while !still_alive.is_empty() && std::time::Instant::now() < deadline {
774 still_alive.retain(|pid| match waitpid(*pid, Some(WaitPidFlag::WNOHANG)) {
775 Ok(WaitStatus::StillAlive) => true,
776 Ok(_) => {
777 reaped += 1;
778 false
779 }
780 Err(_) => false,
781 });
782 if !still_alive.is_empty() {
783 tokio::time::sleep(poll_interval).await;
784 }
785 }
786
787 // 3) SIGKILL forzoso a los que quedan, y wait blocking.
788 for pid in &still_alive {
789 let _ = kill(*pid, Signal::SIGKILL);
790 let _ = waitpid(*pid, None);
791 reaped += 1;
792 }
793 info!(
794 %id,
795 reaped,
796 grace_ms = grace.as_millis() as u64,
797 sigkilled = still_alive.len(),
798 "workspace stopped"
799 );
800 Ok(reaped)
801 }
802
803 /// Ejecuta un comando one-shot dentro de un workspace existente.
804 /// Captura stdout+stderr en un ring buffer accesible vía
805 /// [`get_command_logs`](Self::get_command_logs).
806 pub async fn run(
807 &self,
808 id: WorkspaceId,
809 exec: String,
810 argv: Vec<String>,
811 envp: Vec<(String, String)>,
812 ) -> Result<CommandSummary, CoreError> {
813 self.run_with_options(id, exec, argv, envp, false).await
814 }
815
816 /// Variante con `restart_on_failure`: si el comando muere con
817 /// exit_status != 0, el reaper lo relauncha con backoff exponencial
818 /// (200ms → 400 → 800 → … cap 30s).
819 pub async fn run_with_options(
820 &self,
821 id: WorkspaceId,
822 exec: String,
823 argv: Vec<String>,
824 envp: Vec<(String, String)>,
825 restart_on_failure: bool,
826 ) -> Result<CommandSummary, CoreError> {
827 let workspace_label = {
828 let g = self.inner.lock().await;
829 let ws = g.workspaces.get(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
830 ws.spec.label.clone()
831 };
832 let cmd_ref = CommandRef {
833 label: format!("run-{}", short_ulid(&Ulid::new())),
834 payload: Payload::Native { exec, argv, envp },
835 soma: Default::default(),
836 flows: Default::default(),
837 supervision: Supervision::OneShot,
838 };
839 let card = cmd_ref.to_card(0, &workspace_label)?;
840
841 // Dos pipes O_CLOEXEC: uno para stdout, otro para stderr.
842 use std::os::fd::IntoRawFd;
843 let (sout_r, sout_w) =
844 nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).map_err(|e| {
845 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
846 })?;
847 let (serr_r, serr_w) =
848 nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).map_err(|e| {
849 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
850 })?;
851 let sout_r_fd = sout_r.into_raw_fd();
852 let sout_w_fd = sout_w.into_raw_fd();
853 let serr_r_fd = serr_r.into_raw_fd();
854 let serr_w_fd = serr_w.into_raw_fd();
855
856 let stdout_buf = logbuf::LogBuf::new();
857 let stderr_buf = logbuf::LogBuf::new();
858
859 let stdio = ente_incarnate::ChildStdio {
860 stdin_fd: None,
861 stdout_fd: Some(sout_w_fd),
862 stderr_fd: Some(serr_w_fd),
863 };
864 let out = self.incarnator.incarnate_with(&card, stdio)?;
865 let cmd_id = card.id;
866 let cmd_label = cmd_ref.label.clone();
867 let pid = out.pid;
868
869 spawn_log_drainer(sout_r_fd, stdout_buf.clone());
870 spawn_log_drainer(serr_r_fd, stderr_buf.clone());
871
872 let mut g = self.inner.lock().await;
873 if let Some(ws) = g.workspaces.get_mut(&id) {
874 ws.commands.insert(
875 cmd_id,
876 CommandState {
877 id: cmd_id,
878 label: cmd_label.clone(),
879 pid,
880 alive: true,
881 exit_status: None,
882 stdout: Some(stdout_buf),
883 stderr: Some(stderr_buf),
884 pipeline_id: None,
885 },
886 );
887 }
888 if restart_on_failure {
889 // Reextract exec/argv/envp del payload del CommandRef.
890 if let Payload::Native { exec, argv, envp } = &cmd_ref.payload {
891 g.restart_specs.insert(
892 cmd_id,
893 RestartSpec {
894 workspace: id,
895 exec: exec.clone(),
896 argv: argv.clone(),
897 envp: envp.clone(),
898 backoff_ms: 200,
899 max_backoff_ms: 30_000,
900 restart_count: 0,
901 },
902 );
903 }
904 }
905 for d in &out.degradations {
906 warn!(?d, %id, "command incarnation degradation");
907 }
908 Ok(CommandSummary {
909 id: cmd_id,
910 label: cmd_label,
911 pid: pid.as_raw(),
912 })
913 }
914
915 /// Devuelve el tail del log capturado para `(workspace, command)`.
916 /// `stream` selecciona stdout/stderr/both.
917 pub async fn get_command_logs(
918 &self,
919 workspace: WorkspaceId,
920 command: Ulid,
921 tail_bytes: usize,
922 stream: LogStream,
923 ) -> Option<Vec<u8>> {
924 let g = self.inner.lock().await;
925 let ws = g.workspaces.get(&workspace)?;
926 let cmd = ws.commands.get(&command)?;
927 match stream {
928 LogStream::Stdout => cmd.stdout.as_ref().map(|lb| lb.tail(tail_bytes)),
929 LogStream::Stderr => cmd.stderr.as_ref().map(|lb| lb.tail(tail_bytes)),
930 LogStream::Both => {
931 let so = cmd.stdout.as_ref().map(|lb| lb.tail(tail_bytes)).unwrap_or_default();
932 let se = cmd.stderr.as_ref().map(|lb| lb.tail(tail_bytes)).unwrap_or_default();
933 let mut out = so;
934 out.extend_from_slice(&se);
935 Some(out)
936 }
937 }
938 }
939
940 /// Lista comandos de un workspace.
941 pub async fn list_commands(&self, workspace: WorkspaceId) -> Vec<CommandInfo> {
942 let g = self.inner.lock().await;
943 let Some(ws) = g.workspaces.get(&workspace) else { return Vec::new() };
944 let mut out: Vec<CommandInfo> = ws
945 .commands
946 .values()
947 .map(|c| CommandInfo {
948 id: c.id,
949 label: c.label.clone(),
950 pid: c.pid.as_raw(),
951 alive: c.alive,
952 exit_status: c.exit_status,
953 log_bytes: c.stdout.as_ref().map(|l| l.written_total()).unwrap_or(0)
954 + c.stderr.as_ref().map(|l| l.written_total()).unwrap_or(0),
955 })
956 .collect();
957 // Orden estable por ULID (temporal).
958 out.sort_by_key(|c| c.id);
959 out
960 }
961
962 /// Lanza todas las Cards de un Pipeline. Devuelve (label, pid) por nodo.
963 /// La conexión via flows queda librada al broker (cuando haya integración
964 /// completa con sidecar; v1 sólo lanza).
965 pub async fn run_pipeline(
966 &self,
967 spec: &PipelineSpec,
968 ) -> Result<Vec<(String, Pid)>, CoreError> {
969 spec.validate()?;
970 let workspace_label = {
971 let g = self.inner.lock().await;
972 let ws = g
973 .workspaces
974 .get(&spec.workspace)
975 .ok_or(CoreError::WorkspaceNotFound(spec.workspace))?;
976 ws.spec.label.clone()
977 };
978 let mut launched = Vec::new();
979 for (i, node) in spec.nodes.iter().enumerate() {
980 let card = node.to_card(i, &workspace_label)?;
981 let out = self.incarnator.incarnate(&card)?;
982 let mut g = self.inner.lock().await;
983 if let Some(ws) = g.workspaces.get_mut(&spec.workspace) {
984 ws.commands.insert(
985 card.id,
986 CommandState {
987 id: card.id,
988 label: node.label.clone(),
989 pid: out.pid,
990 alive: true,
991 exit_status: None,
992 stdout: None, // run_pipeline NO captura (conecta por pipes).
993 stderr: None,
994 pipeline_id: None,
995 },
996 );
997 }
998 launched.push((node.label.clone(), out.pid));
999 }
1000 Ok(launched)
1001 }
1002
1003 /// Cosecha hijos terminados (no-bloqueante). Llamar periódicamente desde
1004 /// el daemon o ante SIGCHLD. Marca `alive=false` y guarda exit_status.
1005 pub async fn reap_dead(self: &Arc<Self>) {
1006 let mut to_restart: Vec<RestartSpec> = Vec::new();
1007 let mut to_enforce_kill: Vec<WorkspaceId> = Vec::new();
1008 {
1009 let mut g = self.inner.lock().await;
1010 for ws in g.workspaces.values_mut() {
1011 for cmd in ws.commands.values_mut() {
1012 if !cmd.alive {
1013 continue;
1014 }
1015 match waitpid(cmd.pid, Some(WaitPidFlag::WNOHANG)) {
1016 Ok(WaitStatus::Exited(_, code)) => {
1017 cmd.alive = false;
1018 cmd.exit_status = Some(code);
1019 }
1020 Ok(WaitStatus::Signaled(_, sig, _)) => {
1021 cmd.alive = false;
1022 cmd.exit_status = Some(128 + (sig as i32));
1023 }
1024 _ => {}
1025 }
1026 }
1027 }
1028 // Quota enforcement: chequear breach por workspace y aplicar policy.
1029 // Lo hacemos dentro del mismo lock para tener una lectura
1030 // consistente; el kill real va fuera del lock.
1031 for (ws_id, ws) in g.workspaces.iter() {
1032 let rl = &ws.spec.soma.rlimits;
1033 let qe = &ws.spec.quota_enforce;
1034 // Sólo aplicamos si hay al menos una action != None.
1035 if qe.mem == shipote_card::QuotaAction::None
1036 && qe.nproc == shipote_card::QuotaAction::None
1037 {
1038 continue;
1039 }
1040 // Medir RSS y nproc vivos sin pasar por workspace_stats
1041 // (que tomaría el lock recursivo). Hacemos un read directo.
1042 let alive: Vec<i32> = ws
1043 .commands
1044 .values()
1045 .filter(|c| c.alive)
1046 .map(|c| c.pid.as_raw())
1047 .collect();
1048 let nproc_alive = alive.len() as u32;
1049 let mem_used: u64 = alive
1050 .iter()
1051 .filter_map(|pid| read_proc_rss(*pid))
1052 .sum();
1053
1054 let mem_breach = matches!(rl.mem_bytes, Some(limit) if mem_used > limit);
1055 let nproc_breach = matches!(rl.nproc, Some(limit) if nproc_alive > limit);
1056
1057 let mut kill_needed = false;
1058 if mem_breach {
1059 match qe.mem {
1060 shipote_card::QuotaAction::Log => {
1061 warn!(%ws_id, used = mem_used, limit = ?rl.mem_bytes, "quota breach: memory");
1062 }
1063 shipote_card::QuotaAction::Kill => {
1064 warn!(%ws_id, used = mem_used, limit = ?rl.mem_bytes, "quota breach: KILLING");
1065 kill_needed = true;
1066 }
1067 _ => {}
1068 }
1069 }
1070 if nproc_breach {
1071 match qe.nproc {
1072 shipote_card::QuotaAction::Log => {
1073 warn!(%ws_id, alive = nproc_alive, limit = ?rl.nproc, "quota breach: nproc");
1074 }
1075 shipote_card::QuotaAction::Kill => {
1076 warn!(%ws_id, alive = nproc_alive, limit = ?rl.nproc, "quota breach: KILLING");
1077 kill_needed = true;
1078 }
1079 _ => {}
1080 }
1081 }
1082 if kill_needed {
1083 to_enforce_kill.push(*ws_id);
1084 }
1085 }
1086 // Pipeline supervisor: detectar pipelines cuyos comandos tienen
1087 // failure. Marca para restart si tiene supervisor.
1088 // Esto se hace cuando TODOS los comandos del pipeline están
1089 // dead Y al menos uno tiene exit!=0 (sino podría disparar
1090 // restart mientras otros comandos aún corren — incorrecto).
1091 let supervisor_ids: Vec<Ulid> = g.pipeline_supervisors.keys().copied().collect();
1092 for pipe_id in supervisor_ids {
1093 // ¿Hay algún comando vivo de este pipeline?
1094 let mut all_dead = true;
1095 let mut any_failed = false;
1096 for ws in g.workspaces.values() {
1097 for cmd in ws.commands.values() {
1098 if cmd.pipeline_id != Some(pipe_id) {
1099 continue;
1100 }
1101 if cmd.alive {
1102 all_dead = false;
1103 } else if cmd.exit_status.map_or(false, |s| s != 0) {
1104 any_failed = true;
1105 }
1106 }
1107 }
1108 if all_dead && any_failed {
1109 // Push a queue si no estaba ya.
1110 if !g.pending_pipeline_restarts.contains(&pipe_id) {
1111 g.pending_pipeline_restarts.push(pipe_id);
1112 }
1113 }
1114 }
1115 // Detectar restart_specs cuyo command_id ya está dead con exit!=0.
1116 let mut to_remove: Vec<Ulid> = Vec::new();
1117 for (cmd_id, spec) in g.restart_specs.iter() {
1118 let mut should_restart = false;
1119 let mut should_drop = false;
1120 'outer: for ws in g.workspaces.values() {
1121 if let Some(cmd) = ws.commands.get(cmd_id) {
1122 if !cmd.alive {
1123 match cmd.exit_status {
1124 Some(0) => should_drop = true,
1125 Some(_) => should_restart = true,
1126 None => {}
1127 }
1128 break 'outer;
1129 }
1130 }
1131 }
1132 if should_drop {
1133 to_remove.push(*cmd_id);
1134 } else if should_restart {
1135 to_restart.push(spec.clone());
1136 to_remove.push(*cmd_id);
1137 }
1138 }
1139 for id in to_remove {
1140 g.restart_specs.remove(&id);
1141 }
1142 }
1143 // Quota enforcement: kill workspaces fuera del lock.
1144 for ws_id in to_enforce_kill {
1145 let _ = self.stop_with_grace(ws_id, std::time::Duration::ZERO).await;
1146 }
1147 // Schedule restart fuera del lock.
1148 for mut spec in to_restart {
1149 let mgr = self.clone();
1150 let backoff = std::time::Duration::from_millis(spec.backoff_ms);
1151 // Subir el backoff para la PRÓXIMA falla, no esta.
1152 spec.backoff_ms = (spec.backoff_ms * 2).min(spec.max_backoff_ms);
1153 spec.restart_count += 1;
1154 let restart_n = spec.restart_count;
1155 tokio::spawn(async move {
1156 tokio::time::sleep(backoff).await;
1157 info!(
1158 backoff_ms = backoff.as_millis() as u64,
1159 restart = restart_n,
1160 "restarting failed command"
1161 );
1162 let workspace = spec.workspace;
1163 if let Err(e) = mgr
1164 .run_with_options(workspace, spec.exec.clone(), spec.argv.clone(), spec.envp.clone(), true)
1165 .await
1166 {
1167 warn!(?e, "restart failed to launch");
1168 return;
1169 }
1170 // Preservar backoff acumulado: localizar el nuevo command_id
1171 // (el más reciente vivo en el workspace) y sobreescribir.
1172 let new_cmd_id = {
1173 let g = mgr.inner.lock().await;
1174 g.workspaces.get(&workspace).and_then(|ws| {
1175 ws.commands
1176 .values()
1177 .filter(|c| c.alive)
1178 .max_by_key(|c| c.id)
1179 .map(|c| c.id)
1180 })
1181 };
1182 if let Some(new_id) = new_cmd_id {
1183 let mut g = mgr.inner.lock().await;
1184 if let Some(existing) = g.restart_specs.get_mut(&new_id) {
1185 existing.backoff_ms = spec.backoff_ms;
1186 existing.restart_count = spec.restart_count;
1187 }
1188 }
1189 });
1190 }
1191 }
1192}
1193
1194#[derive(Debug, Clone)]
1195pub struct WorkspaceSnapshot {
1196 pub id: WorkspaceId,
1197 pub label: String,
1198 pub commands: u32,
1199 pub uptime_ms: u64,
1200}
1201
1202fn short_ulid(u: &Ulid) -> String {
1203 let s = u.to_string();
1204 s[s.len() - 6..].to_string()
1205}
1206
1207#[cfg(test)]
1208mod tests {
1209 use super::*;
1210
1211 #[tokio::test]
1212 async fn ttl_auto_stops_workspace() {
1213 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1214 let spec = WorkspaceSpec {
1215 label: "ttl-test".into(),
1216 soma: Default::default(),
1217 permissions: Default::default(),
1218 ttl: Some(std::time::Duration::from_millis(120)),
1219 flow_dirs: vec![],
1220 on_exit: shipote_card::ExitPolicy::Reap,
1221 quota_enforce: Default::default(),
1222 };
1223 let (id, _) = mgr.create(spec).await.unwrap();
1224 assert_eq!(mgr.list().await.len(), 1);
1225 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
1226 assert_eq!(
1227 mgr.list().await.len(),
1228 0,
1229 "TTL expirado: workspace debe haber sido removido"
1230 );
1231 let _ = id;
1232 }
1233
1234 #[tokio::test]
1235 async fn create_and_list_workspace() {
1236 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1237 let spec = WorkspaceSpec {
1238 label: "test".into(),
1239 soma: Default::default(),
1240 permissions: Default::default(),
1241 ttl: None,
1242 flow_dirs: vec![],
1243 on_exit: shipote_card::ExitPolicy::Reap,
1244 quota_enforce: Default::default(),
1245 };
1246 let (id, _w) = mgr.create(spec).await.unwrap();
1247 let list = mgr.list().await;
1248 assert_eq!(list.len(), 1);
1249 assert_eq!(list[0].id, id);
1250 }
1251
1252 #[tokio::test]
1253 async fn run_captures_stdout_to_log() {
1254 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1255 let spec = WorkspaceSpec {
1256 label: "logs".into(),
1257 soma: Default::default(),
1258 permissions: Default::default(),
1259 ttl: None,
1260 flow_dirs: vec![],
1261 on_exit: shipote_card::ExitPolicy::Reap,
1262 quota_enforce: Default::default(),
1263 };
1264 let (id, _) = mgr.create(spec).await.unwrap();
1265 let summary = mgr
1266 .run(id, "/bin/echo".into(), vec!["captured-output".into()], vec![])
1267 .await
1268 .unwrap();
1269 for _ in 0..50 {
1270 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1271 mgr.reap_dead().await;
1272 let logs = mgr
1273 .get_command_logs(id, summary.id, 0, LogStream::Stdout)
1274 .await
1275 .unwrap_or_default();
1276 if !logs.is_empty() {
1277 let s = String::from_utf8_lossy(&logs);
1278 assert!(s.contains("captured-output"), "got: {s:?}");
1279 return;
1280 }
1281 }
1282 panic!("logs never captured");
1283 }
1284
1285 #[tokio::test]
1286 async fn run_captures_stderr_separately() {
1287 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1288 let spec = WorkspaceSpec {
1289 label: "stderr".into(),
1290 soma: Default::default(),
1291 permissions: Default::default(),
1292 ttl: None,
1293 flow_dirs: vec![],
1294 on_exit: shipote_card::ExitPolicy::Reap,
1295 quota_enforce: Default::default(),
1296 };
1297 let (id, _) = mgr.create(spec).await.unwrap();
1298 // sh -c "echo OUT; echo ERR >&2"
1299 let summary = mgr
1300 .run(
1301 id,
1302 "/bin/sh".into(),
1303 vec!["-c".into(), "echo OUT; echo ERR >&2".into()],
1304 vec![],
1305 )
1306 .await
1307 .unwrap();
1308 for _ in 0..50 {
1309 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1310 mgr.reap_dead().await;
1311 let so = mgr
1312 .get_command_logs(id, summary.id, 0, LogStream::Stdout)
1313 .await
1314 .unwrap_or_default();
1315 let se = mgr
1316 .get_command_logs(id, summary.id, 0, LogStream::Stderr)
1317 .await
1318 .unwrap_or_default();
1319 if !so.is_empty() && !se.is_empty() {
1320 let so_s = String::from_utf8_lossy(&so);
1321 let se_s = String::from_utf8_lossy(&se);
1322 assert!(so_s.contains("OUT"), "stdout: {so_s:?}");
1323 assert!(se_s.contains("ERR"), "stderr: {se_s:?}");
1324 assert!(!so_s.contains("ERR"), "stdout no debería tener ERR");
1325 assert!(!se_s.contains("OUT"), "stderr no debería tener OUT");
1326 return;
1327 }
1328 }
1329 panic!("logs never captured on both streams");
1330 }
1331
1332 #[tokio::test]
1333 async fn restart_on_failure_relaunches_failing_command() {
1334 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1335 let spec = WorkspaceSpec {
1336 label: "restart".into(),
1337 soma: Default::default(),
1338 permissions: Default::default(),
1339 ttl: None,
1340 flow_dirs: vec![],
1341 on_exit: shipote_card::ExitPolicy::Reap,
1342 quota_enforce: Default::default(),
1343 };
1344 let (id, _) = mgr.create(spec).await.unwrap();
1345 // /bin/false sale con exit=1. Con restart_on_failure=true debería
1346 // relanzarse al menos 1 vez (tras el backoff inicial de 200ms).
1347 let summary = mgr
1348 .run_with_options(id, "/bin/false".into(), vec![], vec![], true)
1349 .await
1350 .unwrap();
1351 let original_id = summary.id;
1352 // Esperamos ~500ms para que termine + reap + restart corra.
1353 for _ in 0..30 {
1354 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1355 mgr.reap_dead().await;
1356 let g = mgr.inner.lock().await;
1357 if let Some(ws) = g.workspaces.get(&id) {
1358 let new_cmds: Vec<_> = ws.commands.keys().filter(|k| **k != original_id).collect();
1359 if !new_cmds.is_empty() {
1360 // Hay un nuevo command_id → restart funcionó.
1361 return;
1362 }
1363 }
1364 }
1365 panic!("restart never launched a new command");
1366 }
1367
1368 #[tokio::test]
1369 async fn pipeline_supervisor_queues_restart_on_failure() {
1370 use shipote_card::{CommandRef, DiscernPolicy, PipelineSpec};
1371 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1372 let (ws_id, _) = mgr.create(WorkspaceSpec {
1373 label: "psup".into(),
1374 soma: Default::default(),
1375 permissions: Default::default(),
1376 ttl: None,
1377 flow_dirs: vec![],
1378 on_exit: shipote_card::ExitPolicy::Reap,
1379 quota_enforce: Default::default(),
1380 }).await.unwrap();
1381 let spec = PipelineSpec {
1382 label: "fail-pipeline".into(),
1383 workspace: ws_id,
1384 nodes: vec![CommandRef {
1385 label: "boom".into(),
1386 payload: brahman_card::Payload::Native {
1387 exec: "/bin/false".into(),
1388 argv: vec![],
1389 envp: vec![],
1390 },
1391 soma: Default::default(),
1392 flows: Default::default(),
1393 supervision: brahman_card::Supervision::OneShot,
1394 }],
1395 edges: vec![],
1396 discern: DiscernPolicy::default(),
1397 restart_on_failure: true,
1398 restart_backoff_ms: 200,
1399 restart_max_backoff_ms: 30_000,
1400 restart_max: 0,
1401 };
1402 let pipeline_id = ulid::Ulid::new();
1403 // Simulamos lo que haría el daemon: registramos un comando como
1404 // si fuera de pipeline. Usamos `register_pipeline_commands` con
1405 // un pid fake — pero como reaper hace waitpid, mejor lanzar de verdad.
1406 // Hack: usar /bin/false via run() y manualmente marcar pipeline_id.
1407 let summary = mgr.run(ws_id, "/bin/false".into(), vec![], vec![]).await.unwrap();
1408 // Marcar el comando con pipeline_id manualmente.
1409 {
1410 let mut g = mgr.inner.lock().await;
1411 if let Some(ws) = g.workspaces.get_mut(&ws_id) {
1412 if let Some(cmd) = ws.commands.get_mut(&summary.id) {
1413 cmd.pipeline_id = Some(pipeline_id);
1414 }
1415 }
1416 }
1417 mgr.register_pipeline_supervisor(pipeline_id, ws_id, spec, true).await;
1418 // Esperamos que reap detecte la falla y push a pending.
1419 for _ in 0..40 {
1420 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1421 mgr.reap_dead().await;
1422 let pending = mgr.take_pending_restarts().await;
1423 if !pending.is_empty() {
1424 assert_eq!(pending[0].spec.label, "fail-pipeline");
1425 return;
1426 }
1427 }
1428 panic!("supervisor never queued a restart");
1429 }
1430
1431 #[tokio::test]
1432 async fn quota_enforce_nproc_kill_terminates_commands() {
1433 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1434 let mut spec = WorkspaceSpec {
1435 label: "qenforce".into(),
1436 soma: Default::default(),
1437 permissions: Default::default(),
1438 ttl: None,
1439 flow_dirs: vec![],
1440 on_exit: shipote_card::ExitPolicy::Reap,
1441 quota_enforce: shipote_card::QuotaEnforcement {
1442 mem: shipote_card::QuotaAction::None,
1443 nproc: shipote_card::QuotaAction::Kill,
1444 },
1445 };
1446 spec.soma.rlimits.nproc = Some(1);
1447 let (id, _) = mgr.create(spec).await.unwrap();
1448 // Lanzo 2 procesos (cada uno sleep). nproc_limit=1 → breach inmediato.
1449 let _ = mgr.run(id, "/bin/sleep".into(), vec!["5".into()], vec![]).await.unwrap();
1450 let _ = mgr.run(id, "/bin/sleep".into(), vec!["5".into()], vec![]).await.unwrap();
1451 // Reaper detecta breach y mata workspace.
1452 for _ in 0..30 {
1453 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1454 mgr.reap_dead().await;
1455 let alive = mgr.list().await;
1456 if alive.is_empty() {
1457 return; // workspace removido por stop()
1458 }
1459 }
1460 panic!("quota enforce kill never triggered");
1461 }
1462
1463 #[tokio::test]
1464 async fn workspace_stats_history_accumulates() {
1465 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1466 let spec = WorkspaceSpec {
1467 label: "history".into(),
1468 soma: Default::default(),
1469 permissions: Default::default(),
1470 ttl: None,
1471 flow_dirs: vec![],
1472 on_exit: shipote_card::ExitPolicy::Reap,
1473 quota_enforce: Default::default(),
1474 };
1475 let (id, _) = mgr.create(spec).await.unwrap();
1476 // Necesitamos al menos un comando vivo para que `measure` no
1477 // retorne source=none (que igual se appendea, pero con stats vacíos).
1478 let _ = mgr
1479 .run(id, "/bin/sleep".into(), vec!["5".into()], vec![])
1480 .await
1481 .unwrap();
1482 // Llamar stats 5 veces.
1483 for _ in 0..5 {
1484 let _ = mgr.workspace_stats(id).await;
1485 }
1486 let history = mgr.workspace_stats_history(id, 0).await.unwrap();
1487 assert_eq!(history.len(), 5, "history debería tener 5 samples");
1488 // tail=3 retorna los últimos 3.
1489 let tail3 = mgr.workspace_stats_history(id, 3).await.unwrap();
1490 assert_eq!(tail3.len(), 3);
1491 // Cleanup.
1492 let _ = mgr.stop_with_grace(id, std::time::Duration::ZERO).await;
1493 }
1494
1495 #[tokio::test]
1496 async fn run_true_in_workspace() {
1497 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1498 let spec = WorkspaceSpec {
1499 label: "exec".into(),
1500 soma: Default::default(),
1501 permissions: Default::default(),
1502 ttl: None,
1503 flow_dirs: vec![],
1504 on_exit: shipote_card::ExitPolicy::Reap,
1505 quota_enforce: Default::default(),
1506 };
1507 let (id, _) = mgr.create(spec).await.unwrap();
1508 let summary = mgr
1509 .run(id, "/bin/true".into(), vec![], vec![])
1510 .await
1511 .unwrap();
1512 assert!(summary.pid > 0);
1513 // Cosecha.
1514 std::thread::sleep(std::time::Duration::from_millis(100));
1515 mgr.reap_dead().await;
1516 }
1517}