ssh_cli/output/
batch.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Multi-host / multi-result batch emitters (G-COMP-06d).
3//!
4//! Keeps fan-out JSON/text formatting separate from single-record CRUD emitters.
5#![forbid(unsafe_code)]
6
7use super::{is_quiet, report_json_serialize_error};
8use crate::domain::BatchRunId;
9use crate::json_wire::{
10    self, ExecBatchJson, ExecHostJson, HealthBatchJson, HealthHostJson, ScpBatchJson, ScpHostJson,
11    ScpTransferJson, TunnelCloseReason, TunnelClosedJson, TunnelListeningJson,
12};
13#[cfg(feature = "ssh-real")]
14use crate::json_wire::{
15    SftpBatchJson, SftpFsOpJson, SftpListEntryJson, SftpListJson, SftpTransferJson,
16};
17// A6: the SFTP emitters below take types owned by the russh-backed subsystem, so
18// they only exist when that stack is compiled in.
19#[cfg(feature = "ssh-real")]
20use crate::sftp::batch::HostSftpResult;
21#[cfg(feature = "ssh-real")]
22use crate::ssh::sftp_types::{SftpListEntry, SftpStat};
23use crate::vps::{HostExecResult, HostHealthResult};
24use std::io::{self, Write};
25
26/// Prints multi-host health-check results (text or single-root JSON batch).
27///
28/// # Errors
29/// Serialization or stdout I/O.
30pub fn print_health_batch(
31    results: &[HostHealthResult],
32    max_concurrency: usize,
33    json: bool,
34) -> io::Result<()> {
35    if json {
36        // One v7 id per fan-out command (before/with emit; not per host).
37        let batch_run_id = BatchRunId::new().to_string_canonical();
38        let v = HealthBatchJson {
39            event: "health-check-batch".into(),
40            batch_run_id,
41            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
42            results: results
43                .iter()
44                .map(|h| HealthHostJson {
45                    name: h.name.clone(),
46                    status: if h.ok { "ok".into() } else { "error".into() },
47                    latency_ms: h.latency_ms,
48                    error: h.error.clone(),
49                })
50                .collect(),
51        };
52        return match json_wire::print_json_line(&v) {
53            Ok(()) => Ok(()),
54            Err(e) => {
55                report_json_serialize_error(&e);
56                Err(e)
57            }
58        };
59    }
60    if is_quiet() {
61        return Ok(());
62    }
63    let stdout = io::stdout();
64    let mut out = io::BufWriter::new(stdout.lock());
65    writeln!(
66        out,
67        "health-check --all (max_concurrency={max_concurrency}, hosts={})",
68        results.len()
69    )?;
70    for h in results {
71        match (h.ok, h.latency_ms) {
72            (true, Some(ms)) => writeln!(out, "  ok  {}  {ms}ms", h.name)?,
73            (true, None) => writeln!(out, "  ok  {}", h.name)?,
74            (false, _) => {
75                let err = h.error.as_deref().unwrap_or("error");
76                writeln!(out, "  ERR {}  {err}", h.name)?;
77            }
78        }
79    }
80    out.flush()
81}
82
83/// Prints multi-host exec results (text or single-root JSON batch).
84///
85/// # Errors
86/// Serialization or stdout I/O.
87pub fn print_exec_batch(
88    results: &[HostExecResult],
89    max_concurrency: usize,
90    json: bool,
91) -> io::Result<()> {
92    if json {
93        let batch_run_id = BatchRunId::new().to_string_canonical();
94        let v = ExecBatchJson {
95            event: "exec-batch".into(),
96            batch_run_id,
97            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
98            results: results
99                .iter()
100                .map(|h| ExecHostJson {
101                    name: h.name.clone(),
102                    ok: h.ok,
103                    exit_code: h.exit_code,
104                    stdout: h.stdout.clone(),
105                    stderr: h.stderr.clone(),
106                    duration_ms: h.duration_ms,
107                    error: h.error.clone(),
108                })
109                .collect(),
110        };
111        return match json_wire::print_json_line(&v) {
112            Ok(()) => Ok(()),
113            Err(e) => {
114                report_json_serialize_error(&e);
115                Err(e)
116            }
117        };
118    }
119    if is_quiet() {
120        return Ok(());
121    }
122    let stdout = io::stdout();
123    let mut out = io::BufWriter::new(stdout.lock());
124    writeln!(
125        out,
126        "exec --all (max_concurrency={max_concurrency}, hosts={})",
127        results.len()
128    )?;
129    for h in results {
130        let status = if h.ok { "ok" } else { "ERR" };
131        writeln!(
132            out,
133            "  {status}  {}  exit={:?}  {}ms",
134            h.name, h.exit_code, h.duration_ms
135        )?;
136        if !h.stdout.is_empty() {
137            for line in h.stdout.lines() {
138                writeln!(out, "    | {line}")?;
139            }
140        }
141        if !h.stderr.is_empty() {
142            for line in h.stderr.lines() {
143                writeln!(out, "    ! {line}")?;
144            }
145        }
146    }
147    out.flush()
148}
149
150/// Prints multi-host SCP batch results.
151///
152/// # Errors
153/// Serialization or stdout I/O.
154/// `source` is a parameter rather than a constant because this envelope serves two
155/// different designations: the fleet paths (`--all` / `--hosts`) are `selector`, while
156/// single-host multi-file (G-PAR-37 / G-PAR-47) reuses the same shape for a host the
157/// caller typed. Hardcoding `selector` here would make the second case lie about how
158/// its target was chosen — the precise class of silence this field exists to end.
159pub fn print_scp_batch(
160    direction: &str,
161    results: &[crate::scp::HostScpResult],
162    max_concurrency: usize,
163    json: bool,
164    source: json_wire::TargetSource,
165) -> io::Result<()> {
166    if json {
167        let batch_run_id = BatchRunId::new().to_string_canonical();
168        let v = ScpBatchJson {
169            event: "scp-batch".into(),
170            target_source: source,
171            host_source: source,
172            batch_run_id,
173            direction: direction.into(),
174            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
175            results: results
176                .iter()
177                .map(|h| ScpHostJson {
178                    name: h.name.clone(),
179                    ok: h.ok,
180                    bytes: h.bytes,
181                    duration_ms: h.duration_ms,
182                    local: h.local.clone(),
183                    error: h.error.clone(),
184                })
185                .collect(),
186        };
187        return match json_wire::print_json_line(&v) {
188            Ok(()) => Ok(()),
189            Err(e) => {
190                report_json_serialize_error(&e);
191                Err(e)
192            }
193        };
194    }
195    if is_quiet() {
196        return Ok(());
197    }
198    let stdout = io::stdout();
199    let mut out = io::BufWriter::new(stdout.lock());
200    writeln!(
201        out,
202        "scp {direction} --all (max_concurrency={max_concurrency}, hosts={})",
203        results.len()
204    )?;
205    for h in results {
206        if h.ok {
207            writeln!(
208                out,
209                "  ok  {}  bytes={:?}  {:?}ms",
210                h.name, h.bytes, h.duration_ms
211            )?;
212        } else {
213            let err = h.error.as_deref().unwrap_or("error");
214            writeln!(out, "  ERR {}  {err}", h.name)?;
215        }
216    }
217    out.flush()
218}
219
220/// Prints an SCP transfer result as JSON (GAP-SSH-IO-007 / SCP-021 / IO-009).
221///
222/// # Errors
223/// Serialization or stdout I/O (including BrokenPipe).
224///
225/// Takes the resolved [`json_wire::ExecTarget`] rather than a bare name so the
226/// provenance travels with the host instead of as an eighth parameter: this function
227/// already sits at the `clippy::too_many_arguments` ceiling that
228/// `gaps_v062_component_budget` refuses to spend another allow on.
229pub fn print_transfer_json(
230    direction: &str,
231    target: &json_wire::ExecTarget,
232    local: &str,
233    remote: &str,
234    result: &crate::ssh::client::TransferResult,
235) -> io::Result<()> {
236    // GAP-SSH-IO-009: event discriminator (parity with tunnel_listening).
237    let v = ScpTransferJson {
238        ok: true,
239        event: "scp-transfer".into(),
240        target: json_wire::TargetEcho::new(target),
241        direction: direction.to_string(),
242        vps: target.host.clone(),
243        local: local.to_string(),
244        remote: remote.to_string(),
245        bytes: result.bytes_transferred,
246        duration_ms: result.duration_ms,
247        mtime_preserved: result.mtime_preserved,
248        durable: result.durable,
249    };
250    match json_wire::print_json_line(&v) {
251        Ok(()) => Ok(()),
252        Err(e) => {
253            report_json_serialize_error(&e);
254            Err(e)
255        }
256    }
257}
258
259/// Prints an SFTP transfer result as JSON (G-SFTP-09).
260///
261/// # Errors
262/// Serialization or stdout I/O.
263#[cfg(feature = "ssh-real")]
264pub fn print_sftp_transfer_json(
265    direction: &str,
266    target: &json_wire::ExecTarget,
267    local: &str,
268    remote: &str,
269    bytes: u64,
270    duration_ms: u64,
271    recursive: bool,
272) -> io::Result<()> {
273    let v = SftpTransferJson {
274        ok: true,
275        event: "sftp-transfer".into(),
276        target: json_wire::TargetEcho::new(target),
277        direction: direction.to_string(),
278        vps: target.host.clone(),
279        local: local.to_string(),
280        remote: remote.to_string(),
281        bytes,
282        duration_ms,
283        recursive,
284    };
285    match json_wire::print_json_line(&v) {
286        Ok(()) => Ok(()),
287        Err(e) => {
288            report_json_serialize_error(&e);
289            Err(e)
290        }
291    }
292}
293
294/// Prints `sftp ls` JSON.
295///
296/// # Errors
297/// Serialization or stdout I/O.
298#[cfg(feature = "ssh-real")]
299pub fn print_sftp_list_json(vps: &str, path: &str, entries: &[SftpListEntry]) -> io::Result<()> {
300    let v = SftpListJson {
301        ok: true,
302        event: "sftp-list".into(),
303        vps: vps.to_string(),
304        path: path.to_string(),
305        entries: entries
306            .iter()
307            .map(|e| SftpListEntryJson {
308                name: e.name.clone(),
309                path: e.path.clone(),
310                kind: e.kind.clone(),
311                size: e.size,
312                mode: e.mode,
313            })
314            .collect(),
315    };
316    match json_wire::print_json_line(&v) {
317        Ok(()) => Ok(()),
318        Err(e) => {
319            report_json_serialize_error(&e);
320            Err(e)
321        }
322    }
323}
324
325/// Prints `sftp` fs-op JSON (mkdir/rmdir/rm/rename).
326///
327/// # Errors
328/// Serialization or stdout I/O.
329#[cfg(feature = "ssh-real")]
330pub fn print_sftp_fs_op_json(
331    op: &str,
332    vps: &str,
333    path: &str,
334    to: Option<&str>,
335    duration_ms: u64,
336) -> io::Result<()> {
337    let v = SftpFsOpJson {
338        ok: true,
339        event: "sftp-fs-op".into(),
340        op: op.to_string(),
341        vps: vps.to_string(),
342        path: path.to_string(),
343        to: to.map(str::to_owned),
344        duration_ms,
345        kind: None,
346        size: None,
347        mode: None,
348        mtime: None,
349    };
350    match json_wire::print_json_line(&v) {
351        Ok(()) => Ok(()),
352        Err(e) => {
353            report_json_serialize_error(&e);
354            Err(e)
355        }
356    }
357}
358
359/// Prints `sftp stat` JSON.
360///
361/// # Errors
362/// Serialization or stdout I/O.
363#[cfg(feature = "ssh-real")]
364pub fn print_sftp_stat_json(vps: &str, st: &SftpStat) -> io::Result<()> {
365    let v = SftpFsOpJson {
366        ok: true,
367        event: "sftp-fs-op".into(),
368        op: "stat".into(),
369        vps: vps.to_string(),
370        path: st.path.clone(),
371        to: None,
372        duration_ms: 0,
373        kind: Some(st.kind.clone()),
374        size: st.size,
375        mode: st.mode,
376        mtime: st.mtime,
377    };
378    match json_wire::print_json_line(&v) {
379        Ok(()) => Ok(()),
380        Err(e) => {
381            report_json_serialize_error(&e);
382            Err(e)
383        }
384    }
385}
386
387/// Prints multi-host SFTP batch results.
388///
389/// # Errors
390/// Serialization or stdout I/O.
391#[cfg(feature = "ssh-real")]
392pub fn print_sftp_batch(
393    direction: &str,
394    results: &[HostSftpResult],
395    max_concurrency: usize,
396    json: bool,
397) -> io::Result<()> {
398    if json {
399        let batch_run_id = BatchRunId::new().to_string_canonical();
400        let v = SftpBatchJson {
401            event: "sftp-batch".into(),
402            // See `print_scp_batch`: a batch is by construction selector-produced.
403            target_source: json_wire::TargetSource::Selector,
404            host_source: json_wire::TargetSource::Selector,
405            batch_run_id,
406            direction: direction.to_string(),
407            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
408            results: results
409                .iter()
410                .map(|h| ScpHostJson {
411                    name: h.name.clone(),
412                    ok: h.ok,
413                    bytes: h.bytes,
414                    duration_ms: h.duration_ms,
415                    local: h.local.clone(),
416                    error: h.error.clone(),
417                })
418                .collect(),
419        };
420        return match json_wire::print_json_line(&v) {
421            Ok(()) => Ok(()),
422            Err(e) => {
423                report_json_serialize_error(&e);
424                Err(e)
425            }
426        };
427    }
428    if is_quiet() {
429        return Ok(());
430    }
431    let stdout = io::stdout();
432    let mut out = io::BufWriter::new(stdout.lock());
433    writeln!(
434        out,
435        "sftp {direction} --all (max_concurrency={max_concurrency}, hosts={})",
436        results.len()
437    )?;
438    for h in results {
439        if h.ok {
440            writeln!(
441                out,
442                "  ok  {}  bytes={:?}  ms={:?}",
443                h.name, h.bytes, h.duration_ms
444            )?;
445        } else {
446            let err = h.error.as_deref().unwrap_or("error");
447            writeln!(out, "  ERR {}  {err}", h.name)?;
448        }
449    }
450    out.flush()
451}
452
453/// Builds the `tunnel_listening` payload without writing it anywhere.
454///
455/// Split out from the printer so the document can be asserted on. Every field of this
456/// event was previously reachable only by driving a real listener and reading stdout,
457/// which is why `bind` — added by G-TUN-R06 precisely so an agent could audit whether a
458/// service had been published beyond loopback — shipped covered by nothing but a schema
459/// file and a mention in prose.
460#[must_use]
461pub fn build_tunnel_listening(
462    vps: &str,
463    local_port: u16,
464    remote_host: &str,
465    remote_port: u16,
466    timeout_ms: u64,
467    bind: &str,
468    mode: &str,
469) -> TunnelListeningJson {
470    TunnelListeningJson {
471        ok: true,
472        event: "tunnel_listening".into(),
473        vps: vps.to_string(),
474        local_port,
475        remote_host: remote_host.to_string(),
476        remote_port,
477        timeout_ms,
478        bind: bind.to_string(),
479        mode: mode.to_string(),
480    }
481}
482
483/// JSON event when the local tunnel listener comes up (GAP-SSH-IO-008).
484///
485/// # Errors
486/// Serialization or stdout I/O (including BrokenPipe).
487pub fn print_tunnel_listening_json(
488    vps: &str,
489    local_port: u16,
490    remote_host: &str,
491    remote_port: u16,
492    timeout_ms: u64,
493    bind: &str,
494    mode: &str,
495) -> io::Result<()> {
496    let v = build_tunnel_listening(
497        vps,
498        local_port,
499        remote_host,
500        remote_port,
501        timeout_ms,
502        bind,
503        mode,
504    );
505    match json_wire::print_json_line(&v) {
506        Ok(()) => Ok(()),
507        Err(e) => {
508            report_json_serialize_error(&e);
509            Err(e)
510        }
511    }
512}
513
514/// Builds the `tunnel_closed` payload without writing it anywhere.
515///
516/// The whole event — `reason`, `forwards_served`, `capacity_waits`, `ok` — used to be
517/// constructed inside the printer, so nothing could inspect it. The 0.5.4 audit found
518/// the three field names appearing in exactly one place outside the emitter: a
519/// documentation test asserting the CHANGELOG mentions them. Deleting the emission
520/// would not have turned the suite red, which is the same failure mode G-QA-R01 was
521/// written to stop.
522#[must_use]
523pub fn build_tunnel_closed(input: TunnelClosedInput<'_>) -> TunnelClosedJson {
524    TunnelClosedJson {
525        // An accept-error shutdown is not a clean lifetime, even though the process
526        // still exits 0 for having bound successfully.
527        ok: !matches!(input.reason, TunnelCloseReason::AcceptError),
528        event: "tunnel_closed".into(),
529        vps: input.vps.to_string(),
530        reason: input.reason,
531        bind: input.bind.to_string(),
532        local_port: input.local_port,
533        forwards_served: input.forwards_served,
534        capacity_waits: input.capacity_waits,
535        duration_ms: input.duration_ms,
536        mode: input.mode.to_string(),
537    }
538}
539
540/// Inputs for [`build_tunnel_closed`].
541///
542/// B3: three of the eight fields are bare `u64` counters and one is a `u16`
543/// port. Passed positionally, swapping `forwards_served` with `capacity_waits`
544/// compiles and produces a plausible-looking event that misreports the tunnel's
545/// lifetime — the exact class of silent wrongness the suppressed
546/// `too_many_arguments` lint was pointing at.
547pub struct TunnelClosedInput<'a> {
548    /// Registry name of the relay host.
549    pub vps: &'a str,
550    /// Why the tunnel stopped.
551    pub reason: TunnelCloseReason,
552    /// Effective local bind address.
553    pub bind: &'a str,
554    /// Effective local port (OS-assigned when `0` was requested).
555    pub local_port: u16,
556    /// Connections accepted over the tunnel's lifetime.
557    pub forwards_served: u64,
558    /// Times an accept waited on the concurrency semaphore.
559    pub capacity_waits: u64,
560    /// Wall lifetime in milliseconds.
561    pub duration_ms: u64,
562    /// Tunnel mode label (`local`, `reverse`, `socks5`, `streamlocal`).
563    pub mode: &'a str,
564}
565
566/// Emits the `tunnel_closed` shutdown event (G-TUN-R07).
567///
568/// Always emitted, including on the happy deadline path, so an agent can tell the
569/// three endings apart instead of inferring them from a shared exit 0.
570///
571/// # Errors
572/// Serialization or stdout I/O (including BrokenPipe).
573pub fn print_tunnel_closed_json(event: &TunnelClosedJson) -> io::Result<()> {
574    match json_wire::print_json_line(event) {
575        Ok(()) => Ok(()),
576        Err(e) => {
577            report_json_serialize_error(&e);
578            Err(e)
579        }
580    }
581}
ssh_cli/output/batch.rs

ssh_cli/output/
batch.rs
