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.
154pub fn print_scp_batch(
155    direction: &str,
156    results: &[crate::scp::HostScpResult],
157    max_concurrency: usize,
158    json: bool,
159) -> io::Result<()> {
160    if json {
161        let batch_run_id = BatchRunId::new().to_string_canonical();
162        let v = ScpBatchJson {
163            event: "scp-batch".into(),
164            batch_run_id,
165            direction: direction.into(),
166            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
167            results: results
168                .iter()
169                .map(|h| ScpHostJson {
170                    name: h.name.clone(),
171                    ok: h.ok,
172                    bytes: h.bytes,
173                    duration_ms: h.duration_ms,
174                    local: h.local.clone(),
175                    error: h.error.clone(),
176                })
177                .collect(),
178        };
179        return match json_wire::print_json_line(&v) {
180            Ok(()) => Ok(()),
181            Err(e) => {
182                report_json_serialize_error(&e);
183                Err(e)
184            }
185        };
186    }
187    if is_quiet() {
188        return Ok(());
189    }
190    let stdout = io::stdout();
191    let mut out = io::BufWriter::new(stdout.lock());
192    writeln!(
193        out,
194        "scp {direction} --all (max_concurrency={max_concurrency}, hosts={})",
195        results.len()
196    )?;
197    for h in results {
198        if h.ok {
199            writeln!(
200                out,
201                "  ok  {}  bytes={:?}  {:?}ms",
202                h.name, h.bytes, h.duration_ms
203            )?;
204        } else {
205            let err = h.error.as_deref().unwrap_or("error");
206            writeln!(out, "  ERR {}  {err}", h.name)?;
207        }
208    }
209    out.flush()
210}
211
212/// Prints an SCP transfer result as JSON (GAP-SSH-IO-007 / SCP-021 / IO-009).
213///
214/// # Errors
215/// Serialization or stdout I/O (including BrokenPipe).
216pub fn print_transfer_json(
217    direction: &str,
218    vps: &str,
219    local: &str,
220    remote: &str,
221    result: &crate::ssh::client::TransferResult,
222) -> io::Result<()> {
223    // GAP-SSH-IO-009: event discriminator (parity with tunnel_listening).
224    let v = ScpTransferJson {
225        ok: true,
226        event: "scp-transfer".into(),
227        direction: direction.to_string(),
228        vps: vps.to_string(),
229        local: local.to_string(),
230        remote: remote.to_string(),
231        bytes: result.bytes_transferred,
232        duration_ms: result.duration_ms,
233        mtime_preserved: result.mtime_preserved,
234        durable: result.durable,
235    };
236    match json_wire::print_json_line(&v) {
237        Ok(()) => Ok(()),
238        Err(e) => {
239            report_json_serialize_error(&e);
240            Err(e)
241        }
242    }
243}
244
245/// Prints an SFTP transfer result as JSON (G-SFTP-09).
246///
247/// # Errors
248/// Serialization or stdout I/O.
249#[cfg(feature = "ssh-real")]
250pub fn print_sftp_transfer_json(
251    direction: &str,
252    vps: &str,
253    local: &str,
254    remote: &str,
255    bytes: u64,
256    duration_ms: u64,
257    recursive: bool,
258) -> io::Result<()> {
259    let v = SftpTransferJson {
260        ok: true,
261        event: "sftp-transfer".into(),
262        direction: direction.to_string(),
263        vps: vps.to_string(),
264        local: local.to_string(),
265        remote: remote.to_string(),
266        bytes,
267        duration_ms,
268        recursive,
269    };
270    match json_wire::print_json_line(&v) {
271        Ok(()) => Ok(()),
272        Err(e) => {
273            report_json_serialize_error(&e);
274            Err(e)
275        }
276    }
277}
278
279/// Prints `sftp ls` JSON.
280///
281/// # Errors
282/// Serialization or stdout I/O.
283#[cfg(feature = "ssh-real")]
284pub fn print_sftp_list_json(vps: &str, path: &str, entries: &[SftpListEntry]) -> io::Result<()> {
285    let v = SftpListJson {
286        ok: true,
287        event: "sftp-list".into(),
288        vps: vps.to_string(),
289        path: path.to_string(),
290        entries: entries
291            .iter()
292            .map(|e| SftpListEntryJson {
293                name: e.name.clone(),
294                path: e.path.clone(),
295                kind: e.kind.clone(),
296                size: e.size,
297                mode: e.mode,
298            })
299            .collect(),
300    };
301    match json_wire::print_json_line(&v) {
302        Ok(()) => Ok(()),
303        Err(e) => {
304            report_json_serialize_error(&e);
305            Err(e)
306        }
307    }
308}
309
310/// Prints `sftp` fs-op JSON (mkdir/rmdir/rm/rename).
311///
312/// # Errors
313/// Serialization or stdout I/O.
314#[cfg(feature = "ssh-real")]
315pub fn print_sftp_fs_op_json(
316    op: &str,
317    vps: &str,
318    path: &str,
319    to: Option<&str>,
320    duration_ms: u64,
321) -> io::Result<()> {
322    let v = SftpFsOpJson {
323        ok: true,
324        event: "sftp-fs-op".into(),
325        op: op.to_string(),
326        vps: vps.to_string(),
327        path: path.to_string(),
328        to: to.map(str::to_owned),
329        duration_ms,
330        kind: None,
331        size: None,
332        mode: None,
333        mtime: None,
334    };
335    match json_wire::print_json_line(&v) {
336        Ok(()) => Ok(()),
337        Err(e) => {
338            report_json_serialize_error(&e);
339            Err(e)
340        }
341    }
342}
343
344/// Prints `sftp stat` JSON.
345///
346/// # Errors
347/// Serialization or stdout I/O.
348#[cfg(feature = "ssh-real")]
349pub fn print_sftp_stat_json(vps: &str, st: &SftpStat) -> io::Result<()> {
350    let v = SftpFsOpJson {
351        ok: true,
352        event: "sftp-fs-op".into(),
353        op: "stat".into(),
354        vps: vps.to_string(),
355        path: st.path.clone(),
356        to: None,
357        duration_ms: 0,
358        kind: Some(st.kind.clone()),
359        size: st.size,
360        mode: st.mode,
361        mtime: st.mtime,
362    };
363    match json_wire::print_json_line(&v) {
364        Ok(()) => Ok(()),
365        Err(e) => {
366            report_json_serialize_error(&e);
367            Err(e)
368        }
369    }
370}
371
372/// Prints multi-host SFTP batch results.
373///
374/// # Errors
375/// Serialization or stdout I/O.
376#[cfg(feature = "ssh-real")]
377pub fn print_sftp_batch(
378    direction: &str,
379    results: &[HostSftpResult],
380    max_concurrency: usize,
381    json: bool,
382) -> io::Result<()> {
383    if json {
384        let batch_run_id = BatchRunId::new().to_string_canonical();
385        let v = SftpBatchJson {
386            event: "sftp-batch".into(),
387            batch_run_id,
388            direction: direction.to_string(),
389            max_concurrency: u32::try_from(max_concurrency).unwrap_or(u32::MAX),
390            results: results
391                .iter()
392                .map(|h| ScpHostJson {
393                    name: h.name.clone(),
394                    ok: h.ok,
395                    bytes: h.bytes,
396                    duration_ms: h.duration_ms,
397                    local: h.local.clone(),
398                    error: h.error.clone(),
399                })
400                .collect(),
401        };
402        return match json_wire::print_json_line(&v) {
403            Ok(()) => Ok(()),
404            Err(e) => {
405                report_json_serialize_error(&e);
406                Err(e)
407            }
408        };
409    }
410    if is_quiet() {
411        return Ok(());
412    }
413    let stdout = io::stdout();
414    let mut out = io::BufWriter::new(stdout.lock());
415    writeln!(
416        out,
417        "sftp {direction} --all (max_concurrency={max_concurrency}, hosts={})",
418        results.len()
419    )?;
420    for h in results {
421        if h.ok {
422            writeln!(
423                out,
424                "  ok  {}  bytes={:?}  ms={:?}",
425                h.name, h.bytes, h.duration_ms
426            )?;
427        } else {
428            let err = h.error.as_deref().unwrap_or("error");
429            writeln!(out, "  ERR {}  {err}", h.name)?;
430        }
431    }
432    out.flush()
433}
434
435/// Builds the `tunnel_listening` payload without writing it anywhere.
436///
437/// Split out from the printer so the document can be asserted on. Every field of this
438/// event was previously reachable only by driving a real listener and reading stdout,
439/// which is why `bind` — added by G-TUN-R06 precisely so an agent could audit whether a
440/// service had been published beyond loopback — shipped covered by nothing but a schema
441/// file and a mention in prose.
442#[must_use]
443pub fn build_tunnel_listening(
444    vps: &str,
445    local_port: u16,
446    remote_host: &str,
447    remote_port: u16,
448    timeout_ms: u64,
449    bind: &str,
450    mode: &str,
451) -> TunnelListeningJson {
452    TunnelListeningJson {
453        ok: true,
454        event: "tunnel_listening".into(),
455        vps: vps.to_string(),
456        local_port,
457        remote_host: remote_host.to_string(),
458        remote_port,
459        timeout_ms,
460        bind: bind.to_string(),
461        mode: mode.to_string(),
462    }
463}
464
465/// JSON event when the local tunnel listener comes up (GAP-SSH-IO-008).
466///
467/// # Errors
468/// Serialization or stdout I/O (including BrokenPipe).
469pub fn print_tunnel_listening_json(
470    vps: &str,
471    local_port: u16,
472    remote_host: &str,
473    remote_port: u16,
474    timeout_ms: u64,
475    bind: &str,
476    mode: &str,
477) -> io::Result<()> {
478    let v = build_tunnel_listening(
479        vps,
480        local_port,
481        remote_host,
482        remote_port,
483        timeout_ms,
484        bind,
485        mode,
486    );
487    match json_wire::print_json_line(&v) {
488        Ok(()) => Ok(()),
489        Err(e) => {
490            report_json_serialize_error(&e);
491            Err(e)
492        }
493    }
494}
495
496/// Builds the `tunnel_closed` payload without writing it anywhere.
497///
498/// The whole event — `reason`, `forwards_served`, `capacity_waits`, `ok` — used to be
499/// constructed inside the printer, so nothing could inspect it. The 0.5.4 audit found
500/// the three field names appearing in exactly one place outside the emitter: a
501/// documentation test asserting the CHANGELOG mentions them. Deleting the emission
502/// would not have turned the suite red, which is the same failure mode G-QA-R01 was
503/// written to stop.
504#[must_use]
505pub fn build_tunnel_closed(input: TunnelClosedInput<'_>) -> TunnelClosedJson {
506    TunnelClosedJson {
507        // An accept-error shutdown is not a clean lifetime, even though the process
508        // still exits 0 for having bound successfully.
509        ok: !matches!(input.reason, TunnelCloseReason::AcceptError),
510        event: "tunnel_closed".into(),
511        vps: input.vps.to_string(),
512        reason: input.reason,
513        bind: input.bind.to_string(),
514        local_port: input.local_port,
515        forwards_served: input.forwards_served,
516        capacity_waits: input.capacity_waits,
517        duration_ms: input.duration_ms,
518        mode: input.mode.to_string(),
519    }
520}
521
522/// Inputs for [`build_tunnel_closed`].
523///
524/// B3: three of the eight fields are bare `u64` counters and one is a `u16`
525/// port. Passed positionally, swapping `forwards_served` with `capacity_waits`
526/// compiles and produces a plausible-looking event that misreports the tunnel's
527/// lifetime — the exact class of silent wrongness the suppressed
528/// `too_many_arguments` lint was pointing at.
529pub struct TunnelClosedInput<'a> {
530    /// Registry name of the relay host.
531    pub vps: &'a str,
532    /// Why the tunnel stopped.
533    pub reason: TunnelCloseReason,
534    /// Effective local bind address.
535    pub bind: &'a str,
536    /// Effective local port (OS-assigned when `0` was requested).
537    pub local_port: u16,
538    /// Connections accepted over the tunnel's lifetime.
539    pub forwards_served: u64,
540    /// Times an accept waited on the concurrency semaphore.
541    pub capacity_waits: u64,
542    /// Wall lifetime in milliseconds.
543    pub duration_ms: u64,
544    /// Tunnel mode label (`local`, `reverse`, `socks5`, `streamlocal`).
545    pub mode: &'a str,
546}
547
548/// Emits the `tunnel_closed` shutdown event (G-TUN-R07).
549///
550/// Always emitted, including on the happy deadline path, so an agent can tell the
551/// three endings apart instead of inferring them from a shared exit 0.
552///
553/// # Errors
554/// Serialization or stdout I/O (including BrokenPipe).
555pub fn print_tunnel_closed_json(event: &TunnelClosedJson) -> io::Result<()> {
556    match json_wire::print_json_line(event) {
557        Ok(()) => Ok(()),
558        Err(e) => {
559            report_json_serialize_error(&e);
560            Err(e)
561        }
562    }
563}
ssh_cli/output/batch.rs

ssh_cli/output/
batch.rs
