cerulion_cli 2.0.0

// SPDX-License-Identifier: AGPL-3.0-only
//! Pins that `cerulion clean` actually RUNS the state-file
//! diagnostic, and runs it in the right ORDER.
//!
//! # Why a source walk
//!
//! `report_shm_state_population` reads the one directory
//! `iceoryx2-pal-posix` hardcodes — literally `/tmp/` — and, without
//! `--report-only`, DELETES from it. A behavioural test of that function
//! would therefore have to operate on the developer's (or the runner's) real
//! `/tmp`, which a test must never touch and which would race every other
//! process on the machine. The engine half is fully oracle-tested over a tempdir with an
//! injected system probe (`cerulion_cli_engine::shm_state`); what CANNOT be
//! reached that way is the one thing the CLI owns:
//!
//! * that the diagnostic is CALLED at all — the classic inert-shipping shape,
//!   where every engine arm stays green while the verb does nothing, and
//! * that the dead-node SWEEP runs FIRST. Ordering is a correctness property
//!   here, not tidiness: the sweep removes state files of its own through
//!   iceoryx2's `shm_unlink`, so a population reported before it over-counts,
//!   and a reclamation running first would unlink objects the sweep was about
//!   to inspect.
//!
//! Reading source is a weaker oracle than running code, so the walk is
//! literal about what it recognises and carries an anti-tautology arm proving
//! the stripped view still contains the code it claims to read.

use std::path::{Path, PathBuf};

fn main_rs() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("src/main.rs")
}

fn read_main() -> String {
    let p = main_rs();
    std::fs::read_to_string(&p).unwrap_or_else(|e| panic!("read {}: {e}", p.display()))
}

/// Strip `//`-to-end-of-line comments, so the names this module's own docs and
/// the function's doc comment mention are never read as calls.
fn code_only(src: &str) -> String {
    src.lines()
        .map(|line| match line.find("//") {
            Some(i) => &line[..i],
            None => line,
        })
        .collect::<Vec<_>>()
        .join("\n")
}

/// The brace-matched body of `fn <name>`, so an assertion about ONE function
/// cannot be satisfied by a call somewhere else in the file.
///
/// SCOPE: the count includes braces inside string literals, so a body
/// holding an UNBALANCED one would end the scope in the wrong place. Both
/// bodies this file reads hold only balanced ones (`println!("{line}")`), and
/// an unbalanced literal would show up as a panic here rather than as a
/// silently weaker assertion — the fail-loud direction.
fn fn_body(src: &str, name: &str) -> String {
    let sig = format!("fn {name}");
    let start = src
        .find(&sig)
        .unwrap_or_else(|| panic!("`{sig}` must exist in main.rs"));
    let open = src[start..]
        .find('{')
        .unwrap_or_else(|| panic!("`{sig}` must have a body"))
        + start;
    let bytes = src.as_bytes();
    let mut depth = 0usize;
    for (i, b) in bytes.iter().enumerate().skip(open) {
        match b {
            b'{' => depth += 1,
            b'}' => {
                depth -= 1;
                if depth == 0 {
                    return src[open..=i].to_string();
                }
            }
            _ => {}
        }
    }
    panic!("`{sig}` body is not brace-balanced");
}

#[test]
fn the_clean_verb_runs_the_state_file_diagnostic() {
    // Without this the whole feature ships INERT: every engine arm stays
    // green while `cerulion clean` prints exactly what it printed before.
    let src = code_only(&read_main());
    let body = fn_body(&src, "clean_iceoryx2_state");
    assert!(
        body.contains("report_shm_state_population("),
        "`cerulion clean` must call the shm-state population diagnostic; body was:\n{body}"
    );
}

#[test]
fn the_dead_node_sweep_runs_before_the_state_file_diagnostic() {
    // The sweep removes state files of its own (iceoryx2's `shm_unlink`
    // removes the file its object was named by), so a population reported
    // before it over-counts, and a reclamation running first would unlink
    // objects the sweep was about to inspect.
    let src = code_only(&read_main());
    let body = fn_body(&src, "clean_iceoryx2_state");
    let sweep = body
        .find("clean_dead_nodes(")
        .expect("the sweep must still be called");
    let diagnostic = body
        .find("report_shm_state_population(")
        .expect("the diagnostic must still be called");
    assert!(
        sweep < diagnostic,
        "the dead-node sweep must precede the state-file diagnostic; body was:\n{body}"
    );
}

#[test]
fn the_reclamation_stands_down_when_dead_nodes_are_still_registered() {
    // A `.shm_state` file is the ONLY mapping from an iceoryx2
    // resource name to the object behind it, so removing one a still-registered
    // dead node needs makes that node permanently unreclaimable — no later
    // sweep can read its details again. Measured, against a control, in
    // `crates/cerulion_cli_engine/tests/reclaim_ordering_test.rs`.
    //
    // Structural rather than behavioural for the same reason the arms above
    // are: the reclamation reads and deletes from the developer's real `/tmp`.
    let src = code_only(&read_main());
    let body = fn_body(&src, "clean_iceoryx2_state");
    // The CONJUNCTION, not merely a mention of `converged`. Reclaiming requires
    // BOTH "the operator asked for it" and "the registry converged", so an
    // `||` here would reclaim on a wedged desk whenever `--report-only` was
    // absent — which is the whole defect. A bare `contains("converged")` is
    // satisfied by that variant too, so it is pinned as a whole expression.
    let normalized = body.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        normalized.contains("let reclaim = mode.removes() && converged;"),
        "`cerulion clean` must reclaim only when the operator asked AND the sweep left no \
         dead nodes registered — an `||` would reclaim on a wedged desk; body was:\n{body}"
    );
    // The DESTRUCTIVE half is what the gate must govern: `report_only` alone
    // reaching the diagnostic would leave the reclamation ungated.
    let call = body
        .find("report_shm_state_population(")
        .expect("the diagnostic must still be called");
    let arg_end = body[call..]
        .find(')')
        .expect("the diagnostic call must be closed");
    let arg = &body[call..call + arg_end];
    assert!(
        arg.contains("reclaim") || arg.contains("converged"),
        "the diagnostic's report-only argument must carry the convergence gate, not just \
         the flag; argument was:\n{arg}"
    );
}

