tawasuyu/shuma
Interactive shell with zsh/fish parity, native multiplexing + remote sessions, on Llimphi.
⎇ 1e5d3ed2397d6efbb71aa34380f170471508cfd7
02_ruway/shuma/shuma-cli/src/main.rs
Historia
Quién tocó qué
crudo
feat: shuma standalone — shell interactiva sobre Llimphi (front-door, git-dep al monorepo)
1//! `shuma` — CLI de administración del daemon.
2
3use anyhow::{anyhow, Context, Result};
4use clap::{Parser, Subcommand};
5use shuma_card::{load_pipeline_spec, load_workspace_spec, WorkspaceId};
6use shuma_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 = "shuma", version, about = "Administración de shuma-daemon")]
13struct Cli {
14 /// Path al socket del daemon. Default: $XDG_RUNTIME_DIR/shuma.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 /// Health endpoint estructurado.
28 Health,
29
30 /// Capacidades runtime detectadas por el daemon.
31 Caps,
32
33 /// Operaciones sobre Workspaces.
34 #[command(subcommand)]
35 Workspace(WsCmd),
36
37 /// Ejecutar un comando one-shot dentro de un workspace.
38 Run {
39 /// ULID del workspace destino.
40 #[arg(short = 'w', long)]
41 workspace: String,
42 /// Si exit != 0, relanzar con backoff exponencial.
43 #[arg(long)]
44 restart_on_failure: bool,
45 /// Path del ejecutable.
46 exec: String,
47 /// Argumentos del comando.
48 argv: Vec<String>,
49 },
50
51 /// Discernir el tipo de un archivo (ad-hoc, sin workspace).
52 Discern {
53 /// Path al archivo a discernir.
54 path: PathBuf,
55 },
56
57 /// Listar comandos de un workspace.
58 Commands {
59 /// ULID del workspace.
60 workspace: String,
61 },
62
63 /// Mostrar tail del log capturado de un comando.
64 Logs {
65 /// ULID del workspace.
66 workspace: String,
67 /// ULID del comando.
68 command: String,
69 /// Bytes desde el final (0 = todo).
70 #[arg(long, default_value_t = 0)]
71 tail: usize,
72 /// Stream a leer: stdout | stderr | both.
73 #[arg(long, default_value = "both")]
74 stream: String,
75 /// Seguir el log en vivo (poll cada 200ms hasta que el comando termine).
76 #[arg(short = 'f', long)]
77 follow: bool,
78 },
79
80 /// Pipeline DAG con flujo tipado.
81 #[command(subcommand)]
82 Pipeline(PipeCmd),
83
84 /// Flow data plane (subscribirse a streams enriquecidos).
85 #[command(subcommand)]
86 Flow(FlowCmd),
87}
88
89#[derive(Subcommand, Debug)]
90enum FlowCmd {
91 /// Listar pipelines activos con sus sockets de flow.
92 List,
93 /// Throughput por flow socket (bytes_total + bytes/s).
94 Throughput,
95 /// Cerrar el data plane de un pipeline (drop de todos sus sockets).
96 Drop { pipeline: String },
97 /// Suscribirse a un flow socket y volcar bytes a stdout.
98 Tail {
99 /// Path al Unix socket del flow.
100 socket: PathBuf,
101 },
102}
103
104#[derive(Subcommand, Debug)]
105enum PipeCmd {
106 /// Lanzar un Pipeline desde un spec TOML/JSON.
107 Run {
108 /// Path al spec del pipeline.
109 spec: PathBuf,
110 /// Interponer un tap entre productor↔consumidor de cada edge para
111 /// discernir el TypeRef del flujo.
112 #[arg(long)]
113 tap: bool,
114 /// Variables `KEY=VALUE` para sustitución `${KEY}` en el spec.
115 #[arg(long = "var", value_parser = parse_kv)]
116 vars: Vec<(String, String)>,
117 /// Tras lanzar, suscribir al primer flow socket y volcar bytes
118 /// a stdout hasta EOF. Implica `--tap`.
119 #[arg(long)]
120 tail: bool,
121 },
122 /// Guardar un pipeline bajo un nombre (persiste con el snapshot).
123 Save {
124 /// Nombre simbólico.
125 name: String,
126 /// Path al spec.
127 spec: PathBuf,
128 },
129 /// Listar nombres de pipelines guardados.
130 SavedList,
131 /// Eliminar un pipeline guardado (no afecta runs en curso).
132 Drop { name: String },
133 /// Detener un pipeline en curso por ID (SIGTERM → grace → SIGKILL
134 /// sólo a sus comandos).
135 Stop {
136 /// ULID del pipeline (devuelto por `pipeline run`).
137 pipeline: String,
138 #[arg(long, default_value_t = 1000)]
139 grace_ms: u64,
140 },
141 /// Ejecutar un pipeline guardado por nombre.
142 RunSaved {
143 name: String,
144 #[arg(long)]
145 tap: bool,
146 #[arg(long = "var", value_parser = parse_kv)]
147 vars: Vec<(String, String)>,
148 #[arg(long)]
149 tail: bool,
150 },
151}
152
153fn parse_kv(s: &str) -> Result<(String, String), String> {
154 let (k, v) = s.split_once('=').ok_or_else(|| format!("expected KEY=VALUE, got `{s}`"))?;
155 Ok((k.to_string(), v.to_string()))
156}
157
158#[derive(Subcommand, Debug)]
159enum WsCmd {
160 /// Crear un workspace desde un spec TOML/JSON.
161 Create {
162 /// Path al spec del workspace.
163 spec: PathBuf,
164 },
165 /// Listar workspaces vivos.
166 List,
167 /// Detener un workspace por ID.
168 Stop {
169 id: String,
170 /// Milisegundos de gracia tras SIGTERM antes de SIGKILL.
171 #[arg(long, default_value_t = 1000)]
172 grace_ms: u64,
173 },
174 /// Resource accounting (RSS, CPU, comandos vivos).
175 Stats {
176 id: String,
177 },
178 /// Quota report: rlimits declarados vs uso actual.
179 Quota {
180 id: String,
181 },
182}
183
184#[tokio::main]
185async fn main() -> Result<()> {
186 let cli = Cli::parse();
187 let socket = cli.socket.unwrap_or_else(default_socket_path);
188 let mut stream = UnixStream::connect(&socket)
189 .await
190 .with_context(|| format!("connect {}", socket.display()))?;
191
192 match cli.cmd {
193 Cmd::Ping => {
194 let resp = round_trip(&mut stream, Request::Ping).await?;
195 match resp {
196 Response::Pong => println!("pong"),
197 other => print_unexpected(&other),
198 }
199 }
200
201 Cmd::Health => {
202 let resp = round_trip(&mut stream, Request::Health).await?;
203 match resp {
204 Response::Health {
205 version,
206 uptime_ms,
207 alive_workspaces,
208 alive_commands,
209 alive_pipelines,
210 active_flows,
211 dirty,
212 } => {
213 println!("version: {version}");
214 println!("uptime: {} ms", uptime_ms);
215 println!("alive_workspaces: {alive_workspaces}");
216 println!("alive_commands: {alive_commands}");
217 println!("alive_pipelines: {alive_pipelines}");
218 println!("active_flows: {active_flows}");
219 println!("dirty: {dirty}");
220 }
221 other => print_unexpected(&other),
222 }
223 }
224
225 Cmd::Caps => {
226 let resp = round_trip(&mut stream, Request::Capabilities).await?;
227 match resp {
228 Response::Capabilities {
229 kernel_version,
230 user_ns,
231 cgroup_v2,
232 cgroup_delegated,
233 has_cap_sys_admin,
234 } => {
235 println!("kernel: {}.{}.{}", kernel_version.0, kernel_version.1, kernel_version.2);
236 println!("user_ns: {user_ns}");
237 println!("cgroup_v2: {cgroup_v2}");
238 println!("cgroup_delegated: {cgroup_delegated}");
239 println!("cap_sys_admin: {has_cap_sys_admin}");
240 }
241 other => print_unexpected(&other),
242 }
243 }
244
245 Cmd::Workspace(WsCmd::Create { spec }) => {
246 let ws = load_workspace_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
247 let resp = round_trip(&mut stream, Request::WorkspaceCreate { spec: ws }).await?;
248 match resp {
249 Response::WorkspaceCreated { id, warnings } => {
250 println!("{id}");
251 for w in warnings {
252 eprintln!("warning: {w}");
253 }
254 }
255 Response::Error { message } => return Err(anyhow!(message)),
256 other => print_unexpected(&other),
257 }
258 }
259
260 Cmd::Workspace(WsCmd::List) => {
261 let resp = round_trip(&mut stream, Request::WorkspaceList).await?;
262 match resp {
263 Response::WorkspaceList { items } => {
264 if items.is_empty() {
265 println!("(no workspaces)");
266 }
267 for it in items {
268 println!(
269 "{} {:<20} cmds={} uptime={}ms",
270 it.id, it.label, it.commands, it.uptime_ms
271 );
272 }
273 }
274 other => print_unexpected(&other),
275 }
276 }
277
278 Cmd::Workspace(WsCmd::Stats { id }) => {
279 let id = parse_ws_id(&id)?;
280 let resp = round_trip(&mut stream, Request::WorkspaceStats { workspace: id }).await?;
281 match resp {
282 Response::WorkspaceStats { info } => {
283 println!("commands: {} alive / {} total", info.commands_alive, info.commands_total);
284 let fmt_mib = |b: u64| format!("{:.2} MiB", b as f64 / 1024.0 / 1024.0);
285 let rss = info.rss_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
286 let peak = info.rss_peak_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
287 let cpu = info
288 .cpu_usec
289 .map(|u| format!("{:.3} s", u as f64 / 1_000_000.0))
290 .unwrap_or_else(|| "—".into());
291 let cpu_pct = info
292 .cpu_percent
293 .map(|p| format!("{p:.1} % ({:.1}% total / {} cores)",
294 if info.cpu_cores > 0 { p / info.cpu_cores as f32 } else { p },
295 info.cpu_cores))
296 .unwrap_or_else(|| "— (esperando 2do sample)".into());
297 println!("rss: {rss}");
298 println!("rss_peak: {peak}");
299 println!("cpu: {cpu}");
300 println!("cpu_pct: {cpu_pct}");
301 println!("source: {}", info.source);
302 println!("uptime: {} ms", info.uptime_ms);
303 }
304 Response::Error { message } => return Err(anyhow!(message)),
305 other => print_unexpected(&other),
306 }
307 }
308
309 Cmd::Workspace(WsCmd::Quota { id }) => {
310 let id = parse_ws_id(&id)?;
311 let resp = round_trip(&mut stream, Request::WorkspaceQuota { workspace: id }).await?;
312 match resp {
313 Response::WorkspaceQuota { info } => {
314 let mem = info
315 .mem_limit
316 .map(|b| format!("{:.2} MiB", b as f64 / 1024.0 / 1024.0))
317 .unwrap_or_else(|| "—".into());
318 let nproc = info
319 .nproc_limit
320 .map(|n| n.to_string())
321 .unwrap_or_else(|| "—".into());
322 println!("mem_limit: {mem}");
323 println!("nproc_limit: {nproc}");
324 if info.breaches.is_empty() {
325 println!("breaches: (none — dentro de quota)");
326 } else {
327 println!("breaches:");
328 for b in info.breaches {
329 println!(" - {b}");
330 }
331 }
332 }
333 Response::Error { message } => return Err(anyhow!(message)),
334 other => print_unexpected(&other),
335 }
336 }
337
338 Cmd::Workspace(WsCmd::Stop { id, grace_ms }) => {
339 let id = parse_ws_id(&id)?;
340 let resp = round_trip(&mut stream, Request::WorkspaceStop { id, grace_ms }).await?;
341 match resp {
342 Response::WorkspaceStopped { id, reaped } => {
343 println!("stopped {id} (reaped {reaped})");
344 }
345 Response::Error { message } => return Err(anyhow!(message)),
346 other => print_unexpected(&other),
347 }
348 }
349
350 Cmd::Run { workspace, exec, argv, restart_on_failure } => {
351 let id = parse_ws_id(&workspace)?;
352 let resp = round_trip(
353 &mut stream,
354 Request::Run {
355 workspace: id,
356 exec,
357 argv,
358 envp: vec![],
359 restart_on_failure,
360 },
361 )
362 .await?;
363 match resp {
364 Response::RunStarted { command_id, pid, .. } => {
365 println!("{command_id} pid={pid}");
366 }
367 Response::Error { message } => return Err(anyhow!(message)),
368 other => print_unexpected(&other),
369 }
370 }
371
372 Cmd::Pipeline(PipeCmd::Run { spec, tap, vars, tail }) => {
373 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
374 // --tail implica --tap (no hay flow socket sin tap).
375 let effective_tap = tap || tail;
376 let resp = round_trip(
377 &mut stream,
378 Request::PipelineRun {
379 spec: p,
380 tap: effective_tap,
381 vars: vars.into_iter().collect(),
382 },
383 )
384 .await?;
385 let socket = print_pipeline_started_returning_socket(resp)?;
386 if tail {
387 if let Some(sock) = socket {
388 eprintln!("--- tailing {} ---", sock.display());
389 tail_socket(&sock).await?;
390 } else {
391 eprintln!("--tail: no hay flow socket disponible");
392 }
393 }
394 }
395
396 Cmd::Pipeline(PipeCmd::Save { name, spec }) => {
397 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
398 let resp = round_trip(&mut stream, Request::PipelineSave { name: name.clone(), spec: p }).await?;
399 match resp {
400 Response::PipelineSaved { name } => println!("saved {name}"),
401 Response::Error { message } => return Err(anyhow!(message)),
402 other => print_unexpected(&other),
403 }
404 }
405
406 Cmd::Pipeline(PipeCmd::SavedList) => {
407 let resp = round_trip(&mut stream, Request::PipelineSavedList).await?;
408 match resp {
409 Response::PipelineSavedList { names } => {
410 if names.is_empty() {
411 println!("(no saved pipelines)");
412 }
413 for n in names {
414 println!("{n}");
415 }
416 }
417 other => print_unexpected(&other),
418 }
419 }
420
421 Cmd::Pipeline(PipeCmd::Stop { pipeline, grace_ms }) => {
422 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
423 let resp = round_trip(&mut stream, Request::PipelineStop { pipeline: pid, grace_ms }).await?;
424 match resp {
425 Response::PipelineStopped { pipeline, reaped } => {
426 println!("stopped pipeline {pipeline} (reaped {reaped})");
427 }
428 other => print_unexpected(&other),
429 }
430 }
431
432 Cmd::Pipeline(PipeCmd::Drop { name }) => {
433 let resp = round_trip(&mut stream, Request::PipelineDrop { name }).await?;
434 match resp {
435 Response::PipelineDropped { name, existed } => {
436 if existed {
437 println!("dropped {name}");
438 } else {
439 eprintln!("no existía: {name}");
440 }
441 }
442 other => print_unexpected(&other),
443 }
444 }
445
446 Cmd::Pipeline(PipeCmd::RunSaved { name, tap, vars, tail }) => {
447 let effective_tap = tap || tail;
448 let resp = round_trip(
449 &mut stream,
450 Request::PipelineRunSaved {
451 name,
452 tap: effective_tap,
453 vars: vars.into_iter().collect(),
454 },
455 )
456 .await?;
457 let socket = print_pipeline_started_returning_socket(resp)?;
458 if tail {
459 if let Some(sock) = socket {
460 eprintln!("--- tailing {} ---", sock.display());
461 tail_socket(&sock).await?;
462 } else {
463 eprintln!("--tail: no hay flow socket disponible");
464 }
465 }
466 }
467
468 Cmd::Commands { workspace } => {
469 let ws = parse_ws_id(&workspace)?;
470 let resp = round_trip(&mut stream, Request::CommandList { workspace: ws }).await?;
471 match resp {
472 Response::CommandList { items } => {
473 if items.is_empty() {
474 println!("(no commands)");
475 }
476 for c in items {
477 let alive = if c.alive { "alive" } else { "exited" };
478 let exit = c
479 .exit_status
480 .map(|s| format!("exit={s}"))
481 .unwrap_or_default();
482 println!(
483 "{} {:<24} pid={:<7} {:<8} logs={} {}",
484 c.id, c.label, c.pid, alive, c.log_bytes, exit
485 );
486 }
487 }
488 other => print_unexpected(&other),
489 }
490 }
491
492 Cmd::Logs { workspace, command, tail, stream: which_stream, follow } => {
493 let ws = parse_ws_id(&workspace)?;
494 let cmd_id = Ulid::from_string(&command).map_err(|e| anyhow!("invalid command id: {e}"))?;
495 if !follow {
496 let resp = round_trip(
497 &mut stream,
498 Request::CommandLogs {
499 workspace: ws,
500 command: cmd_id,
501 tail_bytes: tail,
502 stream: which_stream,
503 },
504 )
505 .await?;
506 match resp {
507 Response::CommandLogs { bytes } => {
508 use std::io::Write;
509 let _ = std::io::stdout().write_all(&bytes);
510 let _ = std::io::stdout().flush();
511 }
512 Response::Error { message } => return Err(anyhow!(message)),
513 other => print_unexpected(&other),
514 }
515 } else {
516 // Follow mode: poll cada 200ms. Mantenemos el último buffer
517 // visto; cada round imprimimos el delta (suffix nuevo).
518 // Limitación: si el ring rota más rápido que el poll, perdemos
519 // bytes — pero el comportamiento es "best effort".
520 use std::io::Write;
521 let mut prev: Vec<u8> = Vec::new();
522 loop {
523 let resp = round_trip(
524 &mut stream,
525 Request::CommandLogs {
526 workspace: ws,
527 command: cmd_id,
528 tail_bytes: 0,
529 stream: which_stream.clone(),
530 },
531 )
532 .await?;
533 let bytes = match resp {
534 Response::CommandLogs { bytes } => bytes,
535 Response::Error { message } => return Err(anyhow!(message)),
536 other => {
537 print_unexpected(&other);
538 break;
539 }
540 };
541 // Imprimir suffix nuevo si bytes es extension de prev.
542 if bytes.len() >= prev.len() && bytes[..prev.len()] == prev[..] {
543 let _ = std::io::stdout().write_all(&bytes[prev.len()..]);
544 } else {
545 // Ring rotó — reset y print todo.
546 let _ = std::io::stdout().write_all(&bytes);
547 }
548 let _ = std::io::stdout().flush();
549 prev = bytes;
550
551 // Si el comando terminó, salir tras un último read.
552 let list_resp = round_trip(
553 &mut stream,
554 Request::CommandList { workspace: ws },
555 )
556 .await?;
557 let mut still_alive = false;
558 if let Response::CommandList { items } = list_resp {
559 if let Some(c) = items.iter().find(|c| c.id == cmd_id) {
560 still_alive = c.alive;
561 }
562 }
563 if !still_alive {
564 break;
565 }
566 tokio::time::sleep(std::time::Duration::from_millis(200)).await;
567 }
568 }
569 }
570
571 Cmd::Flow(FlowCmd::List) => {
572 let resp = round_trip(&mut stream, Request::FlowList).await?;
573 match resp {
574 Response::FlowList { items } => {
575 if items.is_empty() {
576 println!("(no active flows)");
577 }
578 for it in items {
579 println!("{}", it.pipeline);
580 for s in it.sockets {
581 println!(" {}", s.display());
582 }
583 }
584 }
585 other => print_unexpected(&other),
586 }
587 }
588
589 Cmd::Flow(FlowCmd::Throughput) => {
590 let resp = round_trip(&mut stream, Request::FlowThroughput).await?;
591 match resp {
592 Response::FlowThroughput { items } => {
593 if items.is_empty() {
594 println!("(no active flows)");
595 }
596 for it in items {
597 let name = it.socket.file_name()
598 .map(|n| n.to_string_lossy().to_string())
599 .unwrap_or_else(|| it.socket.display().to_string());
600 let kib = it.bytes_total as f64 / 1024.0;
601 let kbs = it.bytes_per_sec / 1024.0;
602 println!("{:<60} {:>8.1} KiB total {:>8.2} KiB/s", name, kib, kbs);
603 }
604 }
605 other => print_unexpected(&other),
606 }
607 }
608
609 Cmd::Flow(FlowCmd::Drop { pipeline }) => {
610 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
611 let resp = round_trip(&mut stream, Request::FlowDrop { pipeline: pid }).await?;
612 match resp {
613 Response::FlowDropped { pipeline, existed } => {
614 if existed {
615 println!("dropped {pipeline}");
616 } else {
617 eprintln!("no existía: {pipeline}");
618 }
619 }
620 other => print_unexpected(&other),
621 }
622 }
623
624 Cmd::Flow(FlowCmd::Tail { socket }) => {
625 // Subscribirse directo al socket — no pasamos por el daemon.
626 use tokio::io::AsyncReadExt;
627 let mut s = UnixStream::connect(&socket)
628 .await
629 .with_context(|| format!("connect {}", socket.display()))?;
630 let mut buf = [0u8; 4096];
631 loop {
632 let n = s.read(&mut buf).await?;
633 if n == 0 {
634 break;
635 }
636 use std::io::Write;
637 let _ = std::io::stdout().write_all(&buf[..n]);
638 let _ = std::io::stdout().flush();
639 }
640 }
641
642 Cmd::Discern { path } => {
643 let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
644 // Sample: hasta 4 KiB.
645 let sample = bytes.into_iter().take(4096).collect();
646 let resp = round_trip(
647 &mut stream,
648 Request::Discern {
649 sample,
650 hint_path: Some(path),
651 },
652 )
653 .await?;
654 match resp {
655 Response::Discernment { ty, confidence, mime, lens } => {
656 println!("type: {ty}");
657 println!("confidence: {confidence:.2}");
658 if let Some(m) = mime {
659 println!("mime: {m}");
660 }
661 if let Some(l) = lens {
662 println!("lens: {l}");
663 }
664 }
665 Response::Error { message } => return Err(anyhow!(message)),
666 other => print_unexpected(&other),
667 }
668 }
669 }
670
671 Ok(())
672}
673
674async fn round_trip(stream: &mut UnixStream, req: Request) -> Result<Response> {
675 write_frame(stream, &req).await?;
676 let resp: Response = read_frame(stream).await?;
677 Ok(resp)
678}
679
680fn parse_ws_id(s: &str) -> Result<WorkspaceId> {
681 let u = Ulid::from_string(s).map_err(|e| anyhow!("invalid workspace id: {e}"))?;
682 Ok(WorkspaceId(u))
683}
684
685fn print_unexpected(r: &Response) {
686 eprintln!("unexpected response: {r:?}");
687}
688
689/// Imprime el resultado del launch del pipeline y retorna el path del
690/// primer flow socket (si hay), útil para `--tail`.
691fn print_pipeline_started_returning_socket(resp: Response) -> Result<Option<PathBuf>> {
692 match resp {
693 Response::PipelineStarted { pipeline, command_pids, edges } => {
694 println!("pipeline {pipeline}");
695 for (label, pid) in command_pids {
696 println!(" {:<20} pid={pid}", label);
697 }
698 let mut first_socket: Option<PathBuf> = None;
699 if !edges.is_empty() {
700 println!("edges:");
701 for e in &edges {
702 println!(
703 " {}.{} → {}.{} ty={:?} mime={:?} conf={:.2}",
704 e.from_label, e.from_output, e.to_label, e.to_input,
705 e.ty, e.mime, e.confidence,
706 );
707 if first_socket.is_none() {
708 first_socket = e.flow_socket.clone();
709 }
710 }
711 }
712 Ok(first_socket)
713 }
714 Response::Error { message } => Err(anyhow!(message)),
715 other => {
716 print_unexpected(&other);
717 Ok(None)
718 }
719 }
720}
721
722async fn tail_socket(socket: &std::path::Path) -> Result<()> {
723 use tokio::io::AsyncReadExt;
724 // Pequeña ventana de retry — el daemon retiene el flow channel
725 // antes de retornar, así que en la práctica ya está bindeado.
726 let mut s = UnixStream::connect(socket)
727 .await
728 .with_context(|| format!("connect {}", socket.display()))?;
729 let mut buf = [0u8; 4096];
730 loop {
731 let n = s.read(&mut buf).await?;
732 if n == 0 {
733 break;
734 }
735 use std::io::Write;
736 let _ = std::io::stdout().write_all(&buf[..n]);
737 let _ = std::io::stdout().flush();
738 }
739 Ok(())
740}