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