/// The two brace-matched arms of `if <cond> { .. } else { .. }`, located by
/// the CONDITION text.
///
/// Both arms are needed because an assertion over the whole body cannot see a
/// SWAP: `ScanMode::ReportOnly` and `ScanMode::Reclaim` both appear either
/// way, and so does `report_only`.
fn if_else_arms(body: &str, condition: &str) -> (String, String) {
    let needle = format!("if {condition}");
    let start = body
        .find(&needle)
        .unwrap_or_else(|| panic!("`{needle}` must exist in the body:\n{body}"));
    let then = block_after(body, start);
    let after_then = start + body[start..].find(&then).expect("then-block offset") + then.len();
    let else_at = body[after_then..]
        .find("else")
        .unwrap_or_else(|| panic!("`{needle}` must have an else arm:\n{body}"))
        + after_then;
    let otherwise = block_after(body, else_at);
    (then, otherwise)
}

/// The first brace-matched `{ .. }` at or after `from`.
fn block_after(src: &str, from: usize) -> String {
    let open = src[from..]
        .find('{')
        .unwrap_or_else(|| panic!("expected a block after byte {from}"))
        + from;
    let bytes = src.as_bytes();
    let mut depth = 0usize;
    for (i, b) in bytes.iter().enumerate().skip(open) {
        match b {
            b'{' => depth += 1,
            b'}' => {
                depth -= 1;
                if depth == 0 {
                    return src[open..=i].to_string();
                }
            }
            _ => {}
        }
    }
    panic!("unbalanced block after byte {from}");
}

#[test]
fn the_diagnostic_reclaims_only_when_the_operator_did_not_ask_for_a_report() {
    // `--report-only` is the look-first escape hatch, and the load-bearing
    // property is the MAPPING, not the presence of both names: a diagnostic
    // with its arms SWAPPED still mentions both modes and the flag, so it
    // would delete on the one run whose whole point was not to.
    let src = code_only(&read_main());
    let body = fn_body(&src, "report_shm_state_population");
    let (reclaim_arm, report_arm) = if_else_arms(&body, "mode.removes()");

    assert!(
        report_arm.contains("ScanMode::ReportOnly") && !report_arm.contains("ScanMode::Reclaim"),
        "the `--report-only` arm must choose ReportOnly and nothing else; arm was:\n{report_arm}"
    );
    assert!(
        reclaim_arm.contains("ScanMode::Reclaim") && !reclaim_arm.contains("ScanMode::ReportOnly"),
        "the default arm must choose Reclaim and nothing else; arm was:\n{reclaim_arm}"
    );
    // The budgets ride the same fork, so a swap there is the same class of
    // defect: a report that spends the remedy's budget, or a remedy bounded
    // like a report.
    assert!(
        report_arm.contains("SHM_STATE_REPORT_BUDGET")
            && !report_arm.contains("SHM_STATE_RECLAIM_BUDGET"),
        "the report arm must take the report budget; arm was:\n{report_arm}"
    );
    assert!(
        reclaim_arm.contains("SHM_STATE_RECLAIM_BUDGET")
            && !reclaim_arm.contains("SHM_STATE_REPORT_BUDGET"),
        "the reclaim arm must take the reclaim budget; arm was:\n{reclaim_arm}"
    );
}

#[test]
fn the_walk_reads_the_code_it_claims_to_read() {
    // ANTI-TAUTOLOGY: every assertion above is a `contains` over a stripped
    // view, so a stripper that emptied the file would make all of them
    // vacuous — except this one.
    let raw = read_main();
    let src = code_only(&raw);
    assert!(
        src.contains("fn clean_iceoryx2_state"),
        "the stripped view must still hold the function under test"
    );
    assert!(
        src.contains("fn report_shm_state_population"),
        "and the diagnostic it calls"
    );
    // And the stripper must really strip: `main.rs` is full of `//` comments,
    // so a no-op stripper leaves the view identical to the source.
    assert!(
        src.len() < raw.len(),
        "the stripped view must be shorter than the source it came from"
    );
}

// ---------------------------------------------------------------------------
// The per-node refusal listing.
//
// Behavioural where it can be: `render_refused_nodes` is PURE, so the whole
// listing is pinned against hand-written line vectors with no dead node in
// the global iceoryx2 namespace. The subprocess arms in
// `tests/trace_inspect_and_clean_cli_test.rs` plant their dead nodes under a
// PRIVATE registry root rather than the shared `iox2_` namespace, and they
// pin the converged shape (one node swept, nothing refused); minting a
// REFUSED node needs the leaked-loan shape, which
// `cerulion_cli_engine/tests/sweep_dead_nodes_test.rs` owns, so the
// listing's own wiring is pinned by the source walk below.
// ---------------------------------------------------------------------------

use cerulion_cli_engine::ipc_cleanup::{
    classify_cleanup_failures, sweep_one_node, DeadNodeIdentity, FailedNodeCleanup,
    NodeSweepOutcome, SurveyedNode, SweepMode,
};
use cerulion_core::iceoryx_logger::CapturedLog;
use iceoryx2::node::NodeCleanupFailure;
use iceoryx2::prelude::LogLevel;

/// The node token exactly as iceoryx2 0.9.1 renders it — kept verbatim so a
/// reader's grep of the listing against a raw trace matches.
const NODE_A: &str =
    "UniqueNodeId(UniqueSystemId { value: 1, pid: 4242, creation_time: Time { seconds: 7, nanoseconds: 0 } })";

fn refusal(node: &str, variant: &str, causes: &[&str]) -> FailedNodeCleanup {
    FailedNodeCleanup {
        node: node.to_string(),
        variant: variant.to_string(),
        causes: causes.iter().map(|c| c.to_string()).collect(),
    }
}

#[test]
fn a_refused_node_renders_its_variant_and_every_sub_cause_indented_beneath_it() {
    let failures = [refusal(
        NODE_A,
        "InternalError",
        &[
            "Unable to remove stale resources since the service tags could not be read due to an internal error.",
            "Unable to remove stale resources since the stale resources of the port 9 could not be removed due to an internal failure.",
        ],
    )];
    let expected = vec![
        "Refused nodes:".to_string(),
        format!("  node {NODE_A}: InternalError"),
        "    Unable to remove stale resources since the service tags could not be read due to an internal error.".to_string(),
        "    Unable to remove stale resources since the stale resources of the port 9 could not be removed due to an internal failure.".to_string(),
    ];
    assert_eq!(super::render_refused_nodes(&failures), expected);
}

#[test]
fn a_converged_sweep_renders_no_listing_at_all() {
    // Not even the header: `cerulion clean` prints the listing
    // unconditionally after the breakdown, so an empty input must be silent.
    assert_eq!(super::render_refused_nodes(&[]), Vec::<String>::new());
}

