⎇ c22d2480b93b7802be8e8ea5a0075b8175c4fedc
crates/apps/shipote-cli/src/main.rs
History
Who touched what
raw
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 },
67
68 /// Pipeline DAG con flujo tipado.
69 #[command(subcommand)]
70 Pipeline(PipeCmd),
71}
72
73#[derive(Subcommand, Debug)]
74enum PipeCmd {
75 /// Lanzar un Pipeline desde un spec TOML/JSON.
76 Run {
77 /// Path al spec del pipeline.
78 spec: PathBuf,
79 /// Interponer un tap entre productor↔consumidor de cada edge para
80 /// discernir el TypeRef del flujo.
81 #[arg(long)]
82 tap: bool,
83 },
84 /// Guardar un pipeline bajo un nombre (persiste con el snapshot).
85 Save {
86 /// Nombre simbólico.
87 name: String,
88 /// Path al spec.
89 spec: PathBuf,
90 },
91 /// Listar nombres de pipelines guardados.
92 SavedList,
93 /// Eliminar un pipeline guardado.
94 Drop { name: String },
95 /// Ejecutar un pipeline guardado por nombre.
96 RunSaved {
97 name: String,
98 #[arg(long)]
99 tap: bool,
100 },
101}
102
103#[derive(Subcommand, Debug)]
104enum WsCmd {
105 /// Crear un workspace desde un spec TOML/JSON.
106 Create {
107 /// Path al spec del workspace.
108 spec: PathBuf,
109 },
110 /// Listar workspaces vivos.
111 List,
112 /// Detener un workspace por ID.
113 Stop {
114 id: String,
115 },
116}
117
118#[tokio::main]
119async fn main() -> Result<()> {
120 let cli = Cli::parse();
121 let socket = cli.socket.unwrap_or_else(default_socket_path);
122 let mut stream = UnixStream::connect(&socket)
123 .await
124 .with_context(|| format!("connect {}", socket.display()))?;
125
126 match cli.cmd {
127 Cmd::Ping => {
128 let resp = round_trip(&mut stream, Request::Ping).await?;
129 match resp {
130 Response::Pong => println!("pong"),
131 other => print_unexpected(&other),
132 }
133 }
134
135 Cmd::Caps => {
136 let resp = round_trip(&mut stream, Request::Capabilities).await?;
137 match resp {
138 Response::Capabilities {
139 kernel_version,
140 user_ns,
141 cgroup_v2,
142 cgroup_delegated,
143 has_cap_sys_admin,
144 } => {
145 println!("kernel: {}.{}.{}", kernel_version.0, kernel_version.1, kernel_version.2);
146 println!("user_ns: {user_ns}");
147 println!("cgroup_v2: {cgroup_v2}");
148 println!("cgroup_delegated: {cgroup_delegated}");
149 println!("cap_sys_admin: {has_cap_sys_admin}");
150 }
151 other => print_unexpected(&other),
152 }
153 }
154
155 Cmd::Workspace(WsCmd::Create { spec }) => {
156 let ws = load_workspace_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
157 let resp = round_trip(&mut stream, Request::WorkspaceCreate { spec: ws }).await?;
158 match resp {
159 Response::WorkspaceCreated { id, warnings } => {
160 println!("{id}");
161 for w in warnings {
162 eprintln!("warning: {w}");
163 }
164 }
165 Response::Error { message } => return Err(anyhow!(message)),
166 other => print_unexpected(&other),
167 }
168 }
169
170 Cmd::Workspace(WsCmd::List) => {
171 let resp = round_trip(&mut stream, Request::WorkspaceList).await?;
172 match resp {
173 Response::WorkspaceList { items } => {
174 if items.is_empty() {
175 println!("(no workspaces)");
176 }
177 for it in items {
178 println!(
179 "{} {:<20} cmds={} uptime={}ms",
180 it.id, it.label, it.commands, it.uptime_ms
181 );
182 }
183 }
184 other => print_unexpected(&other),
185 }
186 }
187
188 Cmd::Workspace(WsCmd::Stop { id }) => {
189 let id = parse_ws_id(&id)?;
190 let resp = round_trip(&mut stream, Request::WorkspaceStop { id }).await?;
191 match resp {
192 Response::WorkspaceStopped { id, reaped } => {
193 println!("stopped {id} (reaped {reaped})");
194 }
195 Response::Error { message } => return Err(anyhow!(message)),
196 other => print_unexpected(&other),
197 }
198 }
199
200 Cmd::Run { workspace, exec, argv } => {
201 let id = parse_ws_id(&workspace)?;
202 let resp = round_trip(
203 &mut stream,
204 Request::Run {
205 workspace: id,
206 exec,
207 argv,
208 envp: vec![],
209 },
210 )
211 .await?;
212 match resp {
213 Response::RunStarted { command_id, pid, .. } => {
214 println!("{command_id} pid={pid}");
215 }
216 Response::Error { message } => return Err(anyhow!(message)),
217 other => print_unexpected(&other),
218 }
219 }
220
221 Cmd::Pipeline(PipeCmd::Run { spec, tap }) => {
222 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
223 let resp = round_trip(&mut stream, Request::PipelineRun { spec: p, tap }).await?;
224 print_pipeline_started(resp)?;
225 }
226
227 Cmd::Pipeline(PipeCmd::Save { name, spec }) => {
228 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
229 let resp = round_trip(&mut stream, Request::PipelineSave { name: name.clone(), spec: p }).await?;
230 match resp {
231 Response::PipelineSaved { name } => println!("saved {name}"),
232 Response::Error { message } => return Err(anyhow!(message)),
233 other => print_unexpected(&other),
234 }
235 }
236
237 Cmd::Pipeline(PipeCmd::SavedList) => {
238 let resp = round_trip(&mut stream, Request::PipelineSavedList).await?;
239 match resp {
240 Response::PipelineSavedList { names } => {
241 if names.is_empty() {
242 println!("(no saved pipelines)");
243 }
244 for n in names {
245 println!("{n}");
246 }
247 }
248 other => print_unexpected(&other),
249 }
250 }
251
252 Cmd::Pipeline(PipeCmd::Drop { name }) => {
253 let resp = round_trip(&mut stream, Request::PipelineDrop { name }).await?;
254 match resp {
255 Response::PipelineDropped { name, existed } => {
256 if existed {
257 println!("dropped {name}");
258 } else {
259 eprintln!("no existía: {name}");
260 }
261 }
262 other => print_unexpected(&other),
263 }
264 }
265
266 Cmd::Pipeline(PipeCmd::RunSaved { name, tap }) => {
267 let resp = round_trip(&mut stream, Request::PipelineRunSaved { name, tap }).await?;
268 print_pipeline_started(resp)?;
269 }
270
271 Cmd::Commands { workspace } => {
272 let ws = parse_ws_id(&workspace)?;
273 let resp = round_trip(&mut stream, Request::CommandList { workspace: ws }).await?;
274 match resp {
275 Response::CommandList { items } => {
276 if items.is_empty() {
277 println!("(no commands)");
278 }
279 for c in items {
280 let alive = if c.alive { "alive" } else { "exited" };
281 let exit = c
282 .exit_status
283 .map(|s| format!("exit={s}"))
284 .unwrap_or_default();
285 println!(
286 "{} {:<24} pid={:<7} {:<8} logs={} {}",
287 c.id, c.label, c.pid, alive, c.log_bytes, exit
288 );
289 }
290 }
291 other => print_unexpected(&other),
292 }
293 }
294
295 Cmd::Logs { workspace, command, tail } => {
296 let ws = parse_ws_id(&workspace)?;
297 let cmd_id = Ulid::from_string(&command).map_err(|e| anyhow!("invalid command id: {e}"))?;
298 let resp = round_trip(
299 &mut stream,
300 Request::CommandLogs {
301 workspace: ws,
302 command: cmd_id,
303 tail_bytes: tail,
304 },
305 )
306 .await?;
307 match resp {
308 Response::CommandLogs { bytes } => {
309 // stdout raw, sin decoding — el log puede tener bytes binarios.
310 use std::io::Write;
311 let _ = std::io::stdout().write_all(&bytes);
312 let _ = std::io::stdout().flush();
313 }
314 Response::Error { message } => return Err(anyhow!(message)),
315 other => print_unexpected(&other),
316 }
317 }
318
319 Cmd::Discern { path } => {
320 let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
321 // Sample: hasta 4 KiB.
322 let sample = bytes.into_iter().take(4096).collect();
323 let resp = round_trip(
324 &mut stream,
325 Request::Discern {
326 sample,
327 hint_path: Some(path),
328 },
329 )
330 .await?;
331 match resp {
332 Response::Discernment { ty, confidence, mime, lens } => {
333 println!("type: {ty}");
334 println!("confidence: {confidence:.2}");
335 if let Some(m) = mime {
336 println!("mime: {m}");
337 }
338 if let Some(l) = lens {
339 println!("lens: {l}");
340 }
341 }
342 Response::Error { message } => return Err(anyhow!(message)),
343 other => print_unexpected(&other),
344 }
345 }
346 }
347
348 Ok(())
349}
350
351async fn round_trip(stream: &mut UnixStream, req: Request) -> Result<Response> {
352 write_frame(stream, &req).await?;
353 let resp: Response = read_frame(stream).await?;
354 Ok(resp)
355}
356
357fn parse_ws_id(s: &str) -> Result<WorkspaceId> {
358 let u = Ulid::from_string(s).map_err(|e| anyhow!("invalid workspace id: {e}"))?;
359 Ok(WorkspaceId(u))
360}
361
362fn print_unexpected(r: &Response) {
363 eprintln!("unexpected response: {r:?}");
364}
365
366fn print_pipeline_started(resp: Response) -> Result<()> {
367 match resp {
368 Response::PipelineStarted { pipeline, command_pids, edges } => {
369 println!("pipeline {pipeline}");
370 for (label, pid) in command_pids {
371 println!(" {:<20} pid={pid}", label);
372 }
373 if !edges.is_empty() {
374 println!("edges:");
375 for e in edges {
376 println!(
377 " {}.{} → {}.{} ty={:?} mime={:?} conf={:.2}",
378 e.from_label, e.from_output, e.to_label, e.to_input,
379 e.ty, e.mime, e.confidence,
380 );
381 }
382 }
383 Ok(())
384 }
385 Response::Error { message } => Err(anyhow!(message)),
386 other => {
387 print_unexpected(&other);
388 Ok(())
389 }
390 }
391}