tawasuyu/shuma
Interactive shell with zsh/fish parity, native multiplexing + remote sessions, on Llimphi.
⎇ 1e5d3ed2397d6efbb71aa34380f170471508cfd7
02_ruway/shuma/sandbox/shuma-card/src/lib.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat: shuma standalone — shell interactiva sobre Llimphi (front-door, git-dep al monorepo)
1//! `shuma-card` — tipos del runtime shuma.
2//!
3//! Tres entidades nuevas encima del `brahman-card::Card`:
4//!
5//! - [`WorkspaceSpec`] — espacio aislado raíz con su propio `SomaSpec`.
6//! - [`CommandRef`] — un comando dentro de un workspace.
7//! - [`PipelineSpec`] — DAG de `CommandRef` conectados por `FlowEdge`.
8//!
9//! Cada `WorkspaceSpec`/`CommandRef` se **compila** a una o varias
10//! [`card_core::Card`] que el daemon entrega al [`Incarnator`] de
11//! `ente-incarnate`. Esto preserva el contrato canónico del fractal.
12
13#![forbid(unsafe_code)]
14
15use card_core::{Card, Payload, Permissions, SomaSpec, Supervision};
16use sandokan_core::{Intent, IsolationLevel};
17use serde::{Deserialize, Serialize};
18use std::time::Duration;
19use thiserror::Error;
20use ulid::Ulid;
21
22// =====================================================================
23// Identidades
24// =====================================================================
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
27pub struct WorkspaceId(pub Ulid);
28
29impl WorkspaceId {
30 pub fn new() -> Self {
31 Self(Ulid::new())
32 }
33}
34
35impl Default for WorkspaceId {
36 fn default() -> Self {
37 Self::new()
38 }
39}
40
41impl std::fmt::Display for WorkspaceId {
42 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43 self.0.fmt(f)
44 }
45}
46
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
48pub struct PipelineId(pub Ulid);
49
50impl PipelineId {
51 pub fn new() -> Self {
52 Self(Ulid::new())
53 }
54}
55
56impl Default for PipelineId {
57 fn default() -> Self {
58 Self::new()
59 }
60}
61
62impl std::fmt::Display for PipelineId {
63 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64 self.0.fmt(f)
65 }
66}
67
68// =====================================================================
69// Workspace
70// =====================================================================
71
72/// Espacio aislado de shuma. Es la raíz de aislamiento — cualquier comando
73/// que corre dentro hereda restricciones y no puede aflojarlas.
74#[derive(Debug, Clone, Serialize, Deserialize)]
75pub struct WorkspaceSpec {
76 pub label: String,
77
78 /// Aislamiento del workspace mismo (cuando se materializa como Card raíz).
79 #[serde(default)]
80 pub soma: SomaSpec,
81
82 /// Permisos máximos para hijas. Hijas pueden bajar pero no subir.
83 #[serde(default)]
84 pub permissions: Permissions,
85
86 /// `None` = vive hasta `stop`. `Some(d)` = el daemon lo termina tras d.
87 #[serde(default, with = "opt_duration_millis")]
88 pub ttl: Option<Duration>,
89
90 /// Slots de flow pre-declarados. Limitan qué consumidores externos al
91 /// workspace pueden empatar contra los productores internos.
92 #[serde(default)]
93 pub flow_dirs: Vec<FlowSlot>,
94
95 /// Política al terminar el workspace.
96 #[serde(default)]
97 pub on_exit: ExitPolicy,
98
99 /// Política de enforcement automático cuando un recurso excede su
100 /// rlimit declarado en `soma.rlimits`. Default = sólo accounting
101 /// (None) — el quota report sigue funcionando, pero no hay kill.
102 #[serde(default)]
103 pub quota_enforce: QuotaEnforcement,
104}
105
106/// Acción cuando un recurso excede su límite. Aplica por recurso (mem,
107/// nproc, ...).
108#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
109#[serde(rename_all = "lowercase")]
110pub enum QuotaAction {
111 /// Sólo accounting: la breach aparece en `workspace_quota`.
112 #[default]
113 None,
114 /// Loguear la breach (info-level del daemon).
115 Log,
116 /// Matar todos los comandos vivos del workspace (SIGKILL, sin grace).
117 Kill,
118}
119
120#[derive(Debug, Clone, Default, Serialize, Deserialize)]
121pub struct QuotaEnforcement {
122 #[serde(default)]
123 pub mem: QuotaAction,
124 #[serde(default)]
125 pub nproc: QuotaAction,
126}
127
128#[derive(Debug, Clone, Serialize, Deserialize)]
129pub struct FlowSlot {
130 pub name: String,
131 pub direction: FlowDirection,
132 /// Si `Workspace`, sólo otros nodos del mismo workspace pueden empatar.
133 /// Si `Public`, el broker global puede emparejar.
134 #[serde(default)]
135 pub scope: FlowScope,
136}
137
138#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(rename_all = "lowercase")]
140pub enum FlowDirection {
141 Input,
142 Output,
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
146#[serde(rename_all = "lowercase")]
147pub enum FlowScope {
148 #[default]
149 Workspace,
150 Public,
151}
152
153#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
154#[serde(rename_all = "kebab-case")]
155pub enum ExitPolicy {
156 /// Reapear procesos hijos y descartar estado.
157 #[default]
158 Reap,
159 /// Mantener el workspace en `stopped` para inspección.
160 Keep,
161 /// Tomar snapshot del estado (para restart posterior).
162 Snapshot,
163}
164
165mod opt_duration_millis {
166 use serde::{Deserialize, Deserializer, Serialize, Serializer};
167 use std::time::Duration;
168
169 pub fn serialize<S: Serializer>(d: &Option<Duration>, s: S) -> Result<S::Ok, S::Error> {
170 d.map(|x| x.as_millis() as u64).serialize(s)
171 }
172
173 pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Duration>, D::Error> {
174 let v: Option<u64> = Option::deserialize(d)?;
175 Ok(v.map(Duration::from_millis))
176 }
177}
178
179// =====================================================================
180// CommandRef
181// =====================================================================
182
183/// Un comando que vive dentro de un workspace. Se compila a una `Card` con
184/// `pin_to` apuntando al workspace padre (label) y su `SomaSpec`
185/// intersectado con el del workspace.
186#[derive(Debug, Clone, Serialize, Deserialize)]
187pub struct CommandRef {
188 pub label: String,
189 pub payload: Payload,
190
191 /// SomaSpec del comando. El compilador lo intersecta con el del workspace.
192 #[serde(default)]
193 pub soma: SomaSpec,
194
195 /// Inputs/outputs tipados (mismos `Flow` de brahman-card).
196 #[serde(default)]
197 pub flows: card_core::Flows,
198
199 /// Política de supervisión. Default `OneShot` (un comando se ejecuta y muere).
200 #[serde(default = "default_oneshot")]
201 pub supervision: Supervision,
202}
203
204fn default_oneshot() -> Supervision {
205 Supervision::OneShot
206}
207
208// =====================================================================
209// Pipeline
210// =====================================================================
211
212#[derive(Debug, Clone, Serialize, Deserialize)]
213pub struct PipelineSpec {
214 pub label: String,
215 pub workspace: WorkspaceId,
216 pub nodes: Vec<CommandRef>,
217 #[serde(default)]
218 pub edges: Vec<FlowEdge>,
219 #[serde(default)]
220 pub discern: DiscernPolicy,
221 /// Si `true` y cualquier comando del pipeline termina con exit!=0,
222 /// el daemon relaunch el pipeline ENTERO (stop + nuevo run_pipeline).
223 /// Útil para pipelines de procesamiento continuo.
224 #[serde(default)]
225 pub restart_on_failure: bool,
226 /// Backoff inicial entre restarts (ms). Crece exponencialmente
227 /// hasta `restart_max_backoff_ms`. Default 200ms = ~5 restarts/s
228 /// inicial, escalando rápido.
229 #[serde(default = "default_restart_backoff")]
230 pub restart_backoff_ms: u64,
231 /// Backoff máximo (ms). Default 30s. El backoff no crece más allá.
232 #[serde(default = "default_restart_max_backoff")]
233 pub restart_max_backoff_ms: u64,
234 /// Máximo de restarts antes de dar up. `0` = infinito. Default 0.
235 /// Útil para fail-loud cuando un pipeline siempre falla.
236 #[serde(default)]
237 pub restart_max: u32,
238}
239
240fn default_restart_backoff() -> u64 {
241 200
242}
243fn default_restart_max_backoff() -> u64 {
244 30_000
245}
246
247#[derive(Debug, Clone, Serialize, Deserialize)]
248pub struct FlowEdge {
249 /// Índice en `PipelineSpec.nodes` del productor.
250 pub from: usize,
251 /// Nombre del Flow output del productor.
252 pub from_output: String,
253 /// Índice en `PipelineSpec.nodes` del consumidor.
254 pub to: usize,
255 /// Nombre del Flow input del consumidor.
256 pub to_input: String,
257}
258
259#[derive(Debug, Clone, Serialize, Deserialize)]
260pub struct DiscernPolicy {
261 /// Bytes a samplear por flow para el discernidor. Default 4 KiB.
262 #[serde(default = "default_sample_bytes")]
263 pub sample_bytes: usize,
264 /// Si `true`, enriquece la Card del producer con el TypeRef detectado.
265 #[serde(default = "default_true")]
266 pub enrich_producer: bool,
267 /// Chunks que el FlowChannel guarda en replay buffer para subscribers
268 /// tarde. Default 32. Subir si los productores escriben en ráfagas y
269 /// querés que los consumidores tardíos vean toda la salida.
270 #[serde(default = "default_replay_chunks")]
271 pub replay_chunks: usize,
272 /// Tope adicional por **bytes** acumulados en el replay buffer. Lo
273 /// que se exceda primero (chunks o bytes) drop-ea el chunk más viejo.
274 /// `0` = sin tope por bytes (sólo aplica `replay_chunks`). Útil para
275 /// productores con chunks de tamaño variable.
276 #[serde(default)]
277 pub replay_bytes: usize,
278 /// Rate-limit del flow channel (bytes/s). `0` = sin límite. Si está
279 /// definido, el splitter sleeps proporcional al tamaño del chunk
280 /// antes de re-broadcastear. Protege subscribers lentos.
281 #[serde(default)]
282 pub max_bytes_per_sec: u64,
283}
284
285impl Default for DiscernPolicy {
286 fn default() -> Self {
287 Self {
288 sample_bytes: default_sample_bytes(),
289 enrich_producer: default_true(),
290 replay_chunks: default_replay_chunks(),
291 replay_bytes: 0,
292 max_bytes_per_sec: 0,
293 }
294 }
295}
296
297fn default_sample_bytes() -> usize {
298 4096
299}
300fn default_true() -> bool {
301 true
302}
303fn default_replay_chunks() -> usize {
304 32
305}
306
307// =====================================================================
308// Compilación a Card
309// =====================================================================
310
311#[derive(Debug, Error)]
312pub enum CompileError {
313 #[error("workspace label vacío")]
314 EmptyWorkspaceLabel,
315 #[error("comando con label vacío en posición {0}")]
316 EmptyCommandLabel(usize),
317 #[error("edge fuera de rango: from={from}, to={to}, nodes={nodes}")]
318 EdgeOutOfBounds { from: usize, to: usize, nodes: usize },
319}
320
321impl WorkspaceSpec {
322 /// Compila el WorkspaceSpec a una Card raíz que el Incarnator puede
323 /// encarnar. Usa `Payload::Virtual` (el workspace no es un proceso por
324 /// sí solo; sólo aloja hijos).
325 pub fn to_card(&self, id: WorkspaceId) -> Result<Card, CompileError> {
326 if self.label.trim().is_empty() {
327 return Err(CompileError::EmptyWorkspaceLabel);
328 }
329 let mut c = Card::new(format!("shuma.workspace.{}", self.label));
330 c.id = id.0;
331 c.soma = self.soma.clone();
332 c.permissions = self.permissions.clone();
333 c.payload = Payload::Virtual;
334 c.supervision = Supervision::OneShot;
335 Ok(c)
336 }
337
338 /// Puente al orquestador único (sandokan): compila el workspace a una
339 /// `Intent` que cualquier `sandokan_core::Engine` (LocalEngine en este
340 /// host, DaemonEngine o RemoteEngine) puede encarnar. Es la Card
341 /// `Virtual` contenedora — aloja comandos hijos que se le unen.
342 pub fn to_intent(
343 &self,
344 id: WorkspaceId,
345 isolation: IsolationLevel,
346 ) -> Result<Intent, CompileError> {
347 Ok(Intent::new(self.to_card(id)?).with_isolation(isolation))
348 }
349
350 /// Puente "ambiente = un shell aislado" (slice 1: *solo aislarlo*).
351 /// A diferencia de [`Self::to_card`] (Card `Virtual` contenedora), esto
352 /// produce una Card **Native** que ejecuta `exec`/`argv` tomando el
353 /// `soma` del workspace como su propio aislamiento — el shell vive
354 /// directamente dentro de los namespaces del ambiente, sin un paso de
355 /// "unirse" a un contenedor aparte. La devuelve como `Intent` lista
356 /// para `Engine::run`.
357 pub fn shell_intent(
358 &self,
359 id: WorkspaceId,
360 exec: impl Into<String>,
361 argv: Vec<String>,
362 isolation: IsolationLevel,
363 ) -> Result<Intent, CompileError> {
364 if self.label.trim().is_empty() {
365 return Err(CompileError::EmptyWorkspaceLabel);
366 }
367 let mut c = Card::new(format!("shuma.workspace.{}", self.label));
368 c.id = id.0;
369 c.soma = self.soma.clone();
370 c.permissions = self.permissions.clone();
371 c.payload = Payload::Native {
372 exec: exec.into(),
373 argv,
374 envp: vec![],
375 };
376 c.supervision = Supervision::OneShot;
377 Ok(Intent::new(c).with_isolation(isolation))
378 }
379}
380
381impl CommandRef {
382 /// Compila un CommandRef a Card hija de un workspace. La Card resultante
383 /// referencia al workspace por label en `pin_to` de cada Flow.
384 pub fn to_card(&self, idx: usize, workspace_label: &str) -> Result<Card, CompileError> {
385 if self.label.trim().is_empty() {
386 return Err(CompileError::EmptyCommandLabel(idx));
387 }
388 let mut c = Card::new(format!("shuma.cmd.{}.{}", workspace_label, self.label));
389 c.payload = self.payload.clone();
390 c.soma = intersect_soma(&self.soma, /*workspace*/ &SomaSpec::default());
391 c.supervision = self.supervision.clone();
392 c.flow = self.flows.clone();
393 // pin_to del workspace en cada Flow input/output → el broker prefiere
394 // resolver dentro del mismo workspace cuando hay candidatos múltiples.
395 let pin = format!("shuma.workspace.{}", workspace_label);
396 for f in c.flow.input.iter_mut().chain(c.flow.output.iter_mut()) {
397 if f.pin_to.is_none() {
398 f.pin_to = Some(pin.clone());
399 }
400 }
401 Ok(c)
402 }
403}
404
405/// Intersección conservadora: si el workspace pidió aislamiento, la hija
406/// también lo tiene (no puede aflojar). Si la hija pidió aislamiento extra,
407/// se respeta.
408fn intersect_soma(child: &SomaSpec, ws: &SomaSpec) -> SomaSpec {
409 let mut out = child.clone();
410 out.namespaces.mount |= ws.namespaces.mount;
411 out.namespaces.pid |= ws.namespaces.pid;
412 out.namespaces.net |= ws.namespaces.net;
413 out.namespaces.uts |= ws.namespaces.uts;
414 out.namespaces.ipc |= ws.namespaces.ipc;
415 out.namespaces.user |= ws.namespaces.user;
416 out.namespaces.cgroup |= ws.namespaces.cgroup;
417 // rlimits: el menor (más restrictivo) gana.
418 out.rlimits.mem_bytes = min_opt(out.rlimits.mem_bytes, ws.rlimits.mem_bytes);
419 out.rlimits.nproc = min_opt(out.rlimits.nproc, ws.rlimits.nproc);
420 out.rlimits.nofile = min_opt(out.rlimits.nofile, ws.rlimits.nofile);
421 out
422}
423
424fn min_opt<T: Ord + Copy>(a: Option<T>, b: Option<T>) -> Option<T> {
425 match (a, b) {
426 (Some(x), Some(y)) => Some(x.min(y)),
427 (Some(x), None) | (None, Some(x)) => Some(x),
428 (None, None) => None,
429 }
430}
431
432impl PipelineSpec {
433 pub fn validate(&self) -> Result<(), CompileError> {
434 let n = self.nodes.len();
435 for (i, c) in self.nodes.iter().enumerate() {
436 if c.label.trim().is_empty() {
437 return Err(CompileError::EmptyCommandLabel(i));
438 }
439 }
440 for e in &self.edges {
441 if e.from >= n || e.to >= n {
442 return Err(CompileError::EdgeOutOfBounds {
443 from: e.from,
444 to: e.to,
445 nodes: n,
446 });
447 }
448 }
449 Ok(())
450 }
451}
452
453// =====================================================================
454// I/O conveniencia (TOML + JSON)
455// =====================================================================
456
457#[derive(Debug, Error)]
458pub enum LoadError {
459 #[error("io: {0}")]
460 Io(#[from] std::io::Error),
461 #[error("toml: {0}")]
462 Toml(#[from] toml::de::Error),
463 #[error("json: {0}")]
464 Json(#[from] serde_json::Error),
465 #[error("format desconocido (esperado .toml o .json)")]
466 UnknownFormat,
467}
468
469pub fn load_workspace_spec(path: &std::path::Path) -> Result<WorkspaceSpec, LoadError> {
470 let raw = std::fs::read_to_string(path)?;
471 match path.extension().and_then(|s| s.to_str()) {
472 Some("toml") => Ok(toml::from_str(&raw)?),
473 Some("json") => Ok(serde_json::from_str(&raw)?),
474 _ => Err(LoadError::UnknownFormat),
475 }
476}
477
478pub fn load_pipeline_spec(path: &std::path::Path) -> Result<PipelineSpec, LoadError> {
479 let raw = std::fs::read_to_string(path)?;
480 match path.extension().and_then(|s| s.to_str()) {
481 Some("toml") => Ok(toml::from_str(&raw)?),
482 Some("json") => Ok(serde_json::from_str(&raw)?),
483 _ => Err(LoadError::UnknownFormat),
484 }
485}
486
487/// Sustituye `${KEY}` en todos los strings del spec por el valor de
488/// `vars["KEY"]`. Variables sin match quedan intactas (no se borra el
489/// placeholder — útil para detectar olvidos).
490///
491/// Walk recursivo sobre la representación JSON intermedia para cubrir
492/// labels, argv, envp, paths y cualquier String del schema.
493pub fn substitute_vars(
494 spec: &PipelineSpec,
495 vars: &std::collections::HashMap<String, String>,
496) -> Result<PipelineSpec, serde_json::Error> {
497 if vars.is_empty() {
498 return Ok(spec.clone());
499 }
500 let mut v = serde_json::to_value(spec)?;
501 walk_subst(&mut v, vars);
502 serde_json::from_value(v)
503}
504
505fn walk_subst(v: &mut serde_json::Value, vars: &std::collections::HashMap<String, String>) {
506 match v {
507 serde_json::Value::String(s) => {
508 *s = subst_str(s, vars);
509 }
510 serde_json::Value::Array(arr) => {
511 for item in arr {
512 walk_subst(item, vars);
513 }
514 }
515 serde_json::Value::Object(obj) => {
516 for (_, val) in obj.iter_mut() {
517 walk_subst(val, vars);
518 }
519 }
520 _ => {}
521 }
522}
523
524fn subst_str(s: &str, vars: &std::collections::HashMap<String, String>) -> String {
525 let mut out = String::with_capacity(s.len());
526 let bytes = s.as_bytes();
527 let mut i = 0;
528 while i < bytes.len() {
529 if i + 1 < bytes.len() && bytes[i] == b'$' && bytes[i + 1] == b'{' {
530 // Buscar el cierre `}`.
531 if let Some(close) = bytes[i + 2..].iter().position(|&b| b == b'}') {
532 let key = std::str::from_utf8(&bytes[i + 2..i + 2 + close]).unwrap_or("");
533 if let Some(val) = vars.get(key) {
534 out.push_str(val);
535 i += 2 + close + 1;
536 continue;
537 }
538 }
539 }
540 out.push(bytes[i] as char);
541 i += 1;
542 }
543 out
544}
545
546#[cfg(test)]
547mod subst_tests {
548 use super::*;
549 use std::collections::HashMap;
550
551 #[test]
552 fn substitute_in_argv_and_label() {
553 let mut vars = HashMap::new();
554 vars.insert("MSG".into(), "hola-mundo".into());
555 vars.insert("LABEL".into(), "renamed".into());
556 let spec = PipelineSpec {
557 label: "p-${LABEL}".into(),
558 workspace: WorkspaceId::new(),
559 nodes: vec![CommandRef {
560 label: "node-${LABEL}".into(),
561 payload: Payload::Native {
562 exec: "/bin/echo".into(),
563 argv: vec!["${MSG}".into()],
564 envp: vec![],
565 },
566 soma: Default::default(),
567 flows: Default::default(),
568 supervision: Supervision::OneShot,
569 }],
570 edges: vec![],
571 discern: DiscernPolicy::default(),
572 restart_on_failure: false,
573 restart_backoff_ms: 200,
574 restart_max_backoff_ms: 30_000,
575 restart_max: 0,
576 };
577 let out = substitute_vars(&spec, &vars).unwrap();
578 assert_eq!(out.label, "p-renamed");
579 assert_eq!(out.nodes[0].label, "node-renamed");
580 match &out.nodes[0].payload {
581 Payload::Native { argv, .. } => assert_eq!(argv[0], "hola-mundo"),
582 _ => panic!("wrong payload"),
583 }
584 }
585
586 #[test]
587 fn unknown_var_left_intact() {
588 let vars = HashMap::new();
589 let spec = PipelineSpec {
590 label: "p-${UNDEFINED}".into(),
591 workspace: WorkspaceId::new(),
592 nodes: vec![],
593 edges: vec![],
594 discern: DiscernPolicy::default(),
595 restart_on_failure: false,
596 restart_backoff_ms: 200,
597 restart_max_backoff_ms: 30_000,
598 restart_max: 0,
599 };
600 let out = substitute_vars(&spec, &vars).unwrap();
601 assert_eq!(out.label, "p-${UNDEFINED}");
602 }
603}
604
605#[cfg(test)]
606mod tests {
607 use super::*;
608
609 fn sample_workspace() -> WorkspaceSpec {
610 WorkspaceSpec {
611 label: "demo".into(),
612 soma: SomaSpec::default(),
613 permissions: Permissions::default(),
614 ttl: Some(Duration::from_secs(60)),
615 flow_dirs: vec![FlowSlot {
616 name: "out".into(),
617 direction: FlowDirection::Output,
618 scope: FlowScope::Public,
619 }],
620 on_exit: ExitPolicy::Reap,
621 quota_enforce: Default::default(),
622 }
623 }
624
625 #[test]
626 fn workspace_toml_roundtrip() {
627 let ws = sample_workspace();
628 let s = toml::to_string(&ws).unwrap();
629 let back: WorkspaceSpec = toml::from_str(&s).unwrap();
630 assert_eq!(back.label, ws.label);
631 assert_eq!(back.ttl, ws.ttl);
632 assert_eq!(back.flow_dirs.len(), 1);
633 }
634
635 #[test]
636 fn workspace_json_roundtrip() {
637 let ws = sample_workspace();
638 let s = serde_json::to_string(&ws).unwrap();
639 let back: WorkspaceSpec = serde_json::from_str(&s).unwrap();
640 assert_eq!(back.label, ws.label);
641 }
642
643 #[test]
644 fn workspace_compiles_to_card() {
645 let ws = sample_workspace();
646 let id = WorkspaceId::new();
647 let c = ws.to_card(id).unwrap();
648 assert_eq!(c.id, id.0);
649 assert!(c.label.starts_with("shuma.workspace."));
650 assert!(matches!(c.payload, Payload::Virtual));
651 }
652
653 #[test]
654 fn empty_label_rejected() {
655 let mut ws = sample_workspace();
656 ws.label = String::new();
657 assert!(ws.to_card(WorkspaceId::new()).is_err());
658 }
659
660 #[test]
661 fn workspace_compiles_to_intent_virtual() {
662 let ws = sample_workspace();
663 let intent = ws
664 .to_intent(WorkspaceId::new(), IsolationLevel::Standard)
665 .unwrap();
666 assert!(matches!(intent.card.payload, Payload::Virtual));
667 assert_eq!(intent.context.isolation, Some(IsolationLevel::Standard));
668 }
669
670 #[test]
671 fn shell_intent_is_native_and_inherits_workspace_soma() {
672 let mut ws = sample_workspace();
673 ws.soma.namespaces.user = true;
674 ws.soma.namespaces.pid = true;
675 let intent = ws
676 .shell_intent(
677 WorkspaceId::new(),
678 "/bin/sh",
679 vec!["-c".into(), "true".into()],
680 IsolationLevel::Standard,
681 )
682 .unwrap();
683 match &intent.card.payload {
684 Payload::Native { exec, .. } => assert_eq!(exec, "/bin/sh"),
685 other => panic!("esperaba Native, fue {other:?}"),
686 }
687 assert!(
688 intent.card.soma.namespaces.pid,
689 "el shell del ambiente hereda el soma del workspace"
690 );
691 }
692
693 /// Unificación del orquestador: un `WorkspaceSpec` de shuma se materializa
694 /// como un proceso aislado real **a través del `Engine` de sandokan**
695 /// (LocalEngine), sin que shuma maneje su propio ciclo de vida. El shell
696 /// sale 0 sólo si vio PID 1, probando que el aislamiento del ambiente se
697 /// aplicó end-to-end por el camino unificado.
698 #[tokio::test]
699 async fn workspace_shell_runs_isolated_via_sandokan_engine() {
700 use card_core::NamespaceSet;
701 use sandokan_core::Engine;
702 use sandokan_local::LocalEngine;
703
704 let mut ws = sample_workspace();
705 ws.soma.namespaces = NamespaceSet {
706 user: true,
707 pid: true,
708 mount: true,
709 uts: true,
710 ipc: true,
711 net: false,
712 cgroup: false,
713 };
714
715 let intent = ws
716 .shell_intent(
717 WorkspaceId::new(),
718 "/bin/sh",
719 vec!["-c".into(), "test $$ -eq 1".into()],
720 IsolationLevel::Standard,
721 )
722 .unwrap();
723 let id = intent.card_id();
724
725 let engine = LocalEngine::new();
726 engine.run(intent).await.expect("run vía sandokan");
727 tokio::time::sleep(Duration::from_millis(300)).await;
728 let st = engine.status(id).await.expect("status");
729 assert!(st.is_terminal(), "el shell debió terminar, fue {st:?}");
730 assert!(
731 !st.is_failure(),
732 "exit 0 ⇒ el shell vio PID 1 (ambiente aislado). Estado: {st:?}"
733 );
734 }
735
736 #[test]
737 fn pipeline_validates_edges() {
738 let p = PipelineSpec {
739 label: "p".into(),
740 workspace: WorkspaceId::new(),
741 nodes: vec![CommandRef {
742 label: "a".into(),
743 payload: Payload::Virtual,
744 soma: SomaSpec::default(),
745 flows: card_core::Flows::default(),
746 supervision: Supervision::OneShot,
747 }],
748 edges: vec![FlowEdge {
749 from: 0,
750 from_output: "x".into(),
751 to: 5,
752 to_input: "y".into(),
753 }],
754 discern: DiscernPolicy::default(),
755 restart_on_failure: false,
756 restart_backoff_ms: 200,
757 restart_max_backoff_ms: 30_000,
758 restart_max: 0,
759 };
760 assert!(p.validate().is_err());
761 }
762
763 #[test]
764 fn intersect_soma_takes_more_restrictive() {
765 let mut child = SomaSpec::default();
766 child.rlimits.mem_bytes = Some(1_000_000);
767 let mut ws = SomaSpec::default();
768 ws.rlimits.mem_bytes = Some(500_000);
769 ws.namespaces.user = true;
770 let r = intersect_soma(&child, &ws);
771 assert_eq!(r.rlimits.mem_bytes, Some(500_000));
772 assert!(r.namespaces.user);
773 }
774}