#[test]
fn absences_are_stated_never_fabricated() {
    // A refusal line with no parenthesised variant, and a node iceoryx2 gave
    // no captured explanation for: each renders a STATEMENT of the absence,
    // and the no-explanation arm is the one place the trace hint survives.
    let failures = [refusal(NODE_A, "", &[])];
    let lines = super::render_refused_nodes(&failures);
    assert_eq!(lines[1], format!("  node {NODE_A}: (variant not reported)"));
    assert_eq!(
        lines.len(),
        3,
        "header + node + the absence line:\n{lines:?}"
    );
    assert!(
        lines[2].starts_with("    (no sub-cause captured"),
        "the absence must be stated:\n{lines:?}"
    );
    assert!(
        lines[2].contains("RUST_LOG=iceoryx2=trace"),
        "the hint must name the env var that actually gates the raw lines for this verb \
         (the sweep already pins iceoryx2 at Trace, so `IOX2_LOG_LEVEL` is inert):\n{lines:?}"
    );
    // A node WITH causes carries no hint — the listing IS the explanation.
    let explained = [refusal(NODE_A, "InternalError", &["x since the y."])];
    assert!(
        !super::render_refused_nodes(&explained)
            .iter()
            .any(|l| l.contains("RUST_LOG")),
        "an explained node must not carry the trace hint"
    );
}

#[test]
fn the_listing_is_capped_and_says_how_many_it_folded() {
    let cap = super::NODES_SHOWN;
    let node = |i: usize| {
        format!("UniqueNodeId(UniqueSystemId {{ value: {i}, pid: 1, creation_time: 0 }})")
    };
    let make = |n: usize| -> Vec<FailedNodeCleanup> {
        (0..n)
            .map(|i| refusal(&node(i), "InternalError", &["c since the d."]))
            .collect()
    };

    // cap + 2 ⇒ exactly `cap` nodes in full (the FIRST `cap`, in order), then
    // one fold line naming the remainder; nothing from the folded nodes.
    let lines = super::render_refused_nodes(&make(cap + 2));
    let node_lines: Vec<&String> = lines.iter().filter(|l| l.starts_with("  node ")).collect();
    assert_eq!(node_lines.len(), cap, "{lines:?}");
    for (i, l) in node_lines.iter().enumerate() {
        assert_eq!(**l, format!("  node {}: InternalError", node(i)));
    }
    assert_eq!(lines.last().unwrap(), "  … and 2 more", "{lines:?}");
    assert!(
        !lines.iter().any(|l| l.contains(&node(cap))),
        "a folded node must not leak past the cap:\n{lines:?}"
    );

    // Exactly `cap` ⇒ every node shown and NO fold line (`… and 0 more`
    // would be a lie about a remainder that does not exist).
    let lines = super::render_refused_nodes(&make(cap));
    assert_eq!(
        lines.iter().filter(|l| l.starts_with("  node ")).count(),
        cap
    );
    assert!(!lines.iter().any(|l| l.contains("more")), "{lines:?}");

    // cap + 1 ⇒ the boundary on the other side: exactly one folded.
    let lines = super::render_refused_nodes(&make(cap + 1));
    assert_eq!(lines.last().unwrap(), "  … and 1 more", "{lines:?}");
}

#[test]
fn the_cap_is_ten_nodes() {
    // The item's contract, pinned so a drift is a deliberate change.
    assert_eq!(super::NODES_SHOWN, 10);
}

#[test]
fn the_clean_verb_prints_the_listing_between_the_breakdown_and_the_unclassified_arm() {
    // The inert-shipping guard: every arm above stays green with the renderer
    // unwired. Ordering matters for the reader — the breakdown's
    // internal-error remediation points DOWN at the listing, so the
    // listing must follow it, and the unclassified hint closes the report.
    let src = code_only(&read_main());
    // The printing lives in `report_sweep`, the one reporter
    // `clean_dead_nodes` goes through.
    let body = fn_body(&src, "report_sweep");
    let breakdown = body
        .find("Failure breakdown:")
        .expect("the per-cause breakdown must still be printed");
    let listing = body
        .find("render_refused_nodes(")
        .expect("`cerulion clean` must render the refused-nodes per-node listing");
    let unclassified = body
        .find("Unclassified failures")
        .expect("the unclassified arm must still be printed");
    assert!(
        breakdown < listing && listing < unclassified,
        "breakdown, then listing, then unclassified; body was:\n{body}"
    );
    // The internal-error remediation must not send the operator to a
    // trace flag the listing makes unnecessary — such a flag is also
    // inert for this verb (the sweep pins iceoryx2 at Trace itself).
    let internal = body
        .find("\"iceoryx2 internal error\" =>")
        .expect("the internal-error arm must still exist");
    let arm_end = body[internal..].find("}").expect("arm closes") + internal;
    let arm = &body[internal..arm_end];
    assert!(
        !arm.contains("IOX2_LOG_LEVEL") && !arm.contains("RUST_LOG"),
        "the internal-error remediation must point at the listing, not a trace flag; arm was:\n{arm}"
    );
    assert!(
        arm.contains("refused nodes"),
        "the internal-error remediation must point the reader at the listing; arm was:\n{arm}"
    );
    // The verb makes ONE sweep and reclaims nothing. This line survived the
    // reclaim's removal once, still telling the operator that a directory of
    // orphan port tags "is reclaimed by this verb right after this listing",
    // so the pin is on the promise rather than on the wording around it: a
    // remediation may not offer a pass the code no longer has.
    for promise in ["reclaim", "right after this listing", "sweeps once more"] {
        assert!(
            !arm.contains(promise),
            "the internal-error remediation says `{promise}`, promising work this verb no \
             longer does; it makes one sweep. arm was:\n{arm}"
        );
    }
    // The unclassified arm keeps its hint — through the one shared constant.
    let tail = &body[unclassified..];
    assert!(
        tail.contains("RAW_IOX2_LINES_HINT"),
        "the unclassified arm must keep the raw-lines hint; tail was:\n{tail}"
    );
}

/// The paren-balanced argument list of the FIRST `<callee>(` in `body`,
/// whitespace-normalised — so an assertion about WHAT a renderer is handed
/// cannot be satisfied by the call merely existing.
fn call_args(body: &str, callee: &str) -> String {
    let needle = format!("{callee}(");
    let at = body
        .find(&needle)
        .unwrap_or_else(|| panic!("`{needle}` must be called; body was:\n{body}"));
    let open = at + callee.len();
    let bytes = body.as_bytes();
    let mut depth = 0usize;
    for (i, b) in bytes.iter().enumerate().skip(open) {
        match b {
            b'(' => depth += 1,
            b')' => {
                depth -= 1;
                if depth == 0 {
                    let args = body[open + 1..i]
                        .split_whitespace()
                        .collect::<Vec<_>>()
                        .join(" ");
                    // rustfmt leaves a trailing comma on a wrapped argument list.
                    return args.strip_suffix(',').unwrap_or(&args).to_string();
                }
            }
            _ => {}
        }
    }
    panic!("`{needle}` is not paren-balanced; body was:\n{body}");
}

