⎇ d8727a3038c45b4fc69a37fc988e38408ddb5f6c
crates/apps/shipote-cli/src/main.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat(shipote): CPU% + pipeline live-tail + replay por bytes (fase J)
1//! `shipote` — CLI de administración del daemon.
2
3use anyhow::{anyhow, Context, Result};
4use clap::{Parser, Subcommand};
5use shipote_card::{load_pipeline_spec, load_workspace_spec, WorkspaceId};
6use shipote_protocol::{default_socket_path, read_frame, write_frame, Request, Response};
7use std::path::PathBuf;
8use tokio::net::UnixStream;
9use ulid::Ulid;
10
11#[derive(Parser, Debug)]
12#[command(name = "shipote", version, about = "Administración de shipote-daemon")]
13struct Cli {
14 /// Path al socket del daemon. Default: $XDG_RUNTIME_DIR/shipote.sock.
15 #[arg(long, global = true)]
16 socket: Option<PathBuf>,
17
18 #[command(subcommand)]
19 cmd: Cmd,
20}
21
22#[derive(Subcommand, Debug)]
23enum Cmd {
24 /// Health-check del daemon.
25 Ping,
26
27 /// Capacidades runtime detectadas por el daemon.
28 Caps,
29
30 /// Operaciones sobre Workspaces.
31 #[command(subcommand)]
32 Workspace(WsCmd),
33
34 /// Ejecutar un comando one-shot dentro de un workspace.
35 Run {
36 /// ULID del workspace destino.
37 #[arg(short = 'w', long)]
38 workspace: String,
39 /// Path del ejecutable.
40 exec: String,
41 /// Argumentos del comando.
42 argv: Vec<String>,
43 },
44
45 /// Discernir el tipo de un archivo (ad-hoc, sin workspace).
46 Discern {
47 /// Path al archivo a discernir.
48 path: PathBuf,
49 },
50
51 /// Listar comandos de un workspace.
52 Commands {
53 /// ULID del workspace.
54 workspace: String,
55 },
56
57 /// Mostrar tail del log capturado de un comando.
58 Logs {
59 /// ULID del workspace.
60 workspace: String,
61 /// ULID del comando.
62 command: String,
63 /// Bytes desde el final (0 = todo).
64 #[arg(long, default_value_t = 0)]
65 tail: usize,
66 /// Stream a leer: stdout | stderr | both.
67 #[arg(long, default_value = "both")]
68 stream: String,
69 },
70
71 /// Pipeline DAG con flujo tipado.
72 #[command(subcommand)]
73 Pipeline(PipeCmd),
74
75 /// Flow data plane (subscribirse a streams enriquecidos).
76 #[command(subcommand)]
77 Flow(FlowCmd),
78}
79
80#[derive(Subcommand, Debug)]
81enum FlowCmd {
82 /// Listar pipelines activos con sus sockets de flow.
83 List,
84 /// Cerrar el data plane de un pipeline (drop de todos sus sockets).
85 Drop { pipeline: String },
86 /// Suscribirse a un flow socket y volcar bytes a stdout.
87 Tail {
88 /// Path al Unix socket del flow.
89 socket: PathBuf,
90 },
91}
92
93#[derive(Subcommand, Debug)]
94enum PipeCmd {
95 /// Lanzar un Pipeline desde un spec TOML/JSON.
96 Run {
97 /// Path al spec del pipeline.
98 spec: PathBuf,
99 /// Interponer un tap entre productor↔consumidor de cada edge para
100 /// discernir el TypeRef del flujo.
101 #[arg(long)]
102 tap: bool,
103 /// Variables `KEY=VALUE` para sustitución `${KEY}` en el spec.
104 #[arg(long = "var", value_parser = parse_kv)]
105 vars: Vec<(String, String)>,
106 /// Tras lanzar, suscribir al primer flow socket y volcar bytes
107 /// a stdout hasta EOF. Implica `--tap`.
108 #[arg(long)]
109 tail: bool,
110 },
111 /// Guardar un pipeline bajo un nombre (persiste con el snapshot).
112 Save {
113 /// Nombre simbólico.
114 name: String,
115 /// Path al spec.
116 spec: PathBuf,
117 },
118 /// Listar nombres de pipelines guardados.
119 SavedList,
120 /// Eliminar un pipeline guardado (no afecta runs en curso).
121 Drop { name: String },
122 /// Detener un pipeline en curso por ID (SIGTERM → grace → SIGKILL
123 /// sólo a sus comandos).
124 Stop {
125 /// ULID del pipeline (devuelto por `pipeline run`).
126 pipeline: String,
127 #[arg(long, default_value_t = 1000)]
128 grace_ms: u64,
129 },
130 /// Ejecutar un pipeline guardado por nombre.
131 RunSaved {
132 name: String,
133 #[arg(long)]
134 tap: bool,
135 #[arg(long = "var", value_parser = parse_kv)]
136 vars: Vec<(String, String)>,
137 #[arg(long)]
138 tail: bool,
139 },
140}
141
142fn parse_kv(s: &str) -> Result<(String, String), String> {
143 let (k, v) = s.split_once('=').ok_or_else(|| format!("expected KEY=VALUE, got `{s}`"))?;
144 Ok((k.to_string(), v.to_string()))
145}
146
147#[derive(Subcommand, Debug)]
148enum WsCmd {
149 /// Crear un workspace desde un spec TOML/JSON.
150 Create {
151 /// Path al spec del workspace.
152 spec: PathBuf,
153 },
154 /// Listar workspaces vivos.
155 List,
156 /// Detener un workspace por ID.
157 Stop {
158 id: String,
159 /// Milisegundos de gracia tras SIGTERM antes de SIGKILL.
160 #[arg(long, default_value_t = 1000)]
161 grace_ms: u64,
162 },
163 /// Resource accounting (RSS, CPU, comandos vivos).
164 Stats {
165 id: String,
166 },
167}
168
169#[tokio::main]
170async fn main() -> Result<()> {
171 let cli = Cli::parse();
172 let socket = cli.socket.unwrap_or_else(default_socket_path);
173 let mut stream = UnixStream::connect(&socket)
174 .await
175 .with_context(|| format!("connect {}", socket.display()))?;
176
177 match cli.cmd {
178 Cmd::Ping => {
179 let resp = round_trip(&mut stream, Request::Ping).await?;
180 match resp {
181 Response::Pong => println!("pong"),
182 other => print_unexpected(&other),
183 }
184 }
185
186 Cmd::Caps => {
187 let resp = round_trip(&mut stream, Request::Capabilities).await?;
188 match resp {
189 Response::Capabilities {
190 kernel_version,
191 user_ns,
192 cgroup_v2,
193 cgroup_delegated,
194 has_cap_sys_admin,
195 } => {
196 println!("kernel: {}.{}.{}", kernel_version.0, kernel_version.1, kernel_version.2);
197 println!("user_ns: {user_ns}");
198 println!("cgroup_v2: {cgroup_v2}");
199 println!("cgroup_delegated: {cgroup_delegated}");
200 println!("cap_sys_admin: {has_cap_sys_admin}");
201 }
202 other => print_unexpected(&other),
203 }
204 }
205
206 Cmd::Workspace(WsCmd::Create { spec }) => {
207 let ws = load_workspace_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
208 let resp = round_trip(&mut stream, Request::WorkspaceCreate { spec: ws }).await?;
209 match resp {
210 Response::WorkspaceCreated { id, warnings } => {
211 println!("{id}");
212 for w in warnings {
213 eprintln!("warning: {w}");
214 }
215 }
216 Response::Error { message } => return Err(anyhow!(message)),
217 other => print_unexpected(&other),
218 }
219 }
220
221 Cmd::Workspace(WsCmd::List) => {
222 let resp = round_trip(&mut stream, Request::WorkspaceList).await?;
223 match resp {
224 Response::WorkspaceList { items } => {
225 if items.is_empty() {
226 println!("(no workspaces)");
227 }
228 for it in items {
229 println!(
230 "{} {:<20} cmds={} uptime={}ms",
231 it.id, it.label, it.commands, it.uptime_ms
232 );
233 }
234 }
235 other => print_unexpected(&other),
236 }
237 }
238
239 Cmd::Workspace(WsCmd::Stats { id }) => {
240 let id = parse_ws_id(&id)?;
241 let resp = round_trip(&mut stream, Request::WorkspaceStats { workspace: id }).await?;
242 match resp {
243 Response::WorkspaceStats { info } => {
244 println!("commands: {} alive / {} total", info.commands_alive, info.commands_total);
245 let fmt_mib = |b: u64| format!("{:.2} MiB", b as f64 / 1024.0 / 1024.0);
246 let rss = info.rss_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
247 let peak = info.rss_peak_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
248 let cpu = info
249 .cpu_usec
250 .map(|u| format!("{:.3} s", u as f64 / 1_000_000.0))
251 .unwrap_or_else(|| "—".into());
252 let cpu_pct = info
253 .cpu_percent
254 .map(|p| format!("{p:.1} %"))
255 .unwrap_or_else(|| "— (esperando 2do sample)".into());
256 println!("rss: {rss}");
257 println!("rss_peak: {peak}");
258 println!("cpu: {cpu}");
259 println!("cpu_pct: {cpu_pct}");
260 println!("source: {}", info.source);
261 println!("uptime: {} ms", info.uptime_ms);
262 }
263 Response::Error { message } => return Err(anyhow!(message)),
264 other => print_unexpected(&other),
265 }
266 }
267
268 Cmd::Workspace(WsCmd::Stop { id, grace_ms }) => {
269 let id = parse_ws_id(&id)?;
270 let resp = round_trip(&mut stream, Request::WorkspaceStop { id, grace_ms }).await?;
271 match resp {
272 Response::WorkspaceStopped { id, reaped } => {
273 println!("stopped {id} (reaped {reaped})");
274 }
275 Response::Error { message } => return Err(anyhow!(message)),
276 other => print_unexpected(&other),
277 }
278 }
279
280 Cmd::Run { workspace, exec, argv } => {
281 let id = parse_ws_id(&workspace)?;
282 let resp = round_trip(
283 &mut stream,
284 Request::Run {
285 workspace: id,
286 exec,
287 argv,
288 envp: vec![],
289 },
290 )
291 .await?;
292 match resp {
293 Response::RunStarted { command_id, pid, .. } => {
294 println!("{command_id} pid={pid}");
295 }
296 Response::Error { message } => return Err(anyhow!(message)),
297 other => print_unexpected(&other),
298 }
299 }
300
301 Cmd::Pipeline(PipeCmd::Run { spec, tap, vars, tail }) => {
302 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
303 // --tail implica --tap (no hay flow socket sin tap).
304 let effective_tap = tap || tail;
305 let resp = round_trip(
306 &mut stream,
307 Request::PipelineRun {
308 spec: p,
309 tap: effective_tap,
310 vars: vars.into_iter().collect(),
311 },
312 )
313 .await?;
314 let socket = print_pipeline_started_returning_socket(resp)?;
315 if tail {
316 if let Some(sock) = socket {
317 eprintln!("--- tailing {} ---", sock.display());
318 tail_socket(&sock).await?;
319 } else {
320 eprintln!("--tail: no hay flow socket disponible");
321 }
322 }
323 }
324
325 Cmd::Pipeline(PipeCmd::Save { name, spec }) => {
326 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
327 let resp = round_trip(&mut stream, Request::PipelineSave { name: name.clone(), spec: p }).await?;
328 match resp {
329 Response::PipelineSaved { name } => println!("saved {name}"),
330 Response::Error { message } => return Err(anyhow!(message)),
331 other => print_unexpected(&other),
332 }
333 }
334
335 Cmd::Pipeline(PipeCmd::SavedList) => {
336 let resp = round_trip(&mut stream, Request::PipelineSavedList).await?;
337 match resp {
338 Response::PipelineSavedList { names } => {
339 if names.is_empty() {
340 println!("(no saved pipelines)");
341 }
342 for n in names {
343 println!("{n}");
344 }
345 }
346 other => print_unexpected(&other),
347 }
348 }
349
350 Cmd::Pipeline(PipeCmd::Stop { pipeline, grace_ms }) => {
351 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
352 let resp = round_trip(&mut stream, Request::PipelineStop { pipeline: pid, grace_ms }).await?;
353 match resp {
354 Response::PipelineStopped { pipeline, reaped } => {
355 println!("stopped pipeline {pipeline} (reaped {reaped})");
356 }
357 other => print_unexpected(&other),
358 }
359 }
360
361 Cmd::Pipeline(PipeCmd::Drop { name }) => {
362 let resp = round_trip(&mut stream, Request::PipelineDrop { name }).await?;
363 match resp {
364 Response::PipelineDropped { name, existed } => {
365 if existed {
366 println!("dropped {name}");
367 } else {
368 eprintln!("no existía: {name}");
369 }
370 }
371 other => print_unexpected(&other),
372 }
373 }
374
375 Cmd::Pipeline(PipeCmd::RunSaved { name, tap, vars, tail }) => {
376 let effective_tap = tap || tail;
377 let resp = round_trip(
378 &mut stream,
379 Request::PipelineRunSaved {
380 name,
381 tap: effective_tap,
382 vars: vars.into_iter().collect(),
383 },
384 )
385 .await?;
386 let socket = print_pipeline_started_returning_socket(resp)?;
387 if tail {
388 if let Some(sock) = socket {
389 eprintln!("--- tailing {} ---", sock.display());
390 tail_socket(&sock).await?;
391 } else {
392 eprintln!("--tail: no hay flow socket disponible");
393 }
394 }
395 }
396
397 Cmd::Commands { workspace } => {
398 let ws = parse_ws_id(&workspace)?;
399 let resp = round_trip(&mut stream, Request::CommandList { workspace: ws }).await?;
400 match resp {
401 Response::CommandList { items } => {
402 if items.is_empty() {
403 println!("(no commands)");
404 }
405 for c in items {
406 let alive = if c.alive { "alive" } else { "exited" };
407 let exit = c
408 .exit_status
409 .map(|s| format!("exit={s}"))
410 .unwrap_or_default();
411 println!(
412 "{} {:<24} pid={:<7} {:<8} logs={} {}",
413 c.id, c.label, c.pid, alive, c.log_bytes, exit
414 );
415 }
416 }
417 other => print_unexpected(&other),
418 }
419 }
420
421 Cmd::Logs { workspace, command, tail, stream: which_stream } => {
422 let ws = parse_ws_id(&workspace)?;
423 let cmd_id = Ulid::from_string(&command).map_err(|e| anyhow!("invalid command id: {e}"))?;
424 let resp = round_trip(
425 &mut stream,
426 Request::CommandLogs {
427 workspace: ws,
428 command: cmd_id,
429 tail_bytes: tail,
430 stream: which_stream,
431 },
432 )
433 .await?;
434 match resp {
435 Response::CommandLogs { bytes } => {
436 // stdout raw, sin decoding — el log puede tener bytes binarios.
437 use std::io::Write;
438 let _ = std::io::stdout().write_all(&bytes);
439 let _ = std::io::stdout().flush();
440 }
441 Response::Error { message } => return Err(anyhow!(message)),
442 other => print_unexpected(&other),
443 }
444 }
445
446 Cmd::Flow(FlowCmd::List) => {
447 let resp = round_trip(&mut stream, Request::FlowList).await?;
448 match resp {
449 Response::FlowList { items } => {
450 if items.is_empty() {
451 println!("(no active flows)");
452 }
453 for it in items {
454 println!("{}", it.pipeline);
455 for s in it.sockets {
456 println!(" {}", s.display());
457 }
458 }
459 }
460 other => print_unexpected(&other),
461 }
462 }
463
464 Cmd::Flow(FlowCmd::Drop { pipeline }) => {
465 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
466 let resp = round_trip(&mut stream, Request::FlowDrop { pipeline: pid }).await?;
467 match resp {
468 Response::FlowDropped { pipeline, existed } => {
469 if existed {
470 println!("dropped {pipeline}");
471 } else {
472 eprintln!("no existía: {pipeline}");
473 }
474 }
475 other => print_unexpected(&other),
476 }
477 }
478
479 Cmd::Flow(FlowCmd::Tail { socket }) => {
480 // Subscribirse directo al socket — no pasamos por el daemon.
481 use tokio::io::AsyncReadExt;
482 let mut s = UnixStream::connect(&socket)
483 .await
484 .with_context(|| format!("connect {}", socket.display()))?;
485 let mut buf = [0u8; 4096];
486 loop {
487 let n = s.read(&mut buf).await?;
488 if n == 0 {
489 break;
490 }
491 use std::io::Write;
492 let _ = std::io::stdout().write_all(&buf[..n]);
493 let _ = std::io::stdout().flush();
494 }
495 }
496
497 Cmd::Discern { path } => {
498 let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
499 // Sample: hasta 4 KiB.
500 let sample = bytes.into_iter().take(4096).collect();
501 let resp = round_trip(
502 &mut stream,
503 Request::Discern {
504 sample,
505 hint_path: Some(path),
506 },
507 )
508 .await?;
509 match resp {
510 Response::Discernment { ty, confidence, mime, lens } => {
511 println!("type: {ty}");
512 println!("confidence: {confidence:.2}");
513 if let Some(m) = mime {
514 println!("mime: {m}");
515 }
516 if let Some(l) = lens {
517 println!("lens: {l}");
518 }
519 }
520 Response::Error { message } => return Err(anyhow!(message)),
521 other => print_unexpected(&other),
522 }
523 }
524 }
525
526 Ok(())
527}
528
529async fn round_trip(stream: &mut UnixStream, req: Request) -> Result<Response> {
530 write_frame(stream, &req).await?;
531 let resp: Response = read_frame(stream).await?;
532 Ok(resp)
533}
534
535fn parse_ws_id(s: &str) -> Result<WorkspaceId> {
536 let u = Ulid::from_string(s).map_err(|e| anyhow!("invalid workspace id: {e}"))?;
537 Ok(WorkspaceId(u))
538}
539
540fn print_unexpected(r: &Response) {
541 eprintln!("unexpected response: {r:?}");
542}
543
544/// Imprime el resultado del launch del pipeline y retorna el path del
545/// primer flow socket (si hay), útil para `--tail`.
546fn print_pipeline_started_returning_socket(resp: Response) -> Result<Option<PathBuf>> {
547 match resp {
548 Response::PipelineStarted { pipeline, command_pids, edges } => {
549 println!("pipeline {pipeline}");
550 for (label, pid) in command_pids {
551 println!(" {:<20} pid={pid}", label);
552 }
553 let mut first_socket: Option<PathBuf> = None;
554 if !edges.is_empty() {
555 println!("edges:");
556 for e in &edges {
557 println!(
558 " {}.{} → {}.{} ty={:?} mime={:?} conf={:.2}",
559 e.from_label, e.from_output, e.to_label, e.to_input,
560 e.ty, e.mime, e.confidence,
561 );
562 if first_socket.is_none() {
563 first_socket = e.flow_socket.clone();
564 }
565 }
566 }
567 Ok(first_socket)
568 }
569 Response::Error { message } => Err(anyhow!(message)),
570 other => {
571 print_unexpected(&other);
572 Ok(None)
573 }
574 }
575}
576
577async fn tail_socket(socket: &std::path::Path) -> Result<()> {
578 use tokio::io::AsyncReadExt;
579 // Pequeña ventana de retry — el daemon retiene el flow channel
580 // antes de retornar, así que en la práctica ya está bindeado.
581 let mut s = UnixStream::connect(socket)
582 .await
583 .with_context(|| format!("connect {}", socket.display()))?;
584 let mut buf = [0u8; 4096];
585 loop {
586 let n = s.read(&mut buf).await?;
587 if n == 0 {
588 break;
589 }
590 use std::io::Write;
591 let _ = std::io::stdout().write_all(&buf[..n]);
592 let _ = std::io::stdout().flush();
593 }
594 Ok(())
595}