#[test]
fn each_renderer_is_handed_the_reports_own_field_never_an_empty_slice() {
    // The walks above pin that each renderer is CALLED. That is not the
    // wiring: `render_refused_nodes(&[])` keeps every call in place, keeps
    // every hand oracle on the PURE renderer green, and prints an empty
    // listing for every refused node — the inert-shipping shape one level
    // down. The ARGUMENT is pinned by VALUE: the sweep report's own field,
    // and the report is the reporter's one parameter (a shadowing
    // `let report = …` inside the body would make `&report.failures` a
    // local's, so the body may bind no `report` of its own).
    let src = code_only(&read_main());
    let printing = fn_body(&src, "report_sweep");
    assert_eq!(
        call_args(&printing, "render_refused_nodes"),
        "&report.failures",
        "the per-node listing must render the sweep's refusals; body was:\n{printing}"
    );
    assert_eq!(
        call_args(&printing, "render_registry_errors"),
        "&report.registry_errors",
        "the registry block must render the sweep's registry errors; body was:\n{printing}"
    );
    assert!(
        !printing.contains("let report"),
        "`report` must be the parameter, never re-bound; body was:\n{printing}"
    );
    let sig_start = src
        .find("fn report_sweep")
        .expect("the reporter must exist");
    let sig = &src[sig_start..sig_start + src[sig_start..].find('{').expect("body")];
    let sig = sig.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        sig.contains("(report: &cerulion_cli_engine::ipc_cleanup::CleanupReport, mode: SweepMode)"),
        "the reporter takes the sweep report as `report` and the verb's mode; signature \
         was:\n{sig}"
    );
}

/// Every convergence VALUE `clean_dead_nodes` hands back — the second half of
/// each `(Ok(()), <expr>)` tuple, whether behind a `return` or as the tail —
/// in source order, whitespace-normalised.
fn convergence_positions(body: &str) -> Vec<String> {
    body.match_indices("Ok(()),")
        .map(|(at, _)| {
            // The tuple's own `(` is the last non-whitespace byte before
            // `Ok(()),` — rustfmt puts a newline between them when the tail
            // does not fit on one line.
            let open = body[..at]
                .rfind(|c: char| !c.is_whitespace())
                .expect("a tuple opens before `Ok(()),`");
            assert_eq!(
                &body[open..=open],
                "(",
                "every `Ok(()),` sits inside a tuple; body was:\n{body}"
            );
            let bytes = body.as_bytes();
            let mut depth = 0usize;
            for (i, b) in bytes.iter().enumerate().skip(open) {
                match b {
                    b'(' => depth += 1,
                    b')' => {
                        depth -= 1;
                        if depth == 0 {
                            let expr = body[open + 1..i]
                                .trim_start()
                                .strip_prefix("Ok(()),")
                                .expect("the tuple opens with `Ok(()),`")
                                .trim_end();
                            // rustfmt leaves a trailing comma on the last
                            // element of a multi-line tuple.
                            let expr = expr.strip_suffix(',').unwrap_or(expr);
                            return expr.split_whitespace().collect::<Vec<_>>().join(" ");
                        }
                    }
                    _ => {}
                }
            }
            panic!("a `(Ok(()),` tuple is not paren-balanced; body was:\n{body}");
        })
        .collect()
}

#[test]
fn every_convergence_the_verb_hands_back_is_computed_never_a_literal_true() {
    // The `.shm_state` reclamation is gated on this value, so a literal `true`
    // anywhere in the verb would reclaim on a wedged desk: a 0/0 sweep whose
    // registry could not be scanned reports nothing refused and has reached
    // nothing. Every convergence position is enumerated rather than `find`-ed,
    // because a `find` never looks at the positions it did not name.
    let src = code_only(&read_main());
    let body = fn_body(&src, "clean_dead_nodes");
    assert_eq!(
        convergence_positions(&body),
        vec!["converged".to_string()],
        "the verb hands back exactly one convergence, the computed conjunction; body was:\n{body}"
    );
    // And that conjunction is the FULL one: the failure count alone would call
    // a scan failure with nothing refused converged.
    let normalized = body.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        normalized.contains(
            "let converged = report.failed_cleanups == 0 && report.registry_errors.is_empty();"
        ),
        "convergence is the failure count AND an empty registry-error list; body was:\n{body}"
    );
}

/// The cleaner-lock family end to end — capture → classifier → listing. Two
/// sweeps racing over one dead node (every `graph run` startup sweeps, and
/// `try_cleanup_dead_nodes` blocks for ZERO time) refuse with
/// `AnotherInstanceIsCleaningUpTheNode`, and iceoryx2's explanation reads
/// `since another instance …` / `since an interrupt …` (`mod.rs:598/758/762/766`),
/// not `since the …`. A predicate that only knew `since the` dropped those
/// lines, and the listing then sent the operator to `RUST_LOG=iceoryx2=trace`
/// for a line the tool had already captured. Each row must render its
/// verbatim explanation beneath the node and NO trace hint.
#[test]
fn a_cleaner_lock_refusal_renders_its_captured_explanation_not_the_trace_hint() {
    let rows: [(&str, &str); 4] = [
        (
            "Unable to remove stale resources since another instance is already cleaning up the dead nodes resources.",
            "AnotherInstanceIsCleaningUpTheNode",
        ),
        (
            "Unable to acquire monitor cleaner since another instance is already cleaning up all resources.",
            "AnotherInstanceIsCleaningUpTheNode",
        ),
        (
            "Unable to acquire monitor cleaner since another instance has already cleaned up all resources.",
            "ResourcesAlreadyCleanedUp",
        ),
        (
            "Unable to acquire monitor cleaner since an interrupt signal was received.",
            "Interrupt",
        ),
    ];
    let sweep_origin = "Node::<iceoryx2::service::ipc_threadsafe::Service>::cleanup_dead_nodes()";
    for (message, variant) in rows {
        let captured = vec![
            CapturedLog {
                level: LogLevel::Debug,
                origin: sweep_origin.to_string(),
                message: format!("Dead node ({NODE_A}) detected"),
            },
            CapturedLog {
                level: LogLevel::Debug,
                origin: format!("DeadNodeView(AliveNodeView {{ id: {NODE_A}, details: None, _service: PhantomData<iceoryx2::service::ipc_threadsafe::Service> }})"),
                message: message.to_string(),
            },
            CapturedLog {
                level: LogLevel::Trace,
                origin: sweep_origin.to_string(),
                message: format!("Unable to remove dead node {NODE_A} ({variant})."),
            },
        ];

        let lines = super::render_refused_nodes(&classify_cleanup_failures(&captured).failures);

        assert_eq!(
            lines,
            vec![
                "Refused nodes:".to_string(),
                format!("  node {NODE_A}: {variant}"),
                format!("    {message}"),
            ],
            "{variant}: the captured explanation must be the listing, with no trace hint"
        );
        assert!(
            !lines.iter().any(|l| l.contains("RUST_LOG")),
            "{variant}: the hint must not be printed for a line the tool already holds"
        );
    }
}

// ---------------------------------------------------------------------------
// The registry-wide block.
//
// iceoryx2 also logs lines about the registry WALK itself — the full-scan
// failure (`iceoryx2-0.9.1/src/node/mod.rs:1242`), `Node::list`'s own lines
// (`:1138`, `:1145`) and `list_all_nodes` (`:1261`, `:1265`) — which name no
// node. The classifier files them under `registry_errors`; the verb must print
// them under their own heading BEFORE any per-node listing, so a global
// failure reads as global rather than as the cause of whichever node is
// listed next.
// ---------------------------------------------------------------------------

const FULL_SCAN_LINE: &str =
    "Unable to perform a full scan for dead nodes since the all existing nodes could not be listed (InsufficientPermissions).";
const LIST_LINE: &str =
    "Unable to iterate over Node list since the node list could not be acquired (InsufficientPermissions).";
const LIST_ALL_NODES_LINE: &str =
    "Unable to list all nodes due to insufficient permissions while listing all nodes.";

#[test]
fn registry_wide_errors_render_under_their_own_heading_verbatim_and_in_order() {
    // The realistic 0.9.1 chain for a registry that cannot be listed, in
    // emission order (`list_all_nodes` → `Node::list` → the sweep entry).
    let errors = [
        LIST_ALL_NODES_LINE.to_string(),
        LIST_LINE.to_string(),
        FULL_SCAN_LINE.to_string(),
    ];
    let expected = vec![
        "registry-wide sweep errors:".to_string(),
        format!("  {LIST_ALL_NODES_LINE}"),
        format!("  {LIST_LINE}"),
        format!("  {FULL_SCAN_LINE}"),
        "  (the registry itself could not be fully scanned — not any node's fault; a dead node \
         the sweep never reached may still be registered and is counted nowhere above)"
            .to_string(),
    ];
    assert_eq!(super::render_registry_errors(&errors), expected);
}

#[test]
fn a_fully_scanned_registry_renders_no_registry_block_at_all() {
    // Not even the heading: the block is printed unconditionally at the top
    // of the verb's report, so an empty input must be silent.
    assert_eq!(super::render_registry_errors(&[]), Vec::<String>::new());
}

/// The scenario end to end — capture → classifier → BOTH renderers. The
/// full-scan line sits immediately before a refusal, token-free and
/// sub-cause-shaped; the per-node listing must NOT carry it beneath the
/// node, and the registry block must carry it verbatim. Hand oracle on the
/// exact lines of each.
#[test]
fn a_registry_wide_line_before_a_refusal_renders_in_the_registry_block_not_under_the_node() {
    let sweep_origin = "Node::<iceoryx2::service::ipc_threadsafe::Service>::cleanup_dead_nodes()";
    let own = "Unable to remove stale resources since the port tags could not be read due to an internal error.";
    let captured = vec![
        CapturedLog {
            level: LogLevel::Debug,
            origin: sweep_origin.to_string(),
            message: FULL_SCAN_LINE.to_string(),
        },
        CapturedLog {
            level: LogLevel::Debug,
            origin: sweep_origin.to_string(),
            message: format!("Dead node ({NODE_A}) detected"),
        },
        CapturedLog {
            level: LogLevel::Debug,
            origin: format!("DeadNodeView(AliveNodeView {{ id: {NODE_A}, details: None, _service: PhantomData<iceoryx2::service::ipc_threadsafe::Service> }})"),
            message: own.to_string(),
        },
        CapturedLog {
            level: LogLevel::Trace,
            origin: sweep_origin.to_string(),
            message: format!("Unable to remove dead node {NODE_A} (InternalError)."),
        },
    ];

    let parts = classify_cleanup_failures(&captured);

    assert_eq!(
        super::render_refused_nodes(&parts.failures),
        vec![
            "Refused nodes:".to_string(),
            format!("  node {NODE_A}: InternalError"),
            format!("    {own}"),
        ],
        "the node's listing carries exactly its own explanation — never the scan failure"
    );
    assert_eq!(
        super::render_registry_errors(&parts.registry_errors),
        vec![
            "registry-wide sweep errors:".to_string(),
            format!("  {FULL_SCAN_LINE}"),
            "  (the registry itself could not be fully scanned — not any node's fault; a dead node \
             the sweep never reached may still be registered and is counted nowhere above)"
                .to_string(),
        ],
        "the scan failure renders as global"
    );
}

/// A capture exercising every attribution arm at once: two interleaved nodes
/// (the id arm), an ownerless service-layer line (the adjacency arm), a
/// registry-wide line (arm 0), a foreign-token line (withheld), a
/// cleaner-lock explanation (`since another …`), and a variant the table
/// does not classify (`Interrupt`, into `unclassified`).
const NODE_B: &str =
    "UniqueNodeId(UniqueSystemId { value: 2, pid: 2020, creation_time: Time { seconds: 8, nanoseconds: 0 } })";
const NODE_C: &str =
    "UniqueNodeId(UniqueSystemId { value: 3, pid: 3030, creation_time: Time { seconds: 9, nanoseconds: 0 } })";
/// A node the walk VISITED and never refused (an inaccessible neighbour):
/// its line names it, so it is nobody's in the listing.
const NODE_D: &str =
    "UniqueNodeId(UniqueSystemId { value: 4, pid: 4040, creation_time: Time { seconds: 10, nanoseconds: 0 } })";
const OWNERLESS_LINE: &str =
    "Unable to unlink the port resources since the process does not have sufficient permissions.";
const A_OWN_LINE: &str =
    "Unable to remove stale resources since the service tags could not be read due to an internal error.";
const B_LOCK_LINE: &str =
    "Unable to acquire monitor cleaner since another instance is already cleaning up all resources.";

fn every_arm_capture() -> Vec<CapturedLog> {
    let sweep_origin = "Node::<iceoryx2::service::ipc_threadsafe::Service>::cleanup_dead_nodes()";
    let (a, b, c, d) = (NODE_A, NODE_B, NODE_C, NODE_D);
    let view = |node: &str| {
        format!("DeadNodeView(AliveNodeView {{ id: {node}, details: None, _service: PhantomData<iceoryx2::service::ipc_threadsafe::Service> }})")
    };
    let l = |origin: String, message: String| CapturedLog {
        level: LogLevel::Debug,
        origin,
        message,
    };
    vec![
        l(sweep_origin.to_string(), FULL_SCAN_LINE.to_string()),
        l(
            sweep_origin.to_string(),
            format!("Dead node ({a}) detected"),
        ),
        l(
            "Service::remove_node".to_string(),
            OWNERLESS_LINE.to_string(),
        ),
        l(
            sweep_origin.to_string(),
            format!("Dead node ({b}) detected"),
        ),
        l(view(a), A_OWN_LINE.to_string()),
        l(
            format!("Node::get_node_state(Config {{ .. }}, {d})"),
            "Unable to acquire node monitor due to insufficient permissions.".to_string(),
        ),
        l(view(b), B_LOCK_LINE.to_string()),
        l(
            sweep_origin.to_string(),
            format!("Unable to remove dead node {a} (InternalError)."),
        ),
        l(
            sweep_origin.to_string(),
            format!("Unable to remove dead node {b} (AnotherInstanceIsCleaningUpTheNode)."),
        ),
        l(
            sweep_origin.to_string(),
            format!("Unable to remove dead node {c} (Interrupt)."),
        ),
    ]
}

#[test]
fn classification_is_bit_identical_across_runs_and_matches_its_oracle() {
    // Principle #7 for a diagnostic: the same capture must classify the same
    // way every time, or two `cerulion clean` runs over one wedged desk would
    // name different culprits. The two-run compare is anchored to a hand
    // oracle first (the crib, `local_harvest`'s determinism arm, is a bare
    // self-compare — a classifier that deterministically produced the WRONG
    // parts would pass it).
    let captured = every_arm_capture();
    let run1 = classify_cleanup_failures(&captured);
    assert_eq!(
        run1.failures
            .iter()
            .map(|f| (f.variant.as_str(), f.causes.len()))
            .collect::<Vec<_>>(),
        vec![
            ("InternalError", 2),
            ("AnotherInstanceIsCleaningUpTheNode", 1),
            ("Interrupt", 0),
        ],
        "the oracle: A takes the ownerless line and its own, B its cleaner-lock explanation, \
         C nothing (the only line near it names the visited node D): {:?}",
        run1.failures
    );
    assert_eq!(run1.registry_errors, vec![FULL_SCAN_LINE.to_string()]);
    assert_eq!(run1.unclassified.len(), 1, "{:?}", run1.unclassified);
    assert_eq!(
        run1.failures_by_cause,
        std::collections::BTreeMap::from([
            ("iceoryx2 internal error".to_string(), 1),
            ("lock contention".to_string(), 1),
        ])
    );

    let run2 = classify_cleanup_failures(&captured);
    assert_eq!(
        run1, run2,
        "two runs over one capture must be bit-identical"
    );
}

#[test]
fn the_renderers_are_bit_identical_across_runs_and_match_their_oracles() {
    // The three PURE renderers, each run twice over the same input and
    // anchored to the hand oracles the arms above establish.
    let parts = classify_cleanup_failures(&every_arm_capture());
    let refused1 = super::render_refused_nodes(&parts.failures);
    let refused2 = super::render_refused_nodes(&parts.failures);
    assert_eq!(
        refused1,
        vec![
            "Refused nodes:".to_string(),
            format!("  node {NODE_A}: InternalError"),
            format!("    {OWNERLESS_LINE}"),
            format!("    {A_OWN_LINE}"),
            format!("  node {NODE_B}: AnotherInstanceIsCleaningUpTheNode"),
            format!("    {B_LOCK_LINE}"),
            format!("  node {NODE_C}: Interrupt"),
            // C's only nearby line names the visited node D, so C carries the
            // stated absence — never D's line.
            format!(
                "    (no sub-cause captured — {})",
                super::RAW_IOX2_LINES_HINT
            ),
        ],
        "the listing, line for line"
    );
    assert_eq!(refused1, refused2);

    let registry1 = super::render_registry_errors(&parts.registry_errors);
    let registry2 = super::render_registry_errors(&parts.registry_errors);
    assert_eq!(registry1[1], format!("  {FULL_SCAN_LINE}"));
    assert_eq!(registry1, registry2);
}

#[test]
fn the_clean_verb_prints_the_registry_block_first_and_gates_convergence_on_it() {
    // The inert-shipping guard for the block, plus the two ORDERING facts
    // that make it mean anything: it must come BEFORE the per-node listing
    // (a global failure read after a node listing reads as an afterthought)
    // and BEFORE the early "nothing to clean" return — the 0/0 counters of a
    // scan that listed nothing are otherwise indistinguishable from a clean
    // registry, and that return would skip the block entirely.
    let src = code_only(&read_main());
    // The printing lives in `report_sweep`, the one reporter `clean_dead_nodes`
    // goes through, so the ordering facts are read there;
    // the convergence conjunction is the verb's own and is read below.
    let printing = fn_body(&src, "report_sweep");
    let body = fn_body(&src, "clean_dead_nodes");
    let registry = printing
        .find("render_registry_errors(")
        .expect("`cerulion clean` must render the registry-wide block");
    let nothing_to_clean = printing
        .find("No dead iceoryx2 nodes found")
        .expect("the converged early return must still exist");
    let listing = printing
        .find("render_refused_nodes(")
        .expect("the per-node listing must still be rendered");
    assert!(
        registry < nothing_to_clean && registry < listing,
        "the registry block must precede both the early return and the listing; body was:\n{printing}"
    );
    // The sweep prints through it, exactly once: a second call would print one
    // report twice, and zero would leave the block unwired.
    assert_eq!(
        body.matches("report_sweep(").count(),
        1,
        "the sweep must print through `report_sweep`; body was:\n{body}"
    );
    // CONVERGENCE: a registry the walk could not cover leaves dead nodes
    // registered that appear in no counter, so the `.shm_state` reclamation
    // must stand down for it exactly as for a refused node. Pinned as the
    // whole conjunction — `failed_cleanups == 0` alone does not satisfy the pin.
    let normalized = body.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        normalized.contains(
            "let converged = report.failed_cleanups == 0 && report.registry_errors.is_empty();"
        ),
        "convergence must require an empty registry-error set; body was:\n{body}"
    );
    // And the "nothing to clean" claim is reserved for a registry that WAS
    // scanned: the early-return arm forks on the registry errors, and the
    // MAPPING is what matters — swapped arms would claim "nothing to clean"
    // on exactly the run that scanned nothing.
    let (scanned, unscanned) = if_else_arms(&printing, "report.registry_errors.is_empty()");
    assert!(
        scanned.contains("No dead iceoryx2 nodes found") && !scanned.contains("not evidence"),
        "the fully-scanned arm alone may claim nothing to clean; arm was:\n{scanned}"
    );
    assert!(
        unscanned.contains("not evidence") && !unscanned.contains("No dead iceoryx2 nodes found"),
        "the unscanned arm must refuse the absence claim; arm was:\n{unscanned}"
    );
}

// ---------------------------------------------------------------------------
// `--report-only`: the whole report, and the proof nothing was removed.
//
// The listing renderer is PURE, so the whole would-remove report is pinned
// against hand-written line vectors over a fixture-planted registry. The
// "removed nothing" half is pinned at the SEAM the production sweep removes
// through: `sweep_one_node` takes the removal as a parameter, so a counting
// closure observes the same call the real path makes — no filesystem mock,
// and no registry needed.
// ---------------------------------------------------------------------------

/// The registry directory a fixture report names back. A path, never a
/// directory that exists: the renderer only prints it.
const FIXTURE_NODE_DIR: &str = "/fixture/iox2-root/nodes";

/// One planted dead node, named the two ways the walk names one: the entry
/// name under the registry directory, and iceoryx2's `{:?}` token.
fn planted(value: u128, pid: u32) -> DeadNodeIdentity {
    DeadNodeIdentity {
        name: value.to_string(),
        token: format!(
            "UniqueNodeId(UniqueSystemId {{ value: {value}, pid: {pid}, creation_time: Time {{ \
             seconds: 7, nanoseconds: 0 }} }})"
        ),
    }
}

fn would_remove(dead: &[DeadNodeIdentity]) -> Vec<String> {
    super::render_would_remove(dead, std::path::Path::new(FIXTURE_NODE_DIR))
}

#[test]
fn the_report_only_listing_names_every_planted_node_and_claims_no_removal() {
    let dead = [planted(1, 4242), planted(2, 4243)];
    assert_eq!(
        would_remove(&dead),
        vec![
            format!(
                "Dead iceoryx2 node(s) `cerulion clean` would sweep: 2 under \
                 {FIXTURE_NODE_DIR} (report only; none was removed)"
            ),
            "  node 1".to_string(),
            "  node 2".to_string(),
            super::REMOVAL_CAN_STILL_BE_REFUSED.to_string(),
        ],
        "the whole listing, line for line"
    );
    // The listing names what the sweep ATTEMPTS. A refusal is raised from
    // inside the removal, so a report cannot know which of these come off,
    // and the heading must not say "would remove" of the whole set.
    assert!(
        !would_remove(&dead)[0].contains("would remove"),
        "the heading must not promise a removal: {:?}",
        would_remove(&dead)[0]
    );
    assert!(
        super::REMOVAL_CAN_STILL_BE_REFUSED.contains("not a promise each one comes off"),
        "the caveat must say the sweep can still refuse: {}",
        super::REMOVAL_CAN_STILL_BE_REFUSED
    );
    // No orphan port-tag section any more: iceoryx2 0.10 sweeps a dead port's
    // tag with the rest of its stale resources, so the shape the reclaim healed
    // cannot arise and the listing has nothing to say about it.
    assert!(
        !would_remove(&dead)
            .iter()
            .any(|line| line.contains("port tag")),
        "the listing must not mention orphan port tags; the reclaim is gone: {:?}",
        would_remove(&dead)
    );
    // And no line may claim a removal happened.
    assert!(
        !would_remove(&dead)
            .iter()
            .any(|line| line.contains("Cleaned") || line.contains("removed 1")),
        "a report must never render a removal"
    );
}

#[test]
fn a_registry_with_no_dead_node_renders_no_would_remove_listing_at_all() {
    // Not even the heading: `report_sweep` falls through to the
    // "nothing to clean" summary, which is the line that arm already pins.
    assert_eq!(would_remove(&[]), Vec::<String>::new());
}

#[test]
fn the_would_remove_listing_folds_at_the_cap_and_at_the_cap_plus_one() {
    // Both sides of the bound, with the fixtures COMPUTED from the constant.
    let cap = super::NODES_SHOWN;
    let plant =
        |n: usize| -> Vec<DeadNodeIdentity> { (0..n).map(|i| planted(i as u128, 1)).collect() };

    let at = would_remove(&plant(cap));
    assert_eq!(
        at.iter().filter(|l| l.starts_with("  node ")).count(),
        cap,
        "{at:?}"
    );
    assert!(
        !at.iter().any(|l| l.contains("more")),
        "at the cap there is no remainder to fold: {at:?}"
    );

    let over = would_remove(&plant(cap + 1));
    assert_eq!(
        over.iter().filter(|l| l.starts_with("  node ")).count(),
        cap,
        "{over:?}"
    );
    assert_eq!(
        over[over.len() - 2],
        "  … and 1 more",
        "one past the cap folds exactly one, above the closing caveat: {over:?}"
    );
    assert!(
        !over.iter().any(|l| **l == format!("  node {cap}")),
        "a folded node must not leak past the cap: {over:?}"
    );
    // The heading counts what was FOUND, never what was listed.
    assert!(
        over[0].contains(&format!(": {} under ", cap + 1)),
        "the heading counts every dead node, not the shown ones: {over:?}"
    );
}

/// Drive [`sweep_one_node`] over a hand-built survey and report what the
/// seam saw: the outcomes, how many times the removal was called, and every
/// line the sweep traced.
fn drive(
    nodes: Vec<SurveyedNode<Result<(), NodeCleanupFailure>>>,
    mode: SweepMode,
) -> (Vec<NodeSweepOutcome>, usize, Vec<(LogLevel, String)>) {
    let mut calls = 0usize;
    let mut traced = Vec::new();
    let outcomes = nodes
        .into_iter()
        .map(|node| {
            sweep_one_node(
                node,
                mode,
                |result: Result<(), NodeCleanupFailure>| {
                    calls += 1;
                    result
                },
                |level, message| traced.push((level, message)),
            )
        })
        .collect();
    (outcomes, calls, traced)
}

#[test]
fn a_report_only_sweep_classifies_every_dead_node_and_never_calls_the_removal() {
    let (a, b) = (planted(1, 4242), planted(2, 4243));
    let survey = vec![
        SurveyedNode::Dead {
            identity: a.clone(),
            view: Ok(()),
        },
        SurveyedNode::NotDead,
        SurveyedNode::Dead {
            identity: b.clone(),
            view: Ok(()),
        },
    ];

    let (outcomes, calls, traced) = drive(survey, SweepMode::ReportOnly);

    assert_eq!(
        outcomes,
        vec![
            NodeSweepOutcome::WouldRemove(a.clone()),
            NodeSweepOutcome::NotDead,
            NodeSweepOutcome::WouldRemove(b.clone()),
        ],
        "every dead node is classified, and no live one is"
    );
    assert_eq!(
        calls, 0,
        "a report must not call the removal, not once, for any node"
    );
    // Only the detection lines, transcribed from iceoryx2 0.9.1's own sweep
    // loop (`node/mod.rs:1225`): a report emits no removal line, because
    // there was no removal to report.
    assert_eq!(
        traced,
        vec![
            (LogLevel::Debug, format!("Dead node ({}) detected", a.token)),
            (LogLevel::Debug, format!("Dead node ({}) detected", b.token)),
        ]
    );
}

#[test]
fn a_removing_sweep_is_the_same_classification_followed_by_the_removal() {
    // The other side of the ONE decision: the same survey, the same
    // classification, and the removal called exactly once per dead node.
    // Oracles for both trace lines transcribed from iceoryx2 0.9.1
    // (`node/mod.rs:1227` and `:1231`).
    let (a, b) = (planted(1, 4242), planted(2, 4243));
    let survey = vec![
        SurveyedNode::Dead {
            identity: a.clone(),
            view: Ok(()),
        },
        SurveyedNode::NotDead,
        SurveyedNode::Dead {
            identity: b.clone(),
            view: Err(NodeCleanupFailure::InsufficientPermissions),
        },
    ];

    let (outcomes, calls, traced) = drive(survey, SweepMode::Remove);

    assert_eq!(
        outcomes,
        vec![
            NodeSweepOutcome::Removed(a.clone()),
            NodeSweepOutcome::NotDead,
            NodeSweepOutcome::Refused {
                identity: b.clone(),
                failure: "InsufficientPermissions".to_string(),
            },
        ]
    );
    assert_eq!(calls, 2, "one call per DEAD node, and none for a live one");
    assert_eq!(
        traced,
        vec![
            (LogLevel::Debug, format!("Dead node ({}) detected", a.token)),
            (
                LogLevel::Trace,
                format!("The dead node ({}) was successfully removed.", a.token)
            ),
            (LogLevel::Debug, format!("Dead node ({}) detected", b.token)),
            (
                LogLevel::Trace,
                format!(
                    "Unable to remove dead node {} (InsufficientPermissions).",
                    b.token
                )
            ),
        ],
        "the refusal line is the marker `classify_cleanup_failures` parses"
    );
}

#[test]
fn the_clean_verb_renders_the_would_remove_listing_from_the_sweeps_own_field() {
    // The inert-shipping guard for the report-only half: every arm above
    // stays green with the renderer unwired, and a `render_would_remove(&[])`
    // would keep the call in place and print nothing on exactly the run the
    // flag was given for.
    let src = code_only(&read_main());
    let body = fn_body(&src, "report_sweep");
    assert_eq!(
        call_args(&body, "render_would_remove"),
        "&report.dead_nodes, &cerulion_cli_engine::shm_state::iceoryx2_node_dir()",
        "the listing must render the sweep's own dead-node set, under the registry the sweep \
         read; body was:\n{body}"
    );
    // The fork sits AFTER the registry-wide block (a global failure is still
    // read as global) and BEFORE the summary line, whose 0/0 counters would
    // otherwise render a report as a converged sweep.
    let registry = body
        .find("render_registry_errors(")
        .expect("the registry block must still be rendered");
    let fork = body
        .find("if mode.reports_only() && !report.dead_nodes.is_empty() {")
        .expect("the report-only fork must guard the listing");
    let summary = body
        .find("Cleaned {} dead iceoryx2 node(s)")
        .expect("the summary line must still exist");
    assert!(
        registry < fork && fork < summary,
        "registry block, then the report-only fork, then the summary; body was:\n{body}"
    );
    // And the fork RETURNS: a report that fell through would print
    // "No dead iceoryx2 nodes found — nothing to clean." under a listing of
    // the nodes it had just named.
    let block = block_after(&body, fork);
    assert!(
        block.contains("return;"),
        "the report-only fork must return rather than fall through to the summary; block \
         was:\n{block}"
    );
}

#[test]
fn the_verb_maps_the_flag_to_the_mode_in_both_directions() {
    // The ONE mapping no arm in this repository can catch behaviourally in both
    // directions, so it is pinned by VALUE here.
    //
    // An inverted `--report-only` is caught loudly: the report-only arm in
    // `tests/trace_inspect_and_clean_cli_test.rs` would watch its planted node
    // vanish and fail on the registry diff, which is exactly how the
    // flag-read-after-the-sweep mutant dies. The OTHER direction is caught by
    // NOTHING. A bare `cerulion clean` that silently became a report leaves no
    // trace in this suite: no CLI arm may run the verb bare (its `/tmp`
    // reclaim is machine wide), and the engine arms call
    // `sweep_dead_nodes_with_config` directly, so the dispatch would be the
    // only broken thing and every test would stay green while `cerulion clean`
    // stopped cleaning.
    let src = code_only(&read_main());
    let at = src
        .find("Commands::Clean { report_only } =>")
        .expect("the clean verb must still be dispatched on its own flag");
    assert_eq!(
        call_args(&src[at..], "clean_iceoryx2_state"),
        "if report_only { SweepMode::ReportOnly } else { SweepMode::Remove }",
        "the flag maps to the mode in exactly this direction; a swap would make \
         `--report-only` destructive or `clean` inert, and only the first of those \
         is caught anywhere else; dispatch was:\n{}",
        &src[at..at + 200.min(src.len() - at)]
    );
}

#[test]
fn the_verb_closes_a_report_only_run_by_saying_nothing_was_removed() {
    // The flag's whole promise, stated once, after BOTH halves have
    // reported — so it covers the `.shm_state` population too, which the
    // dead-node listing says nothing about.
    let src = code_only(&read_main());
    let body = fn_body(&src, "clean_iceoryx2_state");
    let diagnostic = body
        .find("report_shm_state_population(")
        .expect("the diagnostic must still be called");
    let closing = body
        .find("NOTHING_WAS_REMOVED")
        .expect("a report-only run must close by saying nothing was removed");
    assert!(
        diagnostic < closing,
        "the closing claim comes after both halves have reported; body was:\n{body}"
    );
    let guard = body
        .find("if mode.reports_only() {")
        .expect("the closing line is printed only under the flag");
    assert!(
        guard < closing,
        "the closing line is guarded by the mode; body was:\n{body}"
    );
    assert!(
        super::NOTHING_WAS_REMOVED.contains("nothing was removed"),
        "the sentence must say it: {}",
        super::NOTHING_WAS_REMOVED
    );
}