cerulion_cli 0.1.0

// SPDX-License-Identifier: AGPL-3.0-only
//! Clap CLI definitions.

use std::path::PathBuf;

use cerulion_cli_engine::graph_cmd::PeerLossPolicy as EnginePeerLossPolicy;
use cerulion_cli_engine::graph_cmd::TimeSource as EngineTimeSource;
use clap::{Parser, Subcommand, ValueEnum};
// CER-925: `add = ArgValueCandidates::new(..)` attaches a live value source to
// an arg; `ArgValueCompleter` + `PathCompleter` do the same for paths where the
// extension is a hard contract. Both lower to clap's `Arg::add` extension seam
// (`clap/unstable-ext`).
use clap_complete::engine::{ArgValueCandidates, ArgValueCompleter, PathCompleter};

use crate::completion;

/// Clap-parseable mirror of the engine's [`EngineTimeSource`] (CER-222
/// chunk 4a). The engine crate does not depend on `clap`, so the
/// `ValueEnum` lives here and maps 1:1 to the canonical engine enum.
/// Value names are lowercased by clap's default `ValueEnum` renaming:
/// `real`, `external`, `virtual`.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
pub enum TimeSource {
    /// Live, event-driven loop on `RealClock` (the default).
    Real,
    /// Externally-mastered time (`ExternalClock`) — inert today (CER-354/CER-424).
    External,
    /// Deterministic `VirtualClock` + 1 ms poll loop (replay / bench).
    Virtual,
}

impl From<TimeSource> for EngineTimeSource {
    fn from(ts: TimeSource) -> Self {
        match ts {
            TimeSource::Real => EngineTimeSource::Real,
            TimeSource::External => EngineTimeSource::External,
            TimeSource::Virtual => EngineTimeSource::Virtual,
        }
    }
}

/// CER-349 chunk E item 2: the `--record-env` capture policy — the clap
/// `ValueEnum` mirror of the canonical engine enum
/// (`cerulion_cli_engine::graph_cmd::RecordEnvMode`), mapped 1:1 like
/// [`TimeSource`]. Value names lowercase: `allowlist` / `all` / `none`.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
pub enum RecordEnv {
    /// Names always; values only for CERULION_*, RUST_LOG, IOX2_* — the rest
    /// redacted to an FNV-1a 64 divergence hash. The default.
    Allowlist,
    /// Verbatim values (secrets included — warned loudly at record time).
    All,
    /// Names only (replay env-fidelity off).
    None,
}

impl From<RecordEnv> for cerulion_cli_engine::graph_cmd::RecordEnvMode {
    fn from(re: RecordEnv) -> Self {
        use cerulion_cli_engine::graph_cmd::RecordEnvMode;
        match re {
            RecordEnv::Allowlist => RecordEnvMode::Allowlist,
            RecordEnv::All => RecordEnvMode::All,
            RecordEnv::None => RecordEnvMode::None,
        }
    }
}

/// CER-349 chunk E addendum: the explicit `--record-cpu` VALUE — a core id or
/// the literal `none` (float). Flag ABSENT = the engine's `RecordCpu::Auto`
/// (see [`record_cpu_mode`]).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RecordCpuArg {
    /// Pin bagd to this core id.
    Core(u32),
    /// Do not pin — bagd floats.
    None,
}

/// clap value parser for `--record-cpu`: a `u32` core id or the literal
/// `none`. Anything else is a loud parse error naming both accepted forms.
fn parse_record_cpu(s: &str) -> Result<RecordCpuArg, String> {
    if s.eq_ignore_ascii_case("none") {
        return Ok(RecordCpuArg::None);
    }
    s.parse::<u32>()
        .map(RecordCpuArg::Core)
        .map_err(|_| format!("expected a core id (e.g. `--record-cpu=7`) or `none`, got `{s}`"))
}

/// Map the optional CLI value to the engine's [`RecordCpu`] — flag absent is
/// AUTO (the box-measured default: pin bagd to the highest core on >=4-core
/// Linux boxes; float on capacity-bound small-core boxes).
///
/// [`RecordCpu`]: cerulion_cli_engine::graph_cmd::RecordCpu
pub fn record_cpu_mode(arg: Option<RecordCpuArg>) -> cerulion_cli_engine::graph_cmd::RecordCpu {
    use cerulion_cli_engine::graph_cmd::RecordCpu;
    match arg {
        Option::None => RecordCpu::Auto,
        Some(RecordCpuArg::Core(n)) => RecordCpu::Core(n),
        Some(RecordCpuArg::None) => RecordCpu::None,
    }
}

/// Clap-parseable mirror of the engine's [`EnginePeerLossPolicy`] (CER-508
/// chunks 6+7) — the multi-process worker-death policy. Same mirror pattern as
/// [`TimeSource`]: the engine crate does not depend on `clap`, so the
/// `ValueEnum` lives here and maps 1:1. Value names lowercase: `continue`,
/// `fail`.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
pub enum PeerLoss {
    /// A crashed worker is dropped from the shared barrier; survivors keep
    /// running degraded (loud). The default.
    Continue,
    /// Any worker death stops the whole deployment (fail-loud, deterministic).
    Fail,
}

impl From<PeerLoss> for EnginePeerLossPolicy {
    fn from(p: PeerLoss) -> Self {
        match p {
            PeerLoss::Continue => EnginePeerLossPolicy::Continue,
            PeerLoss::Fail => EnginePeerLossPolicy::Fail,
        }
    }
}

#[derive(Parser)]
#[command(name = "cerulion", about = "Cerulion robotics middleware CLI", version)]
pub struct Cli {
    /// Enable verbose (debug-level) logging (`-v` or `--verbose`)
    #[arg(global = true, short = 'v', long)]
    pub verbose: bool,

    #[command(subcommand)]
    pub command: Commands,
}

/// The `cerulion completions` long help, verbatim.
///
/// A const rather than a doc comment because the zsh install line contains
/// `$+functions[compdef]` — see the variant's doc comment for why every other
/// spelling either breaks the docs gate or corrupts the text a user copies.
/// Written as a literal so what `--help` prints is exactly what is here, with
/// no clap re-wrapping of the install table.
const COMPLETIONS_LONG_ABOUT: &str = "\
Print the shell code that enables `cerulion` tab-completion (CER-925).

Completes everything: subcommands, flags, and enum values from the command
tree, plus LIVE names \u{2014} topic names for `topic echo/info/hz`, node types
for `node build/run/info/...`, graph names for `graph run/levels/partition/...`,
workspace + built-in ROS 2 schemas for `schema info`, and robot names for
`viz --robot`.

Install it once, per shell:

  zsh         echo 'autoload -Uz compinit && (( $+functions[compdef] )) || compinit' >> ~/.zshrc
              echo 'source <(COMPLETE=zsh cerulion)' >> ~/.zshrc
  bash        echo 'source <(COMPLETE=bash cerulion)' >> ~/.bashrc
  fish        echo 'COMPLETE=fish cerulion | source' > ~/.config/fish/completions/cerulion.fish
  elvish      echo 'eval (E:COMPLETE=elvish cerulion | slurp)' >> ~/.elvish/rc.elv
  powershell  echo '$env:COMPLETE = \"powershell\"; cerulion | Out-String | Invoke-Expression; Remove-Item Env:\\COMPLETE' >> $PROFILE

zsh needs BOTH lines: the completion script ends in `compdef`, which is a
function `compinit` defines, and macOS's system zshrc never calls it \u{2014}
without the guard a stock ~/.zshrc prints `command not found: compdef` at every
shell start and nothing completes. The guard is a no-op if a framework
(oh-my-zsh, prezto) already ran compinit.

These forms re-generate the shell code on demand, so they self-correct when
`cerulion` is upgraded or moved \u{2014} the reason they are recommended.
`cerulion completions <shell> > FILE` writes the same code to a file instead;
re-run it after an upgrade.

Live names are read from LOCAL sources only (the workspace, the iceoryx2
service directory, ~/.cerulion) under a hard 150 ms budget, so a TAB press
never opens the network and never hangs.

A REMOTE robot's topics complete only while something HOLDS a mirror of them on
this desk. `cerulion viz --robot NAME` leaves one standing (the vizd daemon
keeps the demand); `topic echo/info/hz` holds one only for as long as that
command runs; `topic list` demands nothing at all. With no such attach, remote
topics do not complete \u{2014} learning their names needs a network round-trip,
which a TAB press must not do.";

#[derive(Subcommand)]
pub enum Commands {
    /// Workspace management
    Workspace {
        #[command(subcommand)]
        action: WorkspaceAction,
    },
    /// Node management
    Node {
        #[command(subcommand)]
        action: NodeAction,
    },
    /// Graph management
    Graph {
        #[command(subcommand)]
        action: GraphAction,
    },
    /// Topic introspection
    Topic {
        #[command(subcommand)]
        action: TopicAction,
    },
    /// Schema management
    Schema {
        #[command(subcommand)]
        action: SchemaAction,
    },
    /// REMOVED (founder ruling): `cerulion ros attach` is now
    /// `cerulion ros2 attach` — attach was the `ros` family's only verb, so
    /// the whole family folded into `ros2` (run / launch / attach). This stub
    /// exists so the OLD spelling fails LOUDLY with the migration message
    /// (`main` intercepts it above the login gate) instead of clap's generic
    /// "unrecognized subcommand". Hidden from help and completions; no alias.
    ///
    /// `disable_help_flag` is load-bearing: clap answers `--help`/`-h` BEFORE
    /// returning a parsed command, so without it `cerulion ros --help` served
    /// clap's help page for the stub (exit 0) instead of the migration error.
    /// With the auto flag gone, `--help` is just another hyphen value the
    /// trailing positional swallows, and EVERY old spelling — help flags
    /// included — reaches the exit-2 migration message. (`cerulion help ros`
    /// still renders this doc comment, which itself names the new spelling.)
    #[command(hide = true, disable_help_flag = true)]
    Ros {
        /// Swallows whatever followed (`attach --iface …`, `--help` included)
        /// so the user sees the migration error, never a clap parse error
        /// about the old verb's own flags.
        #[arg(
            trailing_var_arg = true,
            allow_hyphen_values = true,
            hide = true,
            value_name = "ARGS"
        )]
        args: Vec<String>,
    },
    /// ROS 2 interop family: run stock ROS 2 entry points on Cerulion
    /// transport (`RMW_IMPLEMENTATION=rmw_cerulion`) — swap the command, not
    /// the stack — and attach to a live ROS 2 / DDS robot (`ros2 attach`).
    Ros2 {
        #[command(subcommand)]
        action: Ros2Action,
    },
    /// Visualize live Cerulion topics in a Rerun viewer via the long-lived
    /// `cerulion-vizd` daemon. `cerulion viz` is a thin CLIENT: it ensures the
    /// daemon is running (spawning it detached if absent), ensures a Rerun viewer is
    /// connected to the daemon's advertised proxy, and sends one `attach` per topic
    /// over the daemon's control socket. The daemon owns the runtime taps, the
    /// schema-generic decode/render, and the hosted Rerun proxy — the CLI never
    /// links `rerun`. Zero YAML, zero hand-written nodes, zero scripts.
    ///
    /// Each `TOPIC` is an absolute Cerulion topic (a missing leading `/` is added).
    /// Optionally pin a schema with `TOPIC=SCHEMA` (e.g.
    /// `/go2/cloud=sensor_msgs/PointCloud2`); a local topic's schema otherwise
    /// back-fills from the first frame. The daemon HOSTS a Rerun gRPC proxy by
    /// default and advertises its URL; this verb connects a viewer to it (found +
    /// SPAWNED automatically, or `--no-spawn` to connect one yourself). The spawned
    /// viewer OUTLIVES the run (stays open after Ctrl+C). `$CERULION_RERUN_URL` is a
    /// DAEMON-side knob (read once when the daemon starts) that makes the daemon a
    /// CLIENT of an external endpoint instead of hosting one — it is inert against an
    /// already-running daemon. Without `--detach` the verb stays attached until Ctrl+C,
    /// then detaches ONLY the topics it added (the shared daemon keeps running). Run
    /// with NO arguments to attach every decodable local topic the daemon discovers
    /// (silent / undecodable topics are skipped).
    Viz {
        /// Cerulion topics to visualize, each `TOPIC` or `TOPIC=SCHEMA`.
        // CER-925: completes the bare `TOPIC` form. The `TOPIC=SCHEMA` pin is
        // not completed past the `=` — a schema source here would have to know
        // which topic precedes it, and the pin exists precisely for the case
        // where the type is NOT locally resolvable.
        #[arg(value_name = "TOPIC", add = ArgValueCandidates::new(completion::topics))]
        topics: Vec<String>,
        /// Visualize a REMOTE robot's topics over the network (CER-825): the robot
        /// NAME is passed to the daemon's `attach`, which declares gateway INGRESS
        /// for each topic (re-injecting the robot's frames into desk-local SHM) and
        /// taps the local mirror — the robot serves SHM bytes and nothing else. On a
        /// typical LAN the robot's gateway is found by scouting with ZERO config, so
        /// no schema pin is needed: the daemon resolves each topic's ROS type from
        /// the robot's served CATALOG (CER-823) and fetches any custom type's schema
        /// over the network. You MAY still pin `TOPIC=SCHEMA` (e.g.
        /// `/utlidar/cloud=sensor_msgs/PointCloud2`) to skip the catalog lookup. A
        /// topic the robot genuinely does not serve is a hard ERROR, not a silent
        /// skip.
        #[arg(long, value_name = "NAME", add = ArgValueCandidates::new(completion::robot_names))]
        robot: Option<String>,
        /// Additional zenoh gateway locator(s) to CONNECT to for a robot scouting
        /// can't reach, e.g. `tcp/192.168.123.99:7683`. Repeatable. Requires
        /// `--robot`. Threaded into the daemon's network config when this command
        /// STARTS the daemon (a daemon already running keeps its own boot-time
        /// locators — scouting covers the LAN regardless; the verb notes that).
        #[arg(long, value_name = "LOCATOR", requires = "robot")]
        connect: Vec<String>,
        /// Local zenoh locator(s) to LISTEN on for remote discovery/ingress, e.g.
        /// `tcp/0.0.0.0:7447`. Repeatable. Requires `--robot`. Threaded into the
        /// daemon's network config when this command STARTS the daemon (a daemon
        /// already running keeps its own boot-time locators).
        #[arg(long, value_name = "LOCATOR", requires = "robot")]
        listen: Vec<String>,
        /// Never spawn a viewer — the daemon hosts the Rerun proxy; connect a
        /// viewer to it yourself (the hint prints `rerun <url>`).
        #[arg(long)]
        no_spawn: bool,
        /// Spawn the viewer in web-server mode (`rerun --serve-web`) connected to
        /// the daemon's hosted proxy, and print the browser URL, instead of a
        /// native window. Applies ONLY when the verb spawns a viewer — ignored
        /// with `--no-spawn` or when no `rerun` binary is found.
        #[arg(long)]
        serve_web: bool,
        /// Attach the topics to the long-lived `cerulion-vizd` daemon and RETURN
        /// immediately, leaving the daemon + viewer running (the Studio handoff).
        /// Without it, `cerulion viz` stays attached until Ctrl+C, then detaches
        /// the topics it added (the daemon keeps running for the next run).
        #[arg(long)]
        detach: bool,
    },
    /// Connect to a REMOTE robot's wire plane (CER-824): dial its iroh endpoint,
    /// demand topics, and re-inject each into desk-local SHM so `topic echo` /
    /// `viz` see them as LOCAL topics ("remote = local"). Spawns the
    /// `cerulion-connectd` sibling binary (it links iroh; the `cerulion` CLI stays
    /// iroh-free — build it with `cargo build -p cerulion_connectd`).
    Connect {
        /// The robot: a 64-char-hex endpoint id, OR a name pinned in
        /// `~/.cerulion/robots.toml` (`[robots]` table). `cerulion pair <robot>`
        /// writes that pin, so after pairing this name just works. Omit when
        /// using `--eid`.
        // CER-925: the pins in `robots.toml` are one of `robot_names`' three
        // sources, so a paired robot completes here by construction.
        #[arg(value_name = "ROBOT", add = ArgValueCandidates::new(completion::robot_names))]
        robot: Option<String>,
        /// The robot's iroh endpoint id (64-char hex). Overrides a positional
        /// name. From `cerulion pair`, the robot's mDNS TXT `eid=`, or the operator.
        #[arg(long, value_name = "HEX")]
        eid: Option<String>,
        /// A direct `ip:port` for the robot (LAN direct-dial). Repeatable. When
        /// omitted, the eid is resolved via relay/discovery.
        #[arg(long = "addr", value_name = "IP:PORT")]
        addrs: Vec<String>,
        /// A topic to demand + re-inject, e.g. `/utlidar/cloud`. Repeatable. With
        /// no `--topic` and no `--all`, the catalog is printed and NOTHING is
        /// demanded (the discoverable default).
        #[arg(long = "topic", value_name = "TOPIC")]
        topics: Vec<String>,
        /// Demand + re-inject EVERY topic the robot's catalog lists.
        #[arg(long)]
        all: bool,
        /// The desk device key file (32 raw bytes — an ed25519 seed). When
        /// omitted, the paired desk key `~/.cerulion/desk.key` is used if it
        /// exists (written by `cerulion pair`), else an EPHEMERAL key — which an
        /// un-paired robot REFUSES, so pair the desk first with `cerulion pair`.
        #[arg(long, value_name = "PATH", value_hint = clap::ValueHint::FilePath)]
        key_file: Option<PathBuf>,
        /// Materialize fetched `.msg`/YAML schemas into this dir so `topic echo` /
        /// `viz` can decode a never-seen type. When omitted, materialization is
        /// skipped (built-in types still decode).
        #[arg(long, value_name = "DIR", value_hint = clap::ValueHint::DirPath)]
        schemas_dir: Option<PathBuf>,
        /// Self-hosted relay URL. Default: n0 public relays.
        #[arg(long, value_name = "URL")]
        relay_url: Option<String>,
        /// Disable all relays (LAN / direct-dial only).
        #[arg(long)]
        relay_disabled: bool,
        /// Kill-switch: pass `off` to refuse to connect.
        #[arg(long)]
        network: Option<String>,
    },
    /// Pair THIS desk with a robot (CER-834): run the CPace code-pairing ceremony
    /// over the robot's ops plane so it access-lists this desk's device key, then
    /// pin `name → eid` into `~/.cerulion/robots.toml` so `cerulion connect
    /// <robot>` works afterwards with NO flags. Spawns the `cerulion-connectd`
    /// sibling (it links iroh; the `cerulion` CLI stays iroh-free — build it with
    /// `cargo build -p cerulion_connectd`).
    ///
    /// The robot owner starts code pairing on the robot and shares a short code;
    /// you supply it here (`--code`, else typed at the prompt / piped on stdin).
    /// The desk device key is created at `~/.cerulion/desk.key` on the first pair
    /// (0600 perms) and reused afterwards — an existing key is NEVER overwritten.
    ///
    /// STDOUT emits deterministic, machine-parseable state lines for a Studio
    /// driver: `pairing: robot=<name> eid=<hex>`, then on success `paired:
    /// robot=<name> eid=<hex> account=<hex>` and `pinned: name=<name> eid=<hex>`.
    /// STDERR carries structured logs + the interactive code prompt. The exit code
    /// is the primary signal: 0 = paired · 1 = usage/config · 2 = robot refused
    /// (no armed code / denied / expired window) · 3 = unreachable (dial/timeout) ·
    /// 4 = wrong pairing code / attempts exhausted · 130 = interrupted (Ctrl-C /
    /// SIGTERM during the ceremony — no pairing occurred).
    Pair {
        /// The robot: a 64-char-hex endpoint id, OR a name resolved via the
        /// robot's mDNS TXT `eid=` record, else `~/.cerulion/robots.toml`. Omit
        /// when using `--eid`.
        // CER-925: a robot seen on this LAN lands in the peer cache, so the
        // FIRST pairing completes too — not only a re-pair of a pinned robot.
        #[arg(value_name = "ROBOT", add = ArgValueCandidates::new(completion::robot_names))]
        robot: Option<String>,
        /// The robot's iroh endpoint id (64-char hex). Overrides a positional
        /// name (no name is pinned). From the robot operator or its mDNS `eid=`.
        #[arg(long, value_name = "HEX")]
        eid: Option<String>,
        /// A direct `ip:port` for the robot (LAN direct-dial). Repeatable. When
        /// omitted, the eid is resolved via relay/discovery.
        #[arg(long = "addr", value_name = "IP:PORT")]
        addrs: Vec<String>,
        /// The short pairing code the robot owner shared. PREFER omitting this and
        /// typing the code at the prompt (or piping it on stdin for a Studio
        /// driver) — a code passed via `--code` is VISIBLE in process listings
        /// (`ps`) for the ceremony window, and it authorizes durable enrollment.
        #[arg(long, value_name = "CODE")]
        code: Option<String>,
        /// The account (64-char hex) this pairing is FOR. When omitted, a
        /// self-account is derived from the desk device key (the keypair-only
        /// desk with no cloud/Studio account).
        #[arg(long, value_name = "HEX")]
        account: Option<String>,
        /// The access-list label the robot stores on its pairing row (so the
        /// owner recognizes this desk). Default: this machine's hostname.
        #[arg(long, value_name = "NAME")]
        label: Option<String>,
        /// The desk device key file (32 raw bytes — an ed25519 seed). Default:
        /// `~/.cerulion/desk.key` (created 0600 on the first pair, reused after).
        #[arg(long, value_name = "PATH", value_hint = clap::ValueHint::AnyPath)]
        key_file: Option<PathBuf>,
        /// Self-hosted relay URL. Default: n0 public relays.
        #[arg(long, value_name = "URL")]
        relay_url: Option<String>,
        /// Disable all relays (LAN / direct-dial only).
        #[arg(long)]
        relay_disabled: bool,
    },
    /// Sign in to a Cerulion account (CER-844 A1). Runs the RFC 8628 device-code
    /// flow: prints a short code + a verification URL to copy into a browser on
    /// any device (headless-friendly, like Claude Code's CLI login), then waits
    /// for you to authorize. Use it any time to re-authenticate or switch
    /// accounts; the first identity-needing command auto-triggers it.
    Login,
    /// Manage your Cerulion account (CER-844 A6). Currently: your devices
    /// (`cerulion account devices list` / `... revoke <device_id>`) — self-service
    /// management of your own desks. Robot access management lives in Studio / the
    /// web account page (founder ruling), not the CLI.
    Account {
        #[command(subcommand)]
        action: AccountAction,
    },
    /// Interactive TUI dashboard
    Tui,
    /// Inspect publish-trace bag files written by Regime A
    /// (`cerulion_core::trace::BagWriter`). Reads JSON-Lines records
    /// from a directory and prints a human-readable timeline.
    Trace {
        #[command(subcommand)]
        action: TraceAction,
    },
    // CER-1257: there is no `Replay` variant. CER-982 (founder ruling 65)
    // removed the verb's BEHAVIOUR and left a hidden variant behind to diagnose
    // the removal; the founder ruled on 2026-08-11 that the diagnosis goes too,
    // so `cerulion replay` is now clap's plain `unrecognized subcommand`
    // (exit 2 — the code the tombstone already returned). Pinned by
    // `removed_verb_tests::removed_replay_verb_is_an_unknown_subcommand`.
    /// Bag as a DATA SOURCE (CER-939) — record topics into a bag, inspect one,
    /// and play one back onto local shared memory.
    ///
    /// `bag play` executes NOTHING: it re-publishes the recorded frames verbatim
    /// under their recorded topic names, wall-paced, so the desk-side shell
    /// (`topic list`/`echo`/`hz`, `cerulion viz`, Studio) sees what a live robot
    /// would publish. To RE-EXECUTE a bagged graph against your workspace's
    /// current node builds, add `--resim all`; add `--verify` on top of that to
    /// byte-compare the result against the recording.
    ///
    /// `bag record` produces such a bag, and records LOCAL topics only — run it
    /// on the machine that produces the data (the robot), then transfer the
    /// file. Together: record on the robot, play on your desk.
    Bag {
        #[command(subcommand)]
        action: BagAction,
    },
    /// Print the shell code that enables `cerulion` tab-completion.
    ///
    /// The full help (per-shell install lines + the guarantees) is
    /// [`COMPLETIONS_LONG_ABOUT`], kept OUT of this doc comment on purpose:
    /// the zsh line contains `$+functions[compdef]`, and rustdoc reads
    /// `[compdef]` as an intra-doc link (the `RUSTDOCFLAGS=-D warnings` gate
    /// fails). Escaping the brackets would satisfy rustdoc while putting
    /// literal backslashes into the `--help` text a user copies, and indenting
    /// it into a code block makes rustdoc try to compile it as Rust. An
    /// explicit `long_about` is the only form that keeps the help EXACT.
    #[command(long_about = COMPLETIONS_LONG_ABOUT)]
    Completions {
        /// Shell to emit completion code for.
        shell: completion::CompletionShell,
    },
    /// Capture the moment: write a Flashback covering the last ~30 seconds
    /// plus the next ~15, from the rolling window every serving graph holds.
    ///
    /// Nobody has to arm anything in advance — the window is always on. The
    /// capture lands in `recordings/flashbacks/` and this command waits for it,
    /// so you are handed a file rather than a promise. Turn the whole plane off
    /// with `CERULION_FLASHBACK=off`.
    #[cfg(unix)]
    Flashback {
        /// A note recorded into the capture, so a bag found three weeks later
        /// says what it was about.
        #[arg(long, value_name = "TEXT")]
        note: Option<String>,
        /// Exclude this capture from retention eviction, so it cannot rotate
        /// away.
        #[arg(long)]
        pin: bool,
        /// Return as soon as the capture is accepted, without waiting for the
        /// bag. The accepted line already carries the path it will have.
        #[arg(long)]
        no_wait: bool,
    },
    /// Capture the moment (unsupported on this platform — the recorder that
    /// holds the rolling window is Unix-only).
    #[cfg(not(unix))]
    Flashback,
    /// Wipe lingering iceoryx2 SHM state (services, nodes, event files).
    ///
    /// Useful when the topic topology changes between runs — iceoryx2
    /// services persist on `/tmp/iceoryx2/` until the OS reboots, and a
    /// graph that no longer matches the cached service config can fail
    /// to publish or look like it's missing topics.
    ///
    /// CER-1411: also reports the `/tmp/*.shm_state` population (a
    /// macOS/FreeBSD-only leak that makes every dead-node sweep walk a huge
    /// directory) and reclaims the entries whose creating process is provably
    /// gone.
    Clean {
        /// Report the `.shm_state` population and what is reclaimable, but
        /// delete nothing. The dead-node sweep still runs.
        #[arg(long)]
        report_only: bool,
    },
    /// Standalone zero-copy recorder: taps live topics + scheduler-trace
    /// SHM rings into a standard MCAP bag (CER-349). ATTACH-MODE,
    /// OBSERVABILITY-GRADE by default: attaches to already-running
    /// publishers and learns each channel's schema from the first
    /// message's wire header — schema name recorded as "unknown",
    /// wire_fixed_size 0 (enough for introspection + wire-hash gating,
    /// NOT the replay-grade descriptor). `cerulion graph run --record`
    /// drives the same recorder in EXACT (replay-grade) mode.
    ///
    /// BOXED (CER-1260 producer 3): `BagdArgs` is by far the widest variant here
    /// — it is the recorder's whole argv surface — and growing it by one
    /// `Option<String>` (`--run-id`) pushed the gap past `large_enum_variant`'s
    /// 200-byte threshold. Boxing is what clippy asks for and what the shape
    /// wants anyway: this enum is parsed ONCE per process and `Commands` is
    /// matched by every other verb, so making all of them carry the recorder's
    /// argv inline is the cost, and one allocation on the `bagd` path alone is
    /// the fix. `clap` implements `Args for Box<T>`, so the derive is unchanged.
    #[cfg(unix)]
    Bagd(Box<cerulion_bagd::BagdArgs>),
    /// Standalone zero-copy recorder (unsupported on this platform —
    /// bagd is Unix-only: it is driven by SIGTERM lifecycle signals).
    #[cfg(not(unix))]
    Bagd,
}

/// CER-836: the tracing-verbosity class of a verb, used to pick the default
/// `cerulion` log filter when `RUST_LOG` is unset.
///
/// - `OneShot` — run-and-exit introspection / management verbs (`topic *`,
///   `schema *`, `node list`/`info`/`create`/…, `graph levels`/`list`/…,
///   `workspace *`, `trace inspect`, `clean`). These default QUIET
///   (`cerulion=warn`) so their command output isn't interleaved with
///   lifecycle breadcrumbs.
/// - `LongRunning` — verbs that run a runtime loop / daemon (`graph run`,
///   `graph profile`, `node run`, `bag play`, `ros2 attach`, `viz`, `connect`,
///   and the hidden multi-process `run-worker`/`run-gateway` workers). These
///   keep the legacy `cerulion=info` default.
///
/// `-v/--verbose` raises EITHER class to `cerulion=debug`; an explicit
/// `RUST_LOG` always wins over the default (see `cerulion_core::init_logging`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum VerbLogClass {
    /// Run-and-exit verb — quiet (`warn`) default.
    OneShot,
    /// Runtime / daemon verb — `info` default.
    LongRunning,
}

impl VerbLogClass {
    /// True when this class defaults to the quiet `warn` filter — the
    /// `quiet_default` argument threaded into `cerulion_core::init_logging`.
    pub fn is_quiet_default(self) -> bool {
        matches!(self, VerbLogClass::OneShot)
    }
}

impl Commands {
    /// CER-836: classify this verb for the default tracing filter level.
    ///
    /// One-shot verbs run and exit without a runtime loop, so their lifecycle
    /// breadcrumbs are noise on top of the command's real output — they default
    /// QUIET. Long-running / runtime verbs (and the daemons) keep `info`.
    ///
    /// `Tui` and `Bagd` never drive this default (`main` skips `init_logging`
    /// for the TUI and dispatches `bagd` to its own logging init before
    /// `init_logging` runs), but they are classified here for exhaustiveness.
    pub fn log_verb_class(&self) -> VerbLogClass {
        use VerbLogClass::{LongRunning, OneShot};
        match self {
            // ── Long-running / runtime verbs: keep the `info` default ──
            Commands::Graph { action } => match action {
                // Runtime loops: the live run, the live profiler, and the
                // hidden multi-process worker / network-gateway processes.
                GraphAction::Run { .. }
                | GraphAction::Profile { .. }
                | GraphAction::RunWorker { .. }
                | GraphAction::RunGateway { .. } => LongRunning,
                // create / validate / list / levels / partition all run-and-exit.
                GraphAction::Create { .. }
                | GraphAction::Validate { .. }
                | GraphAction::List
                | GraphAction::Levels { .. }
                | GraphAction::Partition { .. } => OneShot,
            },
            // `node run` is a runtime loop; every other node verb runs-and-exits.
            Commands::Node { action } => match action {
                NodeAction::Run { .. } => LongRunning,
                NodeAction::Create { .. }
                | NodeAction::Delete { .. }
                | NodeAction::Modify { .. }
                | NodeAction::Build { .. }
                | NodeAction::Stage { .. }
                | NodeAction::List
                | NodeAction::Info { .. } => OneShot,
            },
            // `ros2 attach` runs a bridge graph; `viz` spawns/attaches the
            // vizd daemon; `connect` spawns connectd and stays attached;
            // `pair` drives the interactive pairing ceremony via connectd —
            // all runtime verbs.
            // CER-844 A1: `login` waits interactively for browser authorization
            // — keep `info` so any device-registration / auth warns are visible.
            // `ros2 run` exec()s the real `ros2` (a runtime process); the
            // classification covers only the pre-exec window (init_logging
            // runs before `main`'s intercept), where any staging warn should
            // stay visible — `info` default like the other runtime verbs.
            Commands::Ros2 { .. }
            | Commands::Viz { .. }
            | Commands::Connect { .. }
            | Commands::Pair { .. }
            | Commands::Login => LongRunning,
            // The removed-`ros`-family stub never runs anything: `main`
            // intercepts it above the login gate and prints the migration
            // error before `init_logging` matters. Classified for
            // exhaustiveness only.
            Commands::Ros { .. } => OneShot,
            // CER-939: `bag play` and `bag record` run a loop until the bag
            // ends / the duration elapses / Ctrl-C, and their lifecycle
            // breadcrumbs (topics refused, taps attached, schemas learned) are
            // the operator's only window into a run — `info` default. `bag
            // info` is a pure run-and-exit render, so it stays quiet.
            // CER-982: `bag play --resim` RE-EXECUTES nodes, which is the same
            // runtime-verb argument the removed `cerulion replay` made, and it
            // arrives through this same arm.
            // CER-1413: `bag migrate` is a run-and-exit rewrite whose product
            // is the preview it prints and the file it writes — `info` default,
            // like `bag info`.
            Commands::Bag { action } => match action {
                BagAction::Play { .. } | BagAction::Record { .. } => LongRunning,
                BagAction::Info { .. } | BagAction::Migrate { .. } => OneShot,
            },
            // ── One-shot verbs: QUIET (`warn`) default ──
            // CER-844 A6: `account devices list/revoke` run-and-exit.
            Commands::Account { .. }
            | Commands::Workspace { .. }
            | Commands::Topic { .. }
            | Commands::Schema { .. }
            | Commands::Trace { .. }
            // CER-925: prints a shell script to stdout and exits. It is also
            // the one verb whose stdout is EVALUATED by a shell, so a stray
            // lifecycle breadcrumb on stderr would at best confuse and at
            // worst be captured into a completion file.
            | Commands::Completions { .. }
            | Commands::Clean { .. } => OneShot,
            // CER-1260 PR 3: `flashback` publishes ONE request and then WAITS —
            // up to the post window plus the finalize grace — while a recorder
            // records and writes a bag. Its breadcrumbs (which recorders
            // answered, what they refused and why) are the operator's only
            // window into that wait, so it keeps the `info` default like every
            // other verb that stays attached to a running robot.
            #[cfg(unix)]
            Commands::Flashback { .. } => LongRunning,
            #[cfg(not(unix))]
            Commands::Flashback => OneShot,
            // Never drives the default (see the doc comment) — classified for
            // exhaustiveness so a new verb forces an explicit decision here.
            Commands::Tui => OneShot,
            #[cfg(unix)]
            Commands::Bagd(..) => OneShot,
            #[cfg(not(unix))]
            Commands::Bagd => OneShot,
        }
    }
}

/// `cerulion account <action>` (CER-844 A6).
#[derive(Subcommand)]
pub enum AccountAction {
    /// Manage the devices (desks) registered to your account.
    Devices {
        #[command(subcommand)]
        action: DevicesAction,
    },
}

/// `cerulion account devices <action>` (CER-844 A6).
#[derive(Subcommand)]
pub enum DevicesAction {
    /// List the devices registered to your account (id, kind, revoked state).
    List,
    /// Revoke one of your own devices — a lost or decommissioned desk. The device id
    /// comes from `cerulion account devices list`.
    Revoke {
        /// The device id to revoke.
        device_id: String,
    },
}

#[derive(Subcommand)]
pub enum TraceAction {
    /// Print a human-readable timeline of bag-file records in `dir`.
    ///
    /// Reads every `trace_*.jsonl` file in lexicographic order
    /// (matches BagWriter's sequential file-index naming) and prints
    /// one line per record, formatted as
    /// `<topic> seq=<N> t=<TIMESTAMP_NS> schema=<HASH>`. Use
    /// `--filter <topic>` to restrict to a single topic.
    Inspect {
        /// Directory containing `trace_*.jsonl` files.
        // CER-925 R1: same `String`-typed path class as `graph run --record`.
        #[arg(value_hint = clap::ValueHint::DirPath)]
        dir: String,
        /// Only show entries whose `topic` matches this string exactly.
        #[arg(short = 't', long)]
        filter: Option<String>,
        /// Limit the number of records printed (most recent first
        /// if `--reverse`, otherwise oldest-first).
        #[arg(short = 'n', long)]
        limit: Option<usize>,
        /// Print most-recent records first instead of oldest-first.
        #[arg(short = 'r', long)]
        reverse: bool,
    },
}

/// `cerulion ros2 <action>` — the ROS 2 interop family. `Run` and `Launch`
/// are VERBATIM pass-through wrappers of the native `ros2` verbs; `main`
/// intercepts them BEFORE clap parses (so every native token — hyphenated
/// flags included — is forwarded by construction), and those two variants
/// exist for `--help` and shell completion. `Attach` is a NATIVE clap verb
/// (formerly `cerulion ros attach`; the `ros` family folded into `ros2`):
/// clap parses it normally and `run()` dispatches it.
#[derive(Subcommand)]
pub enum Ros2Action {
    /// Run a ROS 2 executable on Cerulion transport: stages the child
    /// environment (`RMW_IMPLEMENTATION=rmw_cerulion`, an `LD_LIBRARY_PATH`
    /// prepend of the Cerulion lib dir, an `AMENT_PREFIX_PATH` prepend of a
    /// minimal ament prefix so ROS 2 can dlopen `librmw_cerulion.so`, and —
    /// when it ships beside the binary — an `LD_PRELOAD` of the Cerulion heap
    /// hook), then exec()s `ros2 run <ARGS...>` — replacing this process so
    /// stdio, signals, and the exit code flow through the kernel untouched.
    /// ALL arguments are forwarded to `ros2 run` verbatim — see
    /// `ros2 run --help` for its surface. Exit codes (pre-exec failures only;
    /// a successful exec inherits ros2's own): 69 librmw_cerulion.so missing,
    /// 127 ros2 not on PATH, 1 other. Unix-only.
    Run {
        /// Forwarded verbatim to `ros2 run` (e.g. `demo_nodes_cpp talker`).
        #[arg(
            trailing_var_arg = true,
            allow_hyphen_values = true,
            value_name = "ARGS"
        )]
        args: Vec<String>,
    },
    /// Launch a ROS 2 launch file on Cerulion transport: stages the same
    /// child environment as `ros2 run` (see above), then exec()s
    /// `ros2 launch <ARGS...>` — replacing this process so stdio, signals,
    /// and the exit code flow through the kernel untouched. ALL arguments are
    /// forwarded to `ros2 launch` verbatim — see `ros2 launch --help` for its
    /// surface. Exit codes (pre-exec failures only; a successful exec
    /// inherits ros2's own): 69 librmw_cerulion.so missing, 127 ros2 not on
    /// PATH, 1 other. Unix-only.
    Launch {
        /// Forwarded verbatim to `ros2 launch` (e.g.
        /// `demo.launch.py use_rviz:=false`, or the `pkg file.launch.py`
        /// package form).
        #[arg(
            trailing_var_arg = true,
            allow_hyphen_values = true,
            value_name = "ARGS"
        )]
        args: Vec<String>,
    },
    /// Discover a robot's DDS topics on `--iface`, generate the `dds_bridge`
    /// mapping config + a lean one-node bridge graph for the resolvable topics,
    /// and (after consent) run it. `--dry-run` prints the discovery report and
    /// stops.
    ///
    /// The generated graph bridges the robot's topics and nothing else — no
    /// visualization node is staged on the robot (CER-905). To SEE the data,
    /// run `cerulion viz --robot <name>` (or Cerulion Studio) on YOUR machine:
    /// the desk demands each topic, `cerulion-netd` mirrors it into desk-local
    /// shared memory, and `cerulion-vizd` decodes and renders it there. The
    /// robot only ships raw frames.
    Attach {
        /// The robot-LAN interface IP to run DDS discovery on (e.g.
        /// `192.168.123.18`). REQUIRED — restricts rustdds to this interface so
        /// CycloneDDS does not drop fragmented discovery data on a multi-homed
        /// host (the Go2 spike's load-bearing knob).
        #[arg(long, value_name = "IP")]
        iface: std::net::IpAddr,
        /// DDS domain id — must match the robot's `ROS_DOMAIN_ID`.
        #[arg(long, default_value_t = 0, value_name = "N")]
        domain: u16,
        /// Discovery collection window in seconds. Must be a finite number
        /// greater than 0 and at most 3600 (one hour) — rejected at parse time
        /// (CER-722 review fix 2 + pass-2; previously 0/negative/nan/inf were
        /// silently coerced to the default, and overflow-large finite values
        /// panicked at runtime).
        #[arg(
            long,
            default_value_t = 5.0,
            value_name = "SECS",
            value_parser = parse_attach_timeout
        )]
        timeout: f64,
        /// Print the discovery report and stop — write nothing, run nothing.
        /// Wins over `--yes`.
        #[arg(long)]
        dry_run: bool,
        /// Skip the interactive confirm and write + run non-interactively
        /// (scripts / CI / non-TTY).
        #[arg(long)]
        yes: bool,
        /// The generated graph name: writes `graphs/<name>.yaml` +
        /// `graphs/<name>.bridge.yaml`.
        #[arg(long, default_value = "attach", value_name = "NAME")]
        graph_name: String,
        /// Optional prefix prepended to each generated Cerulion topic name
        /// (default: mirror the ROS topic name).
        #[arg(long, value_name = "PREFIX")]
        topic_prefix: Option<String>,
        /// The robot-name namespace label written as the attach graph's
        /// `prefix:`. NOTE this is NOT the network announce identity — the name
        /// shown in `topic list` ROBOTS / mDNS `_cerulion._tcp` resolves from
        /// this machine's hostname (or the CERULION_ROBOT_IDENTITY env override)
        /// at runtime, independent of this flag. Default prefix: the robots'
        /// shared ROS namespace if the topics have one, else this machine's
        /// hostname. Set CERULION_ROBOT_IDENTITY to control the announced
        /// identity (e.g. for a remote attach or a stock-image fleet).
        #[arg(long, value_name = "NAME")]
        robot_name: Option<String>,
    },
}

/// CER-722 review fix 2 + pass-2: the maximum `ros2 attach --timeout`, in
/// seconds (one hour — already absurdly generous for a discovery window; the
/// spike-proven default is 5 s). The bound is honest UX AND the overflow
/// guard: `Duration::from_secs_f64` PANICS above ~5.8e11 s, so an unbounded
/// "finite positive" check (the pass-1 parser) still let `--timeout 2e19`
/// through to a runtime panic.
const MAX_ATTACH_TIMEOUT_SECS: f64 = 3600.0;

/// CER-722 review fix 2 + pass-2: `ros2 attach --timeout` must be a FINITE
/// number of seconds in `(0, 3600]`, rejected AT PARSE TIME with a clear
/// message. (`Duration::from_secs_f64` panics on nan/inf/negative AND on
/// overflow-large values; a 0-second or multi-hour discovery window is never
/// what a user meant — the pre-fix code silently coerced the non-finite
/// shapes to the default and let overflow-large values panic at runtime.)
fn parse_attach_timeout(s: &str) -> Result<f64, String> {
    let v: f64 = s
        .parse()
        .map_err(|e| format!("'{s}' is not a number: {e}"))?;
    if !v.is_finite() || v <= 0.0 || v > MAX_ATTACH_TIMEOUT_SECS {
        return Err(format!(
            "'{s}' is not a valid discovery window — --timeout must be a finite number of \
             seconds > 0 and <= {MAX_ATTACH_TIMEOUT_SECS} (one hour; e.g. 5)"
        ));
    }
    Ok(v)
}

#[derive(Subcommand)]
pub enum WorkspaceAction {
    /// Create a new workspace
    Create {
        /// Workspace name
        name: String,
    },
    /// Initialize a workspace at the current directory
    Init {
        /// Location (default: current directory)
        // CER-925 R2: `String`-typed AND named LOCATION, so it was missed by
        // BOTH the original `PathBuf` sweep and the R1 walk that keyed on the
        // value names PATH/DIR/FILE — the same class twice. The walk is now
        // an inverted declared-inventory guard so a name cannot hide an arg.
        #[arg(default_value = ".", value_hint = clap::ValueHint::DirPath)]
        location: String,
    },
}

#[derive(Subcommand)]
pub enum NodeAction {
    /// Create a new node type
    Create {
        /// Node type name
        // CER-925 R1: deliberately NO completer. A create argument names
        // something that does not exist yet, so the existing node types are
        // exactly the set this verb REJECTS (`NodeExists`). Offering them
        // completes a guaranteed error. The correct candidate set is empty.
        node_type: String,
        /// Add output port: SCHEMA NAME (both required)
        #[arg(short = 'o', long = "output", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        output: Vec<String>,
        /// Add input port: SCHEMA NAME (both required). For
        /// trigger inputs use `-T` instead.
        #[arg(short = 'i', long = "input", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        input: Vec<String>,
        /// Add a TRIGGER input: SCHEMA NAME. The new input is
        /// declared as `#[input(trigger)]` and the node's trigger
        /// policy is set to `data_trigger=<NAME>`. SCHEMA and NAME
        /// are BOTH required so that a multi-`-T` user error can't
        /// be silently misread as `-T SCHEMA1=NAME(SCHEMA2)`. At
        /// most one `-T` per node; for sync over multiple inputs
        /// use `-i` for each input plus
        /// `--policy sync_window_ms=N`.
        #[arg(short = 'T', long = "trigger-input", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        trigger_input: Vec<String>,
        /// Trigger policy. SPEC is one of: `period_ms=N`,
        /// `sync_window_ms=N`, `external`,
        /// `data_trigger=NAME` (or `trigger=NAME`). N must be > 0
        /// for the time-based forms; bare `period_ms` (no `=N`)
        /// is rejected. Omitting `--policy` falls back to the
        /// defaulting rules below.
        ///
        /// `external` is a self-triggering INGRESS/DRIVER node: it
        /// watches a non-Cerulion signal (a device fd, a blocking
        /// SDK) and fires itself. The scaffold includes a
        /// `HostDriven` `external_source()` stub (compiles as-is) —
        /// replace it with a device `Fd`/`Blocking` source. Note:
        /// `graph run` refuses a `HostDriven` external node on the
        /// live path (it can never self-trigger there).
        ///
        /// Defaulting rules when `--policy` is omitted:
        /// - 0 inputs (no `-i` and no `-T`): error — a source-only
        ///   node must declare a non-data policy explicitly
        ///   (`--policy period_ms=N` or `--policy external`).
        /// - 1+ inputs (any `-i` count, no `-T`): no policy attr
        ///   is written; the runtime fires on any input arrival
        ///   and emits a warning at graph-build time so the user
        ///   notices and can pick a more specific policy if
        ///   desired. Use `--policy sync_window_ms=N`,
        ///   `--policy data_trigger=NAME`, or `-T SCHEMA NAME`
        ///   to opt into a non-default trigger.
        /// - `-T` declares an explicit trigger input regardless
        ///   of `-i` count, equivalent to
        ///   `--policy data_trigger=<NAME>`.
        ///
        /// `-T` and `--policy data_trigger=NAME` are equivalent;
        /// supply at most one. `--policy` and `-T` may be
        /// combined only when they agree on the trigger input.
        #[arg(long, value_name = "SPEC")]
        policy: Option<String>,
        /// Use raw FFI template instead of #\[cerulion_node\] macro
        #[arg(long)]
        raw_ffi: bool,
    },
    /// Delete a node type
    Delete {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
    },
    /// Modify a node (add ports, change trigger policy)
    Modify {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
        /// Add input port: SCHEMA NAME (both required).
        #[arg(short = 'i', long = "input", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        input: Vec<String>,
        /// Add a TRIGGER input: SCHEMA NAME. Adds the input with
        /// `#[input(trigger)]` AND sets the node's policy to
        /// `data_trigger=<NAME>`, clearing any conflicting
        /// `period_ms` / `sync_window_ms` /
        /// `external` macro args. SCHEMA and NAME are BOTH
        /// required so multi-`-T` user errors produce a 4-element
        /// vec the engine can detect, not a 2-element vec it
        /// silently misreads.
        #[arg(short = 'T', long = "trigger-input", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        trigger_input: Vec<String>,
        /// Add output port: SCHEMA NAME (both required).
        #[arg(short = 'o', long = "output", num_args = 2, value_names = ["SCHEMA", "NAME"])]
        output: Vec<String>,
        /// Set the trigger policy. SPEC is one of: `period_ms=N`,
        /// `sync_window_ms=N`, `external`,
        /// `data_trigger=NAME` (or `trigger=NAME`). N must be > 0
        /// for the time-based forms; bare `period_ms` (no `=N`)
        /// is rejected.
        ///
        /// `external` turns the node into a self-triggering
        /// ingress/driver. Unlike `node create`, toggling it ON
        /// here does NOT inject the required `external_source()`
        /// method — the macro's compile error tells you exactly
        /// what to add (return `ExternalSource::Fd`/`Blocking`, or
        /// `HostDriven` for an embedder-driven node).
        ///
        /// `data_trigger=NAME` requires NAME to match an input —
        /// either an existing one or one being added in the same
        /// `node modify` call via `-i` or `-T`.
        #[arg(long, value_name = "SPEC")]
        policy: Option<String>,
    },
    /// Build a node crate
    Build {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
        /// Build in release mode
        #[arg(long)]
        release: bool,
    },
    /// Stage a node into a graph
    Stage {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
        /// Node instance ID
        #[arg(short = 'i', long)]
        id: Option<String>,
        /// Target graph name
        #[arg(short = 'g', long, add = ArgValueCandidates::new(completion::graph_names))]
        graph: Option<String>,
        /// Prefix for topic resolution
        #[arg(short = 'p', long)]
        prefix: Option<String>,
        /// Input bindings: NAME SOURCE (e.g., -I image \[camera,image\])
        #[arg(short = 'I', num_args = 2, value_names = ["NAME", "SOURCE"])]
        input_binding: Vec<String>,
    },
    /// Run a single node
    Run {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
        /// Topic prefix
        #[arg(short = 'p', long)]
        prefix: Option<String>,
        /// Node instance ID
        #[arg(short = 'i', long)]
        id: Option<String>,
        /// Force the release-profile cdylib (`target/release`) when
        /// loading the node, matching `cerulion node build --release`.
        /// Without this flag the freshest built cdylib is auto-selected
        /// between `target/debug` and `target/release` (the profile you
        /// most recently built wins); the other profile stays a
        /// fallback when only one is built. Mirrors `graph run
        /// --release` (CER-193, CER-426).
        #[arg(long)]
        release: bool,
        /// Disable the CPU C-state cap (Linux PM_QoS via /dev/cpu_dma_latency).
        /// On the live (real/external) clock the cap is DERIVED from the
        /// graph's own tightest timing: a latency-sensitive graph auto-caps
        /// idle to a shallow C-state at C1, recovering MOST of the ~13µs
        /// C-state cold-wake at low power (full ~13µs only via a C0 pin,
        /// `CERULION_CPU_DMA_LOCK=1`); a latency-tolerant graph takes no lock (idles deep
        /// for power). Pass this to disable the cap entirely (laptops /
        /// power-sensitive dev). Force a hard C0 pin with
        /// `CERULION_CPU_DMA_LOCK=1`, or an explicit µs cap with
        /// `CERULION_CPU_DMA_LOCK_US=N`. Linux-only.
        #[arg(long = "no-cpu-dma-lock")]
        no_cpu_dma_lock: bool,
        /// Hard opt-out from the live-loop CPU monitor-wait park (CER-443) —
        /// env cannot re-enable it. Without this flag, on the live
        /// (real/external) clock the park AUTO-resolves when a real CPU
        /// monitor-wait primitive exists (Linux x86_64 WAITPKG / Linux aarch64
        /// WFE): it replaces the blocking WaitSet wait with a shallow per-core
        /// park (no deep cpuidle C-state), cutting the post-idle cold-wake
        /// latency at a small per-core power cost (and suppressing the global
        /// C-state cap). That auto resolution is tunable via
        /// `CERULION_MONITOR_WAIT` / `CERULION_DOORBELL` (`"1"` force-on /
        /// `"0"` disable) — but only when this flag is absent. No effect off
        /// the supported targets (no primitive → blocking WaitSet).
        #[arg(long = "no-monitor-wait")]
        no_monitor_wait: bool,
        /// Network kill-switch (CER-752), mirroring `graph run --network off`.
        /// The only accepted value is `off`: run LOCAL-ONLY — no network
        /// gateway is spawned and the node's topics are not network-viewable.
        /// Absent = the permissive default: a real-clock `node run` spawns the
        /// network gateway and every produced topic is announced (unpaired;
        /// pairing pending — CER-683).
        #[arg(long = "network", value_name = "MODE", value_parser = ["off"])]
        network: Option<String>,
    },
    /// List all node types
    List,
    /// Show info about a node type
    Info {
        /// Node type name
        #[arg(add = ArgValueCandidates::new(completion::node_types))]
        node_type: String,
    },
}

#[derive(Subcommand)]
pub enum GraphAction {
    /// Create a new graph
    Create {
        /// Graph name
        // CER-925 R1: no completer — see `NodeAction::Create`. The existing
        // graphs are exactly what this verb rejects (`GraphExists`).
        name: String,
        /// Namespace prefix
        #[arg(short = 'n', long)]
        prefix: Option<String>,
    },
    /// Run a graph
    Run {
        /// Graph name
        #[arg(add = ArgValueCandidates::new(completion::graph_names))]
        name: String,
        /// Clock that drives the run (CER-222):
        ///   `real`     — live event-driven loop on RealClock (default; wakes
        ///                within microseconds of a publish on the iceoryx2 WaitSet),
        ///   `virtual`  — deterministic VirtualClock + 1 ms poll loop (replay,
        ///                deterministic runs, benchmarking),
        ///   `external` — ExternalClock for an external time master. INERT today
        ///                (CER-354/CER-424 not wired): the clock starts at 0 and
        ///                will not advance, so time-based triggers never fire.
        #[arg(long = "time-source", value_enum, default_value_t = TimeSource::Real)]
        time_source: TimeSource,
        /// Skip the pre-flight validation REPORT — NOT validation as such.
        /// The graph topology, the macro trigger wiring and the node
        /// libraries are re-checked unconditionally when the graph is
        /// built, so this flag cannot get a run past those; it moves you
        /// from one refusal to an identical one. What it does still skip
        /// is the `schema:` family — a NON-EMPTY value that names no
        /// resolvable schema, contradicts the producing node's own
        /// declaration, or disagrees with the consumer's. Those run: the
        /// wire layout comes from the node's Rust type, not the label. The
        /// bill arrives later, as a mislabelled bag channel, a Studio
        /// decoder that renders nothing, or a replay that refuses a
        /// healthy bag. An ABSENT or empty `schema:` is NOT in that set —
        /// it is a topology refusal, so this flag cannot skip it either.
        /// Nor is an AMBIGUOUS spelling — one the workspace defines more
        /// than once (two files declaring it, a file stem beside an entry
        /// of that name, a YAML definition the `.msg` store also spells, a
        /// bare name the store defines in two packages): the workspace
        /// lookup refuses it, naming every source, before the graph is
        /// built, report or no report — and a schema file it cannot read or
        /// parse stops the run the same way, naming that file.
        ///
        /// Conflicts with `--record`: those later bills are exactly what a
        /// RECORDING makes permanent. A bag written with the schema checks
        /// off carries channel labels nothing ever checked, so it cannot be
        /// honestly replay-verified — `bag play --resim` either refuses a
        /// healthy bag or, worse, replays one under a wrong label. Refused
        /// at PARSE, so it costs no shared memory, no workers and no bag.
        #[arg(long, conflicts_with = "record")]
        no_validate: bool,
        /// Force release-profile cdylibs (`target/release`) when
        /// loading node libraries, matching `cerulion node build
        /// --release`. Without this flag the freshest built cdylib is
        /// auto-selected between `target/debug` and `target/release`
        /// (the profile you most recently built wins); the other
        /// profile stays a fallback when only one is built. The pre-run
        /// validation uses the same mode (CER-193, CER-426).
        #[arg(long)]
        release: bool,
        /// Disable the CPU C-state cap (Linux PM_QoS via /dev/cpu_dma_latency).
        /// On the live (real/external) clock the cap is DERIVED from the
        /// graph's own tightest timing: a latency-sensitive graph auto-caps
        /// idle to a shallow C-state at C1, recovering MOST of the ~13µs
        /// C-state cold-wake at low power (full ~13µs only via a C0 pin,
        /// `CERULION_CPU_DMA_LOCK=1`); a latency-tolerant graph takes no lock. No effect
        /// under `--time-source virtual` (the poll loop never idles; auto
        /// takes no lock there). Force a hard C0 pin with
        /// `CERULION_CPU_DMA_LOCK=1`, or an explicit µs cap with
        /// `CERULION_CPU_DMA_LOCK_US=N`. Linux-only.
        #[arg(long = "no-cpu-dma-lock")]
        no_cpu_dma_lock: bool,
        /// Hard opt-out from the live-loop CPU monitor-wait park (CER-443) —
        /// env cannot re-enable it. Without this flag, on the live
        /// (real/external) clock the park AUTO-resolves when a real CPU
        /// monitor-wait primitive exists (Linux x86_64 WAITPKG / Linux aarch64
        /// WFE): it replaces the blocking WaitSet wait with a shallow per-core
        /// park (no deep cpuidle C-state), cutting the post-idle cold-wake
        /// latency at a small per-core power cost (and suppressing the global
        /// C-state cap). That auto resolution is tunable via
        /// `CERULION_MONITOR_WAIT` / `CERULION_DOORBELL` (`"1"` force-on /
        /// `"0"` disable) — but only when this flag is absent. No effect under
        /// `--time-source virtual` (the poll loop never parks).
        #[arg(long = "no-monitor-wait")]
        no_monitor_wait: bool,
        /// Record the run to an MCAP bag (CER-349) — replay-grade for the
        /// graph's OWN declared topics. `--record` alone writes to
        /// `./recordings/`; `--record=DIR` writes to `DIR`
        /// (created if missing; the `=` form is REQUIRED for a custom dir so
        /// the optional value can never swallow the graph-name positional).
        /// The recorder (`cerulion bagd`) taps every graph-owned topic with
        /// its EXACT schema, and also DISCOVERS live producers the graph never
        /// declared and records them in ATTACH mode (CER-942): those channels
        /// carry only the wire schema hash, so a bag that discovered anything
        /// is observability-grade overall (the recorder says so when it creates
        /// the bag) and `cerulion bag play --resim` SKIPS the discovered
        /// channels rather than verifying them. Every live producer the
        /// recorder still could
        /// not record is named in the bag's
        /// `__cerulion/record_coverage.json`, which `cerulion bag info`
        /// prints. This command builds the recorder's own command line, so its
        /// discovery flags are not reachable from here: set
        /// `CERULION_RECORD_DISCOVERY=off` to record only the graph's declared
        /// topics, or `CERULION_RECORD_DISCOVERY_SETTLE_MS` to change how long
        /// bag creation waits for late producers. A single-process run captures faithful real
        /// cumulative `fire_time_ns` on an honest gating clock, so the graph's
        /// own topics replay identically (CER-351). A `process_groups:` graph records
        /// the MULTI-PROCESS run (CER-584: ONE bag — every worker's trace
        /// ring + the supervisor's departure ring, rank-stamped; workers stay
        /// quantum-timed in barrier lockstep). `--single-process` forces the
        /// single-process recording instead. Live-clock only: combining
        /// `--record` with `--time-source virtual|external` is rejected.
        /// Unix-only.
        // CER-925 R1: `String`-typed, so the original `PathBuf` sweep missed
        // it and it would have completed NOTHING. Found by the structural
        // walk in `completion_wiring_tests`.
        #[arg(
            long,
            value_name = "DIR",
            num_args = 0..=1,
            require_equals = true,
            default_missing_value = "recordings",
            value_hint = clap::ValueHint::DirPath
        )]
        record: Option<String>,
        /// How `--record` captures the environment into the bag's `env.json`
        /// attachment (CER-349):
        ///   `allowlist` — (default) record all variable NAMES; record VALUES
        ///                 only for CERULION_*, RUST_LOG, IOX2_*; every other
        ///                 value is redacted to an FNV-1a 64 hash (replay can
        ///                 DETECT divergence without the bag leaking secrets),
        ///   `all`       — verbatim values (full replay fidelity; the bag
        ///                 embeds secrets — warned loudly at record time),
        ///   `none`      — names only (replay env-fidelity off).
        /// Requires `--record` (a loud parse error otherwise).
        #[arg(
            long = "record-env",
            value_enum,
            default_value_t = RecordEnv::Allowlist,
            requires = "record"
        )]
        record_env: RecordEnv,
        /// Where the spawned `bagd` recorder runs (CER-349): a core id to PIN
        /// it to, or `none` to float. Absent = AUTO: on Linux with >= 4 logical
        /// cores bagd is pinned to the highest-numbered core (loudly logged) —
        /// box-measured, pinning bagd is THE x86 recording-tail fix (recorded
        /// p99 21µs → 9µs, p99/p50 3.4 → 1.2); small-core (capacity-bound)
        /// boxes skip the auto pin and recording shares cores. Pinning is
        /// Linux-only: elsewhere an explicit core warns loudly and bagd
        /// floats. Requires `--record`.
        #[arg(
            long = "record-cpu",
            value_name = "CORE|none",
            value_parser = parse_record_cpu,
            requires = "record"
        )]
        record_cpu: Option<RecordCpuArg>,
        /// Multi-process worker-death policy (CER-508; only meaningful when the
        /// graph declares `process_groups:` — passed on a single-process graph
        /// it warns and is ignored). Default: `continue`.
        ///   `continue` — a crashed worker is dropped from the shared barrier;
        ///                the survivors keep running DEGRADED (loud error
        ///                naming the lost group; the run still exits 0 unless
        ///                EVERY worker crashed).
        ///   `fail`     — any worker death stops the whole deployment
        ///                (fail-loud, deterministic; for CI/replay).
        #[arg(long = "peer-loss", value_enum)]
        peer_loss: Option<PeerLoss>,
        /// Force the single-process MONOLITH path even when the graph declares
        /// `process_groups:` (skips the multi-process supervisor; the graph
        /// runs with identical results per the determinism firewall, just
        /// without process isolation). On a graph WITHOUT `process_groups:`
        /// this also skips the multi-process auto-partition default entirely
        /// (CER-509): no derivation, no confirm — the plain monolith run.
        /// Conflicts with --auto-partition (contradictory intent).
        #[arg(long = "single-process")]
        single_process: bool,
        /// Network kill-switch (CER-752). The only accepted value is `off`:
        /// run LOCAL-ONLY — no network gateway is spawned, no zenoh session
        /// opens, and nothing crosses the machine boundary (a loud notice
        /// says so). Absent = the CER-752 default: an enabled `network:`
        /// block runs STRICT (exactly the declared egress/ingress); a graph
        /// with NO (or a disabled) block runs PERMISSIVE — a gateway process
        /// announces every produced topic on the LAN (scouting on; unpaired,
        /// pairing pending — CER-683). Restrict exposure with a `network:`
        /// block, or pass `off` for local-only.
        #[arg(long = "network", value_name = "MODE", value_parser = ["off"])]
        network: Option<String>,
        /// Cap (entries) on the in-memory fire-trace ring. The trace is a
        /// BOUNDED observability window, NOT the replay record (recording is
        /// the bag's job) — unbounded growth would cost ~2 GB/h at
        /// 1 kHz × 10 nodes, so `0` (unbounded) is rejected; there is no
        /// unbounded escape hatch. Multi-process deployments stamp this cap
        /// into every worker.
        #[arg(
            long = "trace-limit",
            default_value_t = cerulion_cli_engine::graph_cmd::PRODUCTION_TRACE_LIMIT as u64,
            value_parser = clap::value_parser!(u64).range(1..)
        )]
        trace_limit: u64,
        /// Decline this run's per-rank SCHEDULER-TRACE rings (CER-1002).
        ///
        /// By default every multi-process `graph run` provisions one
        /// shared-memory scheduler-trace ring per rank (~40 MiB apparent
        /// each, converging to resident only as it fills) so a Flashback
        /// capture of the run can be RE-EXECUTED with
        /// `cerulion bag play --resim`. This declines them, for a
        /// memory-tight robot that would rather not spend it.
        ///
        /// It also stops the Flashback WINDOW RECORDER being started for
        /// this run: with no trace rings nothing captured could be
        /// re-executed, so the run takes no captures at all rather than
        /// frames-only ones. `CERULION_FLASHBACK=off` is the SEPARATE
        /// switch for the state plane + anchors; the two are orthogonal.
        ///
        /// NOT `--trace-limit`, which caps the IN-MEMORY fire-trace deque a
        /// running graph keeps for `cerulion graph` introspection. These are
        /// the SHARED-MEMORY rings a recorder drains.
        ///
        /// Conflicts with `--record`: a recording without a scheduler trace
        /// is not a recording.
        ///
        /// On the shapes that mint no ring anyway it is NOT a no-op: a
        /// `--single-process` or `--time-source external` run still has its
        /// window recorder stopped by it, so the run takes no captures. Each
        /// says at launch which it was.
        #[arg(long = "no-rings", conflicts_with = "record")]
        no_rings: bool,
        /// Re-derive the multi-process partition even when the graph already
        /// declares `process_groups:` (CER-509). Shows the derived partition
        /// (cost artifact at `graphs/<NAME>.costs.yaml` when present, else
        /// the process-per-node baseline) + the diff vs the existing block,
        /// then offers keep-existing vs apply on a TTY; with --yes it
        /// applies; with no TTY and no --yes the run uses the re-derived
        /// groups IN-MEMORY (file untouched) with a loud notice.
        /// Unpartitioned graphs on Unix under the real clock derive a
        /// partition BY DEFAULT — this flag matters for re-deriving over an
        /// existing block. Conflicts with --single-process (contradictory
        /// intent).
        #[arg(long = "auto-partition", conflicts_with = "single_process")]
        auto_partition: bool,
        /// Persist the auto-derived partition into the graph YAML without
        /// the interactive confirm (surgical rewrite + `.bak` backup).
        /// Without it: a TTY run asks y/N; a non-TTY run NEVER mutates the
        /// file and runs the derived groups in-memory (loud notice). Only
        /// meaningful when a partition is being derived (unpartitioned
        /// graphs by default, or --auto-partition).
        #[arg(long)]
        yes: bool,
    },
    /// Validate a graph against the workspace
    Validate {
        /// Graph name
        #[arg(add = ArgValueCandidates::new(completion::graph_names))]
        name: String,
        /// Force release-profile cdylibs (`target/release`) when
        /// resolving node libraries, matching `cerulion node build
        /// --release`. Without this flag the freshest built cdylib is
        /// auto-selected between `target/debug` and `target/release`
        /// (the profile you most recently built wins); the other
        /// profile stays a fallback when only one is built. The report
        /// shows which profile was found (CER-193, CER-426).
        #[arg(long)]
        release: bool,
    },
    /// List all graphs
    List,
    /// Show the derived DAG levelization of a graph (read-only, CER-509):
    /// one row per level (nodes in graph order + each node's trigger
    /// policy), the triggering topic edges leaving each level, a summary
    /// line, and — when the graph declares `process_groups:` — the
    /// group -> owned-level band mapping with the spawner-consumability
    /// verdict. Derives from the SAME levelization the runtime uses.
    Levels {
        /// Graph name
        #[arg(add = ArgValueCandidates::new(completion::graph_names))]
        name: String,
    },
    /// Profile a graph LIVE and write its cost snapshot (CER-509): runs the
    /// graph on the real clock until every node reaches its fire target (or
    /// the duration cap / Ctrl+C stops it early), measures per-node p50 tick
    /// durations + per-edge fire rates, and writes them to
    /// `graphs/<NAME>.costs.yaml` — the input of the cost-aware
    /// auto-partitioner. By DEFAULT each node's fire target is AUTO-DERIVED
    /// from a short warm-up observation of its own rate (CER-655), so
    /// low-rate graphs profile at the defaults; `--fires N` overrides with
    /// one uniform target for every node. Under-sampled nodes are ISOLATED
    /// (no cost recorded — they stay in their own process group) and warned
    /// loudly; isolation is a valid outcome, not an error (exit 0).
    Profile {
        /// Graph name
        #[arg(add = ArgValueCandidates::new(completion::graph_names))]
        name: String,
        /// Observation-window cap in seconds: the run stops here even if some
        /// node has not reached its fire target yet (that node is isolated).
        #[arg(long, default_value_t = 30, value_parser = clap::value_parser!(u64).range(1..))]
        duration: u64,
        /// Uniform per-node fire target OVERRIDE: the run stops early once
        /// EVERY node has fired this many times. Omit (the default) to
        /// AUTO-DERIVE each node's target from a warm-up observation of its
        /// own rate (clamp(cap/10, 1s, 3s) warm-up; targets scaled to each
        /// node's projected fires over the cap, clamped into [20, 1000]).
        /// Use the override for nodes whose FIRST fire lands after the
        /// warm-up (e.g. a period longer than 3 s) — silent-through-warm-up
        /// nodes derive no target and stay isolated.
        #[arg(long, value_parser = clap::value_parser!(u64).range(1..))]
        fires: Option<u64>,
        /// Artifact output path (default: `graphs/<NAME>.costs.yaml` in the
        /// workspace).
        #[arg(short = 'o', long, value_hint = clap::ValueHint::AnyPath)]
        out: Option<PathBuf>,
    },
    /// Derive and write the graph's `process_groups:` partition (CER-509) +
    /// `level_assignments:` refined levels (CER-660). With a cost snapshot
    /// (from `cerulion graph profile`) the levels are first REFINED
    /// (cost-aware: expensive low-rate nodes shift later within their
    /// topological slack, and a dense slow chain with no slack CASCADES
    /// later as a whole — growing the level count — so cheap fast chains'
    /// gating levels stay cheap; the max-rate chain's nodes never move) and
    /// the partition is the cost-aware fused optimum banded over the REFINED
    /// levels; without a snapshot it is the process-per-node baseline over
    /// the derived (Kahn) levels. The `level_assignments:` block is emitted
    /// ONLY when refinement moves ≥1 node — its presence always means "these
    /// levels differ from Kahn"; when refinement changes nothing (including
    /// every no-costs run) the block is OMITTED and a stale existing one is
    /// REMOVED (a hand-written block equal to Kahn changes nothing and is
    /// removed too — the preview names it). The runtime and bag replays
    /// consume the written block, so refined levels only ever take effect
    /// PERSISTED (a plain `graph run` never refines in-memory). The graph
    /// file is rewritten SURGICALLY — only the `process_groups:` /
    /// `level_assignments:` blocks change (a stale `process_group_order:`
    /// block is also removed); every comment and all other formatting are
    /// preserved byte-for-byte, and the previous file is backed up to
    /// `<file>.bak`. Never writes without consent: an interactive y/N
    /// confirm shows the proposed grouping + level delta + YAML diff;
    /// non-TTY runs require `--yes`. A stale/invalid existing block of
    /// either kind is REPLACED, not fatal — this verb is the recovery tool.
    Partition {
        /// Graph name
        #[arg(add = ArgValueCandidates::new(completion::graph_names))]
        name: String,
        /// Cost-snapshot artifact path (default: `graphs/<NAME>.costs.yaml`).
        /// An explicitly named file must exist and parse — it never silently
        /// falls back to the baseline; only the ABSENT default path does.
        #[arg(long, value_hint = clap::ValueHint::FilePath)]
        costs: Option<PathBuf>,
        /// Per-group compute budget in ns (Σ member p50 ≤ budget gates each
        /// fusion). Omit (the default) to use the cost artifact's FROZEN
        /// core-count budget (CER-654: `ceil(Σ p50 / cores)`, computed at
        /// profile time on the profiling box); a pre-CER-654 artifact or an
        /// absent frozen value falls back to unbounded fusion with a loud
        /// info suggesting a re-profile. An explicit value always overrides
        /// the frozen one. Zero is rejected.
        #[arg(long = "budget-ns", value_parser = clap::value_parser!(u64).range(1..))]
        budget_ns: Option<u64>,
        /// Print the derived partition + the YAML diff and exit WITHOUT
        /// writing. Wins over --yes when both are passed.
        #[arg(long = "dry-run")]
        dry_run: bool,
        /// Skip the interactive confirmation and write (scripts / non-TTY
        /// runs). Overridden by --dry-run.
        #[arg(long)]
        yes: bool,
    },
    /// HIDDEN (CER-508): run ONE worker process of a multi-process
    /// deployment from a serialized `WorkerPlan`.
    ///
    /// NOT a user-facing verb — the multi-process SUPERVISOR (chunk 3, the
    /// `graph run` multi-process path) spawns this once per process group,
    /// handing each worker its own plan file (its subgraph, cross-process
    /// rank, barrier participant-map, shared handed quantum + iceoryx2 Config).
    /// Users always run the documented `graph run`; this stays `hide = true`.
    #[command(hide = true)]
    RunWorker {
        /// Path to the serialized `WorkerPlan` JSON the supervisor wrote.
        #[arg(long)]
        #[arg(value_hint = clap::ValueHint::FilePath)]
        plan: PathBuf,
    },
    /// HIDDEN (CER-752): run the network GATEWAY process of a `graph run`.
    ///
    /// NOT a user-facing verb — the `graph run` monolith arm AND the
    /// multi-process supervisor spawn this once per running graph, handing it
    /// a serialized `GatewayHandoff` (the network plan + zenoh params + the
    /// run's iceoryx2 namespace). The gateway owns the robot's whole network
    /// plane (one robot = one peer); graph/worker processes stay network-free.
    /// Users always run the documented `graph run`; this stays `hide = true`.
    #[command(hide = true)]
    RunGateway {
        /// Path to the serialized `GatewayHandoff` JSON the parent wrote.
        #[arg(long)]
        #[arg(value_hint = clap::ValueHint::FilePath)]
        handoff: PathBuf,
    },
}

#[derive(Subcommand)]
pub enum TopicAction {
    /// List active topics — genuinely-LOCAL shared-memory topics first, then
    /// REMOTE topics (CER-752: automagic — remote discovery runs BY DEFAULT with
    /// multicast + gossip scouting ON, so unpaired robots on the LAN show up with
    /// no flags). CER-837: a local topic that is really a MIRROR of a remote robot
    /// (its origin is in the `/__cerulion/mirrors` provenance registry) folds into
    /// REMOTE as a `● streaming` row attributed to its robot, never LOCAL. Pass
    /// extra `--connect`/`--listen` locators to reach peers scouting can't; pass
    /// `--no-network` to skip the remote NETWORK query (scripts / CI).
    List {
        /// Skip the REMOTE network discovery query (scripts / CI / offline).
        /// Without it, remote discovery runs automatically (scouting ON). NOTE
        /// (CER-837): locally-mirrored remote topics still render as
        /// REMOTE·streaming — their provenance is read from LOCAL SHM, not the
        /// network — so this skips only the network gather, not the mirror fold.
        #[arg(long = "no-network")]
        no_network: bool,
        /// Additional zenoh locator to connect to for discovery (repeatable),
        /// e.g. tcp/192.168.123.99:7683 (the well-known permissive-gateway
        /// port). Adds to the default scouting session.
        #[arg(long, value_name = "LOCATOR")]
        connect: Vec<String>,
        /// Additional zenoh locator to listen on for discovery (repeatable),
        /// e.g. tcp/0.0.0.0:7447. Adds to the default scouting session.
        #[arg(long, value_name = "LOCATOR")]
        listen: Vec<String>,
        /// OPT-IN: also unicast-sweep the local /24(s) on the well-known
        /// gateway port for robots that BOTH multicast and mDNS reflection
        /// hide (CER-760 rung 4). OFF by default — a horizontal connect sweep
        /// reads as port-scan reconnaissance to corporate/enterprise IDS, so it
        /// never fires without this flag. Only confirmed Cerulion gateways
        /// (verified by a beacon probe) are shown.
        #[arg(long)]
        scan: bool,
    },
    /// Show info about a topic (schema name + last sequence/timestamp)
    ///
    /// Reads a LOCAL topic, or — if the name is not a local topic — a REMOTE
    /// robot's topic discovered on the LAN: it is demanded automatically via
    /// gateway ingress and released when the command exits. Set
    /// CERULION_NETWORK=off to disable remote discovery (local-only).
    Info {
        /// Topic name
        #[arg(add = ArgValueCandidates::new(completion::topics))]
        topic: String,
    },
    /// Echo messages from a topic (decoded fields)
    ///
    /// Reads a LOCAL topic, or — if the name is not a local topic — a REMOTE
    /// robot's topic discovered on the LAN: it is demanded automatically via
    /// gateway ingress and released when the command exits. Set
    /// CERULION_NETWORK=off to disable remote discovery (local-only).
    Echo {
        /// Topic name
        #[arg(add = ArgValueCandidates::new(completion::topics))]
        topic: String,
        /// Max array elements to render before truncating with `...` + a
        /// `(N elements)` tail (default 128; matches `ros2 topic echo
        /// --truncate-length`). Must be a positive integer — 0 / negative are
        /// rejected at parse.
        #[arg(
            long = "truncate-length",
            value_name = "N",
            default_value_t = cerulion_cli_engine::topic_cmd::DEFAULT_ECHO_TRUNCATE_LENGTH as u64,
            value_parser = clap::value_parser!(u64).range(1..)
        )]
        truncate_length: u64,
    },
    /// Measure publish rate on a topic
    ///
    /// Reads a LOCAL topic, or — if the name is not a local topic — a REMOTE
    /// robot's topic discovered on the LAN: it is demanded automatically via
    /// gateway ingress and released when the command exits. Set
    /// CERULION_NETWORK=off to disable remote discovery (local-only).
    Hz {
        /// Topic name
        #[arg(add = ArgValueCandidates::new(completion::topics))]
        topic: String,
    },
}

/// CER-939: the `cerulion bag` verb family — bag as a data source.
#[derive(Subcommand)]
pub enum BagAction {
    /// Play a bag's recorded frames back onto local shared memory, wall-paced.
    ///
    /// Every frame is re-published BYTE-VERBATIM (its wire header's `sequence`
    /// and `timestamp_ns` intact) under its RECORDED topic name, in the order
    /// the recorder saw them, paced from the bag's own log times. Nothing is
    /// re-executed and no node is loaded — to anything attached to local SHM
    /// this is indistinguishable from a robot publishing. Local only: the frames
    /// land in THIS machine's shared memory.
    ///
    /// A topic whose publisher slot is already held by a live producer is
    /// refused BY NAME and skipped; the remaining topics still play.
    ///
    /// `--resim` (CER-982) switches the verb into RE-EXECUTION: the bag's graph
    /// runs again against your workspace's current node builds instead of its
    /// frames being republished. Bare `--resim all` makes no claim about whether
    /// the result matches the recording (divergence is the product, and a
    /// completed re-execution exits 0); `--verify` adds the byte-comparison and
    /// its 0–6 exit contract. This is where the removed `cerulion replay` verb
    /// went (CER-982); that spelling no longer exists.
    Play {
        /// Path to the `.mcap` bag — a `cerulion graph run --record` recording,
        /// a Flashback capture, a `cerulion bag record --run` mid-run attach, or
        /// a standalone `cerulion bag record`. `--resim` additionally needs a
        /// SCHEDULER TRACE, which the first three carry (CER-1002: every
        /// multi-process run provisions the rings, recording or not) and a
        /// standalone `bag record` does not.
        // CER-925: `.mcap` is a hard contract here (`cerulion_bag` writes
        // standard MCAP and nothing else), so this narrows to bags rather than
        // offering every file. Directories still complete — `complete_path`
        // keeps a filtered-out directory as a traversal candidate — so a bag in
        // a subdirectory stays reachable.
        #[arg(add = ArgValueCompleter::new(
            PathCompleter::any().filter(|p| {
                p.is_dir() || p.extension().is_some_and(|e| e.eq_ignore_ascii_case("mcap"))
            })
        ))]
        bag: PathBuf,
        /// RE-EXECUTE the bag's graph instead of republishing its frames
        /// (CER-982). `all` re-executes every node; a node subset is not built
        /// yet and is refused by name. Without `--verify` this makes NO claim
        /// about matching the recording and exits 0 on any completed run.
        //
        // CER-982: deliberately a free-form `String` validated in the engine,
        // NOT a clap `value_parser` possible-values list. A clap rejection would
        // frame `--resim planner` as a bad value; the engine's refusal names
        // CER-982, says a node subset is future work, and points at the spelling
        // that works. The completer still offers the one accepted value.
        #[arg(long, value_name = "NODES|all", add = ArgValueCandidates::new(completion::resim_selections))]
        resim: Option<String>,
        /// With `--resim`: byte-compare every re-executed frame against the
        /// recording and apply the stable exit contract — 0 = identical,
        /// 1 = data violation, 2 = bag I/O or not-replay-grade, 3 = node cdylib
        /// load error or a node that PANICKED, 4 = tolerance-YAML validation
        /// error, 5 = internal error, 6 = structural trace divergence.
        #[arg(long)]
        verify: bool,
        /// Playback rate multiplier: 1.0 = the recorded pace (default), 2.0 =
        /// twice as fast, 0.5 = half speed. Must be finite and greater than 0.
        /// Playback only — a resim runs on the recording's own gating clock.
        //
        // CER-982: an `Option` rather than `default_value_t = 1.0`, because
        // clap's default erases the difference between "the user asked for the
        // recorded pace" and "the user asked for nothing" — which is exactly
        // what the `--resim` refusal has to tell apart. The default is applied
        // by `resim_cmd::resolve_play_mode` instead.
        #[arg(long, short = 'r', value_name = "N")]
        rate: Option<f64>,
        /// Restart from the beginning when the bag ends, until interrupted.
        /// NOTE: the wire timestamps REGRESS on each wrap (the frames are
        /// verbatim), which consumers read as a publisher clock-epoch reset.
        /// Playback only.
        #[arg(long = "loop")]
        repeat: bool,
        /// Play only these topics (repeatable). Default: every topic in the bag.
        /// A name the bag does not carry is a loud error listing what it does.
        /// Playback only — a resim's topic set follows the nodes that execute.
        #[arg(long, value_name = "TOPIC")]
        topics: Vec<String>,
        /// Stop after `D` SECONDS of BAG TIME (fractional accepted, e.g. 12.5).
        /// Playback stops republishing there; a resim re-executes each rank up
        /// to `D` of recorded time measured from the run's shared bag-time
        /// origin. Omit to cover the whole bag.
        //
        // CER-1289 2c2 (founder ruling, 2026-08-19): this REPLACES `--max-ticks`,
        // which is deleted outright (no alias, no shim — the CER-1257 precedent,
        // so a stale `--max-ticks` is clap's unknown-argument error). A step cap
        // cannot bound a per-rank resim: there are k step axes and no shared step
        // number. Time is better because it has a shared ORIGIN (the minimum
        // first-boundary target across the ranks), NOT because it is uniform —
        // each rank's clock starts at its own live-loop entry, so a bound names
        // one wall interval for all of them and a later-booting rank loses more
        // of its own tail.
        #[arg(long, short = 'u', value_name = "D")]
        duration: Option<f64>,
        /// Skip the first `S` SECONDS of BAG TIME (fractional accepted).
        /// PLAYBACK ONLY: a resim refuses it by name, because re-executing from
        /// the middle needs per-rank resume ANCHORS, which is the open follow-up
        /// behind the free-run mid-run resume refusal.
        #[arg(long, short = 's', value_name = "S")]
        start_offset: Option<f64>,
        /// With `--resim --verify`: write the verdict as a JSON report to this
        /// path. The same verdict is always printed to stderr; this persists the
        /// machine-readable `ReplayOutcome` for CI. Needs `--verify` — a bare
        /// `--resim` produces no verdict for it to persist.
        #[arg(long, value_name = "PATH", value_hint = clap::ValueHint::AnyPath)]
        report: Option<PathBuf>,
        /// With `--resim --verify`: path to a tolerance YAML. RELAXES the
        /// byte-exact diff by a `max_abs` / `max_rel` / `rmse` / `bbox_iou` /
        /// `set_equal` / `set_subset` / `ordered_list_equal` metric (a divergence
        /// past the bound is an exit-1 `tolerance-exceeded` violation), resolved
        /// per field by the precedence `fields[path]` > topic-wide `metric:` >
        /// `default_metric` — so a bare topic `metric:` covers every field of
        /// that topic and a `default_metric` covers every field of every topic.
        /// Every field that resolves to `bit_exact` — and every untargeted topic
        /// — stays BYTE-EXACT (a drifting sibling field is still caught).
        /// Validated STRICTLY as a PRE-FLIGHT gate: an unknown key, out-of-range
        /// threshold, unresolvable topic/field name, or a non-`bit_exact` metric
        /// on a publisher-opaque field (per-field OR via a topic-wide/default
        /// metric expanded over the schema) is a hard exit-4 error (with
        /// suggestions) BEFORE any node loads.
        #[arg(long, value_name = "PATH", value_hint = clap::ValueHint::FilePath)]
        tolerance: Option<PathBuf>,
        /// With `--resim`: refuse the re-execution unless EVERY executed node's
        /// state was restored (CER-1053 chunk 9).
        ///
        /// Only a recording that begins MID-RUN restores anything — one that
        /// begins at step 0 starts every node from its constructor, which IS
        /// the recorded state — so this flag is inert on a from-start bag.
        ///
        /// It governs COVERAGE, not drift: a recorded state whose SHAPE differs
        /// from this build's is already terminal, so there is no looser
        /// behaviour for it to tighten. What it refuses is the reading a
        /// default re-execution takes silently — "this node declares no
        /// restorable state, so it starts fresh" — which is right for a
        /// stateless node and a gap for one that simply has no state derive
        /// yet. Only you know which, so the default reports and this flag
        /// refuses.
        ///
        /// Valid in BOTH resim modes: it is a PRECONDITION on the run, not a
        /// claim about the recording, so a neutral `--resim all` honours it too
        /// (see `resim_cmd`).
        #[arg(long)]
        strict_state: bool,
    },
    /// Show what a bag holds — topics, frame counts, schemas, time span —
    /// without publishing anything. The same pre-scan `bag play` prints.
    Info {
        /// Path to the `.mcap` bag.
        #[arg(add = ArgValueCompleter::new(
            PathCompleter::any().filter(|p| {
                p.is_dir() || p.extension().is_some_and(|e| e.eq_ignore_ascii_case("mcap"))
            })
        ))]
        bag: PathBuf,
    },
    /// Rewrite a bag whose embedded graph carries keys the graph format no
    /// longer defines (CER-1413), into a NEW bag that `bag play --resim` reads.
    ///
    /// A bag recorded before 2026-07-31 embeds your on-disk `graphs/<name>.yaml`
    /// rather than the config the run actually executed, so a since-removed
    /// setting — a pre-CER-377 `policy:` block is the one that really happens —
    /// travels into the bag and now refuses to resim. An MCAP attachment is
    /// sealed, so there is no way to edit it in place; this writes a corrected
    /// copy instead.
    ///
    /// It lists every key it would remove, with the path and line it sits on in
    /// the embedded document, and asks before writing. The input bag is NEVER
    /// modified: the migrated bag is a second file, `<name>.migrated.mcap`
    /// beside the original unless `-o` says otherwise. Every recorded frame,
    /// the scheduler trace and every other attachment are copied through
    /// unchanged.
    ///
    /// A bag whose graph already parses is refused — there is nothing to
    /// migrate and no copy worth making.
    Migrate {
        /// Path to the `.mcap` bag to migrate. It is read, never written.
        // CER-925: same `.mcap` filter + directory traversal as `bag play`.
        #[arg(add = ArgValueCompleter::new(
            PathCompleter::any().filter(|p| {
                p.is_dir() || p.extension().is_some_and(|e| e.eq_ignore_ascii_case("mcap"))
            })
        ))]
        bag: PathBuf,
        /// Where to write the migrated bag. Defaults to `<name>.migrated.mcap`
        /// beside the input. A path that already exists is refused rather than
        /// overwritten.
        #[arg(
            short = 'o',
            long,
            value_name = "PATH",
            add = ArgValueCompleter::new(
                PathCompleter::any().filter(|p| {
                    p.is_dir() || p.extension().is_some_and(|e| e.eq_ignore_ascii_case("mcap"))
                })
            )
        )]
        out: Option<PathBuf>,
        /// Show exactly which keys would be removed and write nothing. Wins
        /// over --yes.
        #[arg(long)]
        dry_run: bool,
        /// Write the migrated bag without asking. Required when stdin is not a
        /// terminal (scripts / CI).
        #[arg(long)]
        yes: bool,
    },
    /// Record LOCAL topics into a bag (CER-939) — the `ros2 bag record` shape,
    /// driving Cerulion's production recorder (`cerulion bagd`).
    ///
    /// LOCAL ONLY: this taps THIS machine's shared memory and never opens the
    /// network. Recording a robot's topics means running this verb ON the robot
    /// and transferring the file afterwards — a desk-side recorder that pulled
    /// every topic across the wire is exactly what this must not be. A named
    /// topic that is not live here is refused by name.
    ///
    /// The capture is OBSERVABILITY-GRADE: each channel carries the real wire
    /// `schema_hash` learned from its first frame, but the schema NAME is
    /// recorded as "unknown" (the wire carries no name). That is everything
    /// `cerulion bag play` needs, and `cerulion topic echo` still decodes the
    /// frames from the workspace schema store. It is NOT the replay-grade
    /// descriptor `graph run --record` writes: `cerulion bag play --resim`
    /// compatibility is not claimed.
    Record {
        /// Topics to record. Omit when using --all or --regex.
        #[arg(value_name = "TOPIC", add = ArgValueCandidates::new(completion::topics))]
        topics: Vec<String>,
        /// Record every live local topic (`ros2 bag record -a`). Cerulion's own
        /// internal channels (`__cerulion/*`, `/bagd/*`) are never auto-selected.
        #[arg(short = 'a', long)]
        all: bool,
        /// Record live local topics matching this regex (`ros2 bag record -e`).
        #[arg(short = 'e', long, value_name = "PATTERN")]
        regex: Option<String>,
        /// Drop selected topics matching this regex (repeatable). Narrows any
        /// selection, including an explicit list.
        #[arg(short = 'x', long, value_name = "PATTERN")]
        exclude: Vec<String>,
        /// Where to write the bag. One recording is one file.
        #[arg(
            short = 'o',
            long,
            value_name = "PATH",
            default_value = "recording.mcap",
            value_hint = clap::ValueHint::AnyPath
        )]
        out: PathBuf,
        // CER-1226 (founder ruling 2026-08-10): `-b/--max-bag-size`, the
        // size-cap rotation flag, is DE-PRODUCTIZED before launch — one
        // recording is one artifact. The recorder's roll path is intact and
        // library-reachable but no CLI flag drives it; see
        // `cerulion_bagd::BagdConfig::size_cap_bytes` for why re-adding a flag
        // needs the post-launch artifact-shape issue first.
        /// Stop recording after this many seconds. Omit to record until Ctrl-C.
        /// (ros2's `--max-bag-duration` SPLITS files; this one just stops.)
        #[arg(long, value_name = "SECS", value_parser = clap::value_parser!(u64).range(1..))]
        duration: Option<u64>,
        /// Milliseconds to wait for each topic's first frame before creating the
        /// bag, so its channel can carry the real wire schema hash. Frames seen
        /// during the wait are recorded, not dropped. A topic still silent when
        /// it elapses gets a hash-0 placeholder, permanently for that bag.
        #[arg(long = "schema-wait-ms", value_name = "MS", default_value_t = 2000)]
        schema_wait_ms: u64,
        /// ATTACH to a live `cerulion graph run` (CER-981).
        ///
        /// The bag then describes a RUN, not just a set of topics: it carries
        /// the run's effective graph, its env snapshot, its host identity and
        /// its run identity, and it records the topics the run DECLARES rather
        /// than everything live on this machine.
        ///
        /// Pass the flag alone when one run is live. Name a run with
        /// `--run=<RUN>` — a run id (as printed in `run.json`) or a graph
        /// name — to pick among several; several live runs and no id is a
        /// refusal, never a guess.
        ///
        /// The recording begins where it attaches: frames and scheduler trace
        /// start at the attach point, every channel is marked `attached_late`,
        /// and `__cerulion/run.json` records `attached_mid_run`. Nothing before
        /// the attach is recoverable, and nothing implies otherwise.
        // CER-925: no completer. Run ids live only in shared memory, and a TAB
        // press must never open a transport (`completions_test.rs` walks for
        // exactly that); this arg is classified free-form in the inventory.
        //
        // `require_equals` is LOAD-BEARING, not style. With a bare
        // `num_args = 0..=1` the flag GREEDILY swallows the next word, so
        // `bag record --run /topic` bound the RUN NAME to `/topic` and
        // recorded no topics — i.e. the one composition the verb explicitly
        // supports (attach to a run, record a hand-named subset of it) was
        // unspellable in its natural form, and failed by binding rather than
        // by complaining. Requiring `=` splits the two cleanly: bare `--run`
        // takes `default_missing_value` and every following word stays a
        // positional topic, while `--run=NAME` names a run. The cost, stated:
        // `--run NAME` no longer names a run — it attaches to the sole run and
        // records the topic `NAME`, which is a well-formed request rather than
        // a silent misparse, and clap's own help shows the `=` form.
        #[arg(
            long,
            value_name = "RUN",
            num_args = 0..=1,
            require_equals = true,
            default_missing_value = ""
        )]
        run: Option<String>,
    },
}

#[derive(Subcommand)]
pub enum SchemaAction {
    /// Create a new schema
    Create {
        /// Schema name
        // CER-925 R1: no completer, and this arm was the worst of the three.
        // The workspace half is what the verb rejects (already-exists), and
        // the 254 built-in `pkg/Type` names are worse than useless here: a
        // `sensor_msgs/Image` either fails on the `/` or silently creates a
        // workspace schema that SHADOWS the built-in for the whole workspace.
        name: String,
    },
    /// Delete a schema
    Delete {
        /// Schema name
        #[arg(add = ArgValueCandidates::new(completion::schema_names))]
        name: String,
    },
    /// Show info about a schema
    Info {
        /// Schema name
        #[arg(add = ArgValueCandidates::new(completion::schema_names))]
        name: String,
    },
    /// List all schemas: workspace-local (schemas/*.yaml) and built-in ROS 2 types by package
    List,
}

#[cfg(test)]
mod graph_levels_dispatch_tests {
    use super::*;

    /// CER-509 chunk A: `graph levels <name>` parses to `GraphAction::Levels`
    /// with the positional graph name.
    #[test]
    fn graph_levels_parses_name() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "levels", "perception"])
            .expect("`graph levels perception` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Levels { name },
            } => assert_eq!(name, "perception"),
            _ => panic!("expected Graph::Levels"),
        }
    }

    /// `graph levels` without a name is a loud parse error (the name is a
    /// required positional), never a silent default.
    #[test]
    fn graph_levels_requires_name() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "levels"]).is_err(),
            "`graph levels` without a graph name must be rejected"
        );
    }
}

#[cfg(test)]
mod pair_dispatch_tests {
    use super::*;

    /// CER-834: `cerulion pair go2` parses to `Commands::Pair` with the positional
    /// robot name and every optional flag defaulting to absent.
    #[test]
    fn pair_parses_positional_name_with_defaults() {
        let cli = Cli::try_parse_from(["cerulion", "pair", "go2"])
            .expect("`cerulion pair go2` must parse");
        match cli.command {
            Commands::Pair {
                robot,
                eid,
                addrs,
                code,
                account,
                label,
                key_file,
                relay_url,
                relay_disabled,
            } => {
                assert_eq!(robot.as_deref(), Some("go2"));
                assert!(eid.is_none());
                assert!(addrs.is_empty());
                assert!(code.is_none(), "the pairing code defaults to stdin");
                assert!(account.is_none(), "the account defaults to the desk key");
                assert!(label.is_none(), "the label defaults to the desk hostname");
                assert!(
                    key_file.is_none(),
                    "the key file defaults to ~/.cerulion/desk.key"
                );
                assert!(relay_url.is_none());
                assert!(!relay_disabled);
            }
            _ => panic!("expected Commands::Pair"),
        }
    }

    /// All flags together: `--eid`, `--addr` (repeatable), `--code`, `--account`,
    /// `--label`, `--key-file`, `--relay-url`, `--relay-disabled`.
    #[test]
    fn pair_parses_all_flags() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "pair",
            "--eid",
            "deadbeef",
            "--addr",
            "192.168.1.20:7842",
            "--addr",
            "[::1]:9000",
            "--code",
            "SWAN-42",
            "--account",
            "aabb",
            "--label",
            "my-desk",
            "--key-file",
            "/keys/desk.key",
            "--relay-url",
            "https://relay.example",
            "--relay-disabled",
        ])
        .expect("all pair flags must parse");
        match cli.command {
            Commands::Pair {
                robot,
                eid,
                addrs,
                code,
                account,
                label,
                key_file,
                relay_url,
                relay_disabled,
            } => {
                assert!(robot.is_none(), "--eid replaces the positional");
                assert_eq!(eid.as_deref(), Some("deadbeef"));
                assert_eq!(addrs, vec!["192.168.1.20:7842", "[::1]:9000"]);
                assert_eq!(code.as_deref(), Some("SWAN-42"));
                assert_eq!(account.as_deref(), Some("aabb"));
                assert_eq!(label.as_deref(), Some("my-desk"));
                assert_eq!(key_file, Some(PathBuf::from("/keys/desk.key")));
                assert_eq!(relay_url.as_deref(), Some("https://relay.example"));
                assert!(relay_disabled);
            }
            _ => panic!("expected Commands::Pair"),
        }
    }

    /// `cerulion pair` with no robot AND no `--eid` still PARSES (both are
    /// optional at the clap layer) — the engine's `resolve_pair_target` produces
    /// the "specify a robot" error, matching the `connect` verb's shape.
    #[test]
    fn pair_with_no_target_parses_then_engine_errors() {
        let cli = Cli::try_parse_from(["cerulion", "pair"]).expect("`cerulion pair` parses");
        match cli.command {
            Commands::Pair { robot, eid, .. } => {
                assert!(robot.is_none());
                assert!(eid.is_none());
            }
            _ => panic!("expected Commands::Pair"),
        }
    }
}

#[cfg(test)]
mod graph_profile_dispatch_tests {
    use super::*;

    /// CER-509 chunk C3 / CER-655: `graph profile <name>` (no flags) parses
    /// with the documented defaults — 30 s cap, AUTO-DERIVED per-node fire
    /// targets (`fires: None`), default artifact location (`out: None`).
    #[test]
    fn graph_profile_parses_name_with_defaults() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "profile", "perception"])
            .expect("`graph profile perception` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Profile {
                        name,
                        duration,
                        fires,
                        out,
                    },
            } => {
                assert_eq!(name, "perception");
                assert_eq!(duration, 30, "default duration cap is 30 s");
                assert_eq!(
                    fires, None,
                    "default = AUTO-DERIVE per-node targets (CER-655), not a scalar"
                );
                assert!(out.is_none(), "default artifact location when -o absent");
            }
            _ => panic!("expected Graph::Profile"),
        }
    }

    /// The graph name is a required positional — a loud parse error, never a
    /// silent default.
    #[test]
    fn graph_profile_requires_name() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "profile"]).is_err(),
            "`graph profile` without a graph name must be rejected"
        );
    }

    /// All flags together: `--duration`, `--fires`, and the short `-o`.
    #[test]
    fn graph_profile_parses_all_flags() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "profile",
            "perception",
            "--duration",
            "120",
            "--fires",
            "50",
            "-o",
            "custom/costs.yaml",
        ])
        .expect("all profile flags must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Profile {
                        name,
                        duration,
                        fires,
                        out,
                    },
            } => {
                assert_eq!(name, "perception");
                assert_eq!(duration, 120);
                assert_eq!(
                    fires,
                    Some(50),
                    "--fires N = the uniform-target override (CER-655)"
                );
                assert_eq!(out, Some(PathBuf::from("custom/costs.yaml")));
            }
            _ => panic!("expected Graph::Profile"),
        }
    }

    /// The long `--out` form parses to the same field as `-o`.
    #[test]
    fn graph_profile_parses_long_out() {
        let cli =
            Cli::try_parse_from(["cerulion", "graph", "profile", "g", "--out", "x.costs.yaml"])
                .expect("--out must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Profile { out, .. },
            } => assert_eq!(out, Some(PathBuf::from("x.costs.yaml"))),
            _ => panic!("expected Graph::Profile"),
        }
    }

    /// Zero is rejected at PARSE for both knobs (`range(1..)`) — a 0-second
    /// cap or a 0-fire override would profile nothing; the engine
    /// re-validates, but the CLI surface enforces the same invariant
    /// (contract alignment). CER-655: `--fires` is now an OPTIONAL override —
    /// an explicit `--fires 0` still fails at parse (omitting the flag, not
    /// zeroing it, is how you select auto-derive).
    #[test]
    fn graph_profile_rejects_zero_duration_and_fires() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "profile", "g", "--duration", "0"]).is_err(),
            "--duration 0 must be a parse error"
        );
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "profile", "g", "--fires", "0"]).is_err(),
            "--fires 0 must be a parse error (auto-derive is selected by OMITTING --fires)"
        );
    }
}

#[cfg(test)]
mod graph_partition_dispatch_tests {
    use super::*;

    /// CER-509 chunk D2 / CER-654: `graph partition <name>` (no flags) parses
    /// with the documented defaults — default artifact location
    /// (`costs: None`), the artifact's FROZEN budget (`budget_ns: None`),
    /// interactive (no dry-run, no yes).
    #[test]
    fn graph_partition_parses_name_with_defaults() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "partition", "perception"])
            .expect("`graph partition perception` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Partition {
                        name,
                        costs,
                        budget_ns,
                        dry_run,
                        yes,
                    },
            } => {
                assert_eq!(name, "perception");
                assert!(
                    costs.is_none(),
                    "default artifact location when --costs absent"
                );
                assert_eq!(
                    budget_ns, None,
                    "default = the artifact's frozen core-count budget (CER-654), \
                     not a scalar"
                );
                assert!(!dry_run, "not a dry run by default");
                assert!(!yes, "interactive by default — never auto-consent");
            }
            _ => panic!("expected Graph::Partition"),
        }
    }

    /// The graph name is a required positional — a loud parse error, never a
    /// silent default.
    #[test]
    fn graph_partition_requires_name() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "partition"]).is_err(),
            "`graph partition` without a graph name must be rejected"
        );
    }

    /// All flags together: `--costs`, `--budget-ns`, `--dry-run`, `--yes`.
    #[test]
    fn graph_partition_parses_all_flags() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "partition",
            "perception",
            "--costs",
            "custom/costs.yaml",
            "--budget-ns",
            "5000000",
            "--dry-run",
            "--yes",
        ])
        .expect("all partition flags must parse (dry-run + yes coexist; dry-run wins)");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Partition {
                        name,
                        costs,
                        budget_ns,
                        dry_run,
                        yes,
                    },
            } => {
                assert_eq!(name, "perception");
                assert_eq!(costs, Some(PathBuf::from("custom/costs.yaml")));
                assert_eq!(
                    budget_ns,
                    Some(5_000_000),
                    "--budget-ns N = the explicit override (CER-654)"
                );
                assert!(dry_run);
                assert!(yes);
            }
            _ => panic!("expected Graph::Partition"),
        }
    }

    /// Zero is rejected at PARSE (`range(1..)`) — a 0 budget rejects every
    /// fusion; the engine re-validates, but the CLI surface enforces the same
    /// invariant (contract alignment).
    #[test]
    fn graph_partition_rejects_zero_budget() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "partition", "g", "--budget-ns", "0"])
                .is_err(),
            "--budget-ns 0 must be a parse error"
        );
    }

    /// An unknown flag is a loud parse error (no silent typo absorption).
    #[test]
    fn graph_partition_rejects_unknown_flag() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "partition", "g", "--force"]).is_err(),
            "an unknown --force flag must be rejected"
        );
    }
}

#[cfg(test)]
mod graph_run_auto_partition_flag_tests {
    use super::*;

    /// CER-509 chunk D3: `graph run <name>` (no flags) defaults BOTH new
    /// knobs off — no re-derive request, no auto-consent (the engine's
    /// intent matrix + consent ladder own the rest).
    #[test]
    fn graph_run_defaults_auto_partition_and_yes_off() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        auto_partition,
                        yes,
                        ..
                    },
            } => {
                assert!(!auto_partition, "no re-derive by default");
                assert!(!yes, "never auto-consent by default");
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// Both flags parse (together — a scripted re-derive-and-persist run).
    #[test]
    fn graph_run_parses_auto_partition_with_yes() {
        let cli =
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--auto-partition", "--yes"])
                .expect("--auto-partition --yes must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        auto_partition,
                        yes,
                        ..
                    },
            } => {
                assert!(auto_partition);
                assert!(yes);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `--single-process` + `--auto-partition` is CONTRADICTORY intent —
    /// rejected at PARSE (`conflicts_with`); the engine's intent matrix
    /// re-rejects it (defense in depth / CLI-engine contract alignment).
    #[test]
    fn graph_run_rejects_single_process_plus_auto_partition() {
        assert!(
            Cli::try_parse_from([
                "cerulion",
                "graph",
                "run",
                "g",
                "--single-process",
                "--auto-partition"
            ])
            .is_err(),
            "--single-process + --auto-partition must be a parse error"
        );
    }

    /// `--yes` composes with `--single-process` (it is simply inert there —
    /// nothing derives, nothing to consent to).
    #[test]
    fn graph_run_yes_with_single_process_parses() {
        let cli =
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--single-process", "--yes"])
                .expect("--single-process --yes must parse (yes is inert on the monolith)");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        single_process,
                        yes,
                        ..
                    },
            } => {
                assert!(single_process);
                assert!(yes);
            }
            _ => panic!("expected Graph::Run"),
        }
    }
}

#[cfg(test)]
mod time_source_flag_tests {
    use super::*;

    /// `graph run <name>` (no flag) defaults to `TimeSource::Real` — the
    /// live-by-default behavior (CER-222 chunk 4a; CER-398 chunk 25b flip).
    #[test]
    fn graph_run_defaults_to_real() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run { time_source, .. },
            } => assert_eq!(time_source, TimeSource::Real, "default must be Real"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `graph run <name> --time-source virtual` parses to `TimeSource::Virtual`.
    #[test]
    fn graph_run_time_source_virtual() {
        let cli =
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--time-source", "virtual"])
                .expect("`graph run g --time-source virtual` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run { time_source, .. },
            } => assert_eq!(time_source, TimeSource::Virtual),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `graph run <name> --time-source external` parses to `TimeSource::External`.
    #[test]
    fn graph_run_time_source_external() {
        let cli =
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--time-source", "external"])
                .expect("`graph run g --time-source external` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run { time_source, .. },
            } => assert_eq!(time_source, TimeSource::External),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// The removed `--sim-clock` flag is now an unknown arg and must fail to
    /// parse (CER-222 chunk 4a hard-removed it, no alias).
    #[test]
    fn graph_run_rejects_removed_sim_clock_flag() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--sim-clock"]).is_err(),
            "--sim-clock was removed and must be rejected"
        );
    }

    /// **CER-1002: `--no-rings` parses, defaults OFF, and is refused beside
    /// `--record`.**
    ///
    /// The default is the whole issue: every multi-process `graph run` now
    /// provisions per-rank scheduler-trace rings, so a Flashback capture of the
    /// DEFAULT run shape can be re-executed (founder ruling 119). A flag that
    /// silently defaulted ON would ship that inert.
    ///
    /// The conflict is not tidiness: a recording without a scheduler trace is not
    /// a recording — the bag would finalize looking complete while
    /// `bag play --resim` refused it. Refused at PARSE, so it costs no SHM, no
    /// workers and no bag.
    #[test]
    fn graph_run_no_rings_parses_defaults_off_and_conflicts_with_record() {
        let flagged = Cli::try_parse_from(["cerulion", "graph", "run", "g", "--no-rings"])
            .expect("`graph run g --no-rings` must parse");
        match flagged.command {
            Commands::Graph {
                action: GraphAction::Run { no_rings, .. },
            } => assert!(no_rings, "the flag must set no_rings = true"),
            _ => panic!("expected Graph::Run"),
        }

        let bare = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match bare.command {
            Commands::Graph {
                action: GraphAction::Run { no_rings, .. },
            } => assert!(
                !no_rings,
                "rings are ON by default — a default-off flag would ship CER-1002 inert, which \
                 is exactly the state it exists to leave"
            ),
            _ => panic!("expected Graph::Run"),
        }

        let Err(err) = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-rings",
            "--record=recordings",
        ]) else {
            panic!("a recording with no scheduler trace is not a recording — it must be refused");
        };
        let text = err.to_string();
        assert!(
            text.contains("--no-rings") && text.contains("--record"),
            "the refusal names BOTH flags, since either one is the operator's to drop: {text}"
        );
    }

    /// **CER-1002:** `--no-rings` composes with the shapes it is INERT on rather
    /// than being refused by them.
    ///
    /// Ruling 95(ii) leaves the wall-gated shapes ringless, so on
    /// `--single-process` and the virtual/external clocks the flag changes only
    /// whether captures are taken. Making it a parse error there would mean an
    /// operator could not put one line in a service file and have it mean the
    /// same thing across every graph they run.
    #[test]
    fn graph_run_no_rings_composes_with_the_shapes_it_is_inert_on() {
        for extra in [
            vec!["--single-process"],
            vec!["--time-source", "virtual"],
            vec!["--time-source", "external"],
        ] {
            let mut argv = vec!["cerulion", "graph", "run", "g", "--no-rings"];
            argv.extend_from_slice(&extra);
            let cli =
                Cli::try_parse_from(&argv).unwrap_or_else(|e| panic!("`{argv:?}` must parse: {e}"));
            match cli.command {
                Commands::Graph {
                    action: GraphAction::Run { no_rings, .. },
                } => assert!(no_rings, "…and the flag still reaches the run: {argv:?}"),
                _ => panic!("expected Graph::Run"),
            }
        }
    }

    /// **`--no-validate` is refused beside `--record`, in every spelling.**
    ///
    /// Founder ruling: a recording must never be made with the schema checks
    /// off. `--no-validate` skips exactly the `schema:` family — a label that
    /// resolves to nothing, contradicts the producing node, or disagrees with
    /// the consumer — and a RUN pays for that later and loudly. A BAG pays for
    /// it permanently and silently: the mislabelled channel is written into the
    /// artifact, so `bag play --resim` refuses a healthy bag, or replays one
    /// under a label nothing ever checked. Refused at PARSE, like the
    /// `--no-rings` conflict above: no shared memory, no workers, no bag.
    ///
    /// Driven in EVERY spelling because `--record` takes an optional
    /// `require_equals` value, so `--record` and `--record=DIR` reach clap as
    /// different shapes, and a conflict is order-independent — a pin that drove
    /// one spelling would leave the others resting on that assumption.
    ///
    /// The anti-tautology half is in the same body: each flag ALONE must still
    /// parse. Without it, an arg accidentally made unparseable in every
    /// combination would satisfy every assertion above.
    #[test]
    fn graph_run_no_validate_conflicts_with_record_in_every_spelling() {
        for argv in [
            vec!["cerulion", "graph", "run", "g", "--no-validate", "--record"],
            vec!["cerulion", "graph", "run", "g", "--record", "--no-validate"],
            vec![
                "cerulion",
                "graph",
                "run",
                "g",
                "--no-validate",
                "--record=recordings",
            ],
            vec![
                "cerulion",
                "graph",
                "run",
                "g",
                "--record=recordings",
                "--no-validate",
            ],
        ] {
            let Err(err) = Cli::try_parse_from(&argv) else {
                panic!("a recording with the schema checks off must be refused: {argv:?}");
            };
            let text = err.to_string();
            assert!(
                text.contains("--no-validate") && text.contains("--record"),
                "the refusal names BOTH flags, since either one is the operator's to \
                 drop ({argv:?}): {text}"
            );
        }

        // ANTI-TAUTOLOGY: each flag alone still parses, so the refusals above
        // are attributable to the COMBINATION.
        for argv in [
            vec!["cerulion", "graph", "run", "g", "--no-validate"],
            vec!["cerulion", "graph", "run", "g", "--record=recordings"],
        ] {
            Cli::try_parse_from(&argv).unwrap_or_else(|e| panic!("`{argv:?}` must parse: {e}"));
        }
    }

    /// CER-440: `graph run g --no-cpu-dma-lock` parses with the flag set.
    #[test]
    fn graph_run_no_cpu_dma_lock_flag_sets_true() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g", "--no-cpu-dma-lock"])
            .expect("`graph run g --no-cpu-dma-lock` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run {
                    no_cpu_dma_lock, ..
                },
            } => assert!(no_cpu_dma_lock, "flag must set no_cpu_dma_lock = true"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-440: bare `graph run g` leaves `no_cpu_dma_lock` false (lock
    /// default-on for the live path).
    #[test]
    fn graph_run_no_cpu_dma_lock_defaults_false() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run {
                    no_cpu_dma_lock, ..
                },
            } => assert!(!no_cpu_dma_lock, "default must be no_cpu_dma_lock = false"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-440: `--no-cpu-dma-lock` coexists with `--time-source virtual`
    /// (the flag is independent of the clock selection).
    #[test]
    fn graph_run_no_cpu_dma_lock_coexists_with_virtual() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-cpu-dma-lock",
            "--time-source",
            "virtual",
        ])
        .expect("`graph run g --no-cpu-dma-lock --time-source virtual` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        no_cpu_dma_lock,
                        time_source,
                        ..
                    },
            } => {
                assert!(no_cpu_dma_lock);
                assert_eq!(time_source, TimeSource::Virtual);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-440: `--no-cpu-dma-lock` coexists with `--time-source external`
    /// — the opt-out is honored even on the live `External` clock (whose
    /// default would otherwise acquire the lock).
    #[test]
    fn graph_run_no_cpu_dma_lock_coexists_with_external() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-cpu-dma-lock",
            "--time-source",
            "external",
        ])
        .expect("`graph run g --no-cpu-dma-lock --time-source external` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        no_cpu_dma_lock,
                        time_source,
                        ..
                    },
            } => {
                assert!(no_cpu_dma_lock);
                assert_eq!(time_source, TimeSource::External);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-443: `graph run g --no-monitor-wait` parses with the flag set.
    #[test]
    fn graph_run_no_monitor_wait_flag_sets_true() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g", "--no-monitor-wait"])
            .expect("`graph run g --no-monitor-wait` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run {
                    no_monitor_wait, ..
                },
            } => assert!(no_monitor_wait, "flag must set no_monitor_wait = true"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-443: bare `graph run g` leaves `no_monitor_wait` false (park
    /// auto-on for the live path where the primitive exists).
    #[test]
    fn graph_run_no_monitor_wait_defaults_false() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run {
                    no_monitor_wait, ..
                },
            } => assert!(!no_monitor_wait, "default must be no_monitor_wait = false"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-443: `--no-monitor-wait` coexists with `--time-source virtual` (the
    /// flag is independent of the clock selection; the park is live-only so it
    /// is a no-op under virtual, but the flag must still parse).
    #[test]
    fn graph_run_no_monitor_wait_coexists_with_virtual() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-monitor-wait",
            "--time-source",
            "virtual",
        ])
        .expect("`graph run g --no-monitor-wait --time-source virtual` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        no_monitor_wait,
                        time_source,
                        ..
                    },
            } => {
                assert!(no_monitor_wait);
                assert_eq!(time_source, TimeSource::Virtual);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-443: `--no-monitor-wait` coexists with `--time-source external` —
    /// the opt-out is honored on the live `External` clock (whose default would
    /// otherwise arm the park where the primitive exists).
    #[test]
    fn graph_run_no_monitor_wait_coexists_with_external() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-monitor-wait",
            "--time-source",
            "external",
        ])
        .expect("`graph run g --no-monitor-wait --time-source external` must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        no_monitor_wait,
                        time_source,
                        ..
                    },
            } => {
                assert!(no_monitor_wait);
                assert_eq!(time_source, TimeSource::External);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-508: the hidden `graph run-worker --plan <file>` verb parses and
    /// captures the plan path. Clap renames the `RunWorker` variant to the
    /// kebab-case `run-worker` subcommand.
    #[test]
    fn graph_run_worker_parses_plan_path() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run-worker",
            "--plan",
            "/tmp/cer508_worker_P0.json",
        ])
        .expect("`graph run-worker --plan <file>` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::RunWorker { plan },
            } => assert_eq!(
                plan,
                std::path::PathBuf::from("/tmp/cer508_worker_P0.json"),
                "the --plan path must be captured verbatim"
            ),
            _ => panic!("expected Graph::RunWorker"),
        }
    }

    /// CER-508: `graph run-worker` WITHOUT `--plan` is rejected by clap (the
    /// plan file is required — a worker cannot run without its plan).
    #[test]
    fn graph_run_worker_requires_plan() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run-worker"]).is_err(),
            "`graph run-worker` must require --plan"
        );
    }

    /// CER-443: `--no-monitor-wait` and `--no-cpu-dma-lock` coexist (independent
    /// opt-outs; both set true together).
    #[test]
    fn graph_run_no_monitor_wait_and_no_cpu_dma_lock_coexist() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "graph",
            "run",
            "g",
            "--no-monitor-wait",
            "--no-cpu-dma-lock",
        ])
        .expect("both opt-out flags must parse together");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        no_monitor_wait,
                        no_cpu_dma_lock,
                        ..
                    },
            } => {
                assert!(no_monitor_wait);
                assert!(no_cpu_dma_lock);
            }
            _ => panic!("expected Graph::Run"),
        }
    }

    /// CER-349 chunk E (+ review finding #3): `--record` clap semantics with
    /// `require_equals` — absent → `None`, `--record` alone → the default dir,
    /// `--record=DIR` → `DIR`, and CRUCIALLY `graph run --record demo` parses
    /// `demo` as the GRAPH NAME (the optional value can no longer greedily
    /// swallow the required positional). All four orderings pinned.
    #[test]
    fn graph_run_record_flag_orderings() {
        let parse = |args: &[&str]| -> (String, Option<String>) {
            match Cli::try_parse_from(args).expect("parse").command {
                Commands::Graph {
                    action: GraphAction::Run { name, record, .. },
                } => (name, record),
                _ => panic!("expected Graph::Run"),
            }
        };
        // (1) no --record → None.
        assert_eq!(
            parse(&["cerulion", "graph", "run", "g"]),
            ("g".to_string(), None),
            "no --record → recording off"
        );
        // (2) bare --record after the positional → the default dir.
        assert_eq!(
            parse(&["cerulion", "graph", "run", "g", "--record"]),
            ("g".to_string(), Some("recordings".to_string())),
            "--record alone → the default `recordings` dir"
        );
        // (3) --record=DIR → DIR.
        assert_eq!(
            parse(&["cerulion", "graph", "run", "g", "--record=bags"]),
            ("g".to_string(), Some("bags".to_string())),
            "--record=DIR → DIR"
        );
        // (4) THE finding-#3 footgun ordering: `--record` BEFORE the graph
        // name must leave the positional intact (record → default dir).
        assert_eq!(
            parse(&["cerulion", "graph", "run", "--record", "demo"]),
            ("demo".to_string(), Some("recordings".to_string())),
            "--record before the positional must NOT swallow the graph name"
        );
        // require_equals: the space-separated value form is gone — a token
        // after --record is a positional, so `g --record bags2` now has TWO
        // positionals and is a loud parse error (not a silent record=bags2).
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--record", "bags2"]).is_err(),
            "space-separated --record value must be rejected (use --record=DIR)"
        );
    }

    /// CER-349 chunk E item 2 (+ pass-2 finding #8): `--record-env` parses its
    /// three modes, defaults to `allowlist` (the secret-safe default), and
    /// REQUIRES `--record` — a lone `--record-env` is a loud parse error, never
    /// silently accepted-and-ignored.
    #[test]
    fn graph_run_record_env_three_modes_default_allowlist_requires_record() {
        let env_of = |args: &[&str]| -> RecordEnv {
            match Cli::try_parse_from(args).expect("parse").command {
                Commands::Graph {
                    action: GraphAction::Run { record_env, .. },
                } => record_env,
                _ => panic!("expected Graph::Run"),
            }
        };
        // Default (no flags) stays the secret-safe allowlist.
        assert_eq!(
            env_of(&["cerulion", "graph", "run", "g"]),
            RecordEnv::Allowlist,
            "default must be the secret-safe allowlist"
        );
        // With --record, all three modes parse.
        assert_eq!(
            env_of(&[
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-env",
                "all"
            ]),
            RecordEnv::All
        );
        assert_eq!(
            env_of(&[
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-env",
                "none"
            ]),
            RecordEnv::None
        );
        assert_eq!(
            env_of(&[
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-env",
                "allowlist"
            ]),
            RecordEnv::Allowlist
        );
        // Pass-2 finding #8: --record-env WITHOUT --record is a loud parse
        // error (the loud-over-silent norm; a user typing it expected
        // recording side effects).
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--record-env", "all"]).is_err(),
            "--record-env without --record must be a loud parse error"
        );
        assert!(
            Cli::try_parse_from([
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-env",
                "bogus"
            ])
            .is_err(),
            "unknown mode rejected"
        );
    }

    /// CER-349 chunk E addendum: `--record-cpu` clap arms — a core id, the
    /// literal `none`, requires-`--record` rejection, and bad-value rejection.
    /// Flag absent maps to the engine's AUTO.
    #[test]
    fn graph_run_record_cpu_arms() {
        use cerulion_cli_engine::graph_cmd::RecordCpu;
        let cpu_of = |args: &[&str]| -> RecordCpu {
            match Cli::try_parse_from(args).expect("parse").command {
                Commands::Graph {
                    action: GraphAction::Run { record_cpu, .. },
                } => record_cpu_mode(record_cpu),
                _ => panic!("expected Graph::Run"),
            }
        };
        // Absent → AUTO.
        assert_eq!(
            cpu_of(&["cerulion", "graph", "run", "g", "--record"]),
            RecordCpu::Auto,
            "flag absent = auto placement"
        );
        // Explicit core.
        assert_eq!(
            cpu_of(&[
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-cpu",
                "7"
            ]),
            RecordCpu::Core(7)
        );
        // Literal none → float.
        assert_eq!(
            cpu_of(&[
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-cpu",
                "none"
            ]),
            RecordCpu::None
        );
        // requires = "record": a lone --record-cpu is a loud parse error.
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--record-cpu", "3"]).is_err(),
            "--record-cpu without --record must be a loud parse error"
        );
        // Bad value: neither a core id nor `none`.
        assert!(
            Cli::try_parse_from([
                "cerulion",
                "graph",
                "run",
                "g",
                "--record",
                "--record-cpu",
                "fastest"
            ])
            .is_err(),
            "a non-numeric non-`none` value must be rejected"
        );
    }

    /// CER-349 chunk E item 1: the folded `cerulion bagd` subcommand parses
    /// (attach-mode flags reach BagdArgs through the embedding).
    #[cfg(unix)]
    #[test]
    fn bagd_subcommand_parses_attach_mode_args() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "bagd",
            "--out",
            "/tmp/x.mcap",
            "--topic",
            "/a",
            "--topic",
            "/b",
        ])
        .expect("`cerulion bagd` must parse");
        match cli.command {
            Commands::Bagd(args) => {
                assert_eq!(args.out, std::path::PathBuf::from("/tmp/x.mcap"));
                assert_eq!(args.topics, vec!["/a".to_string(), "/b".to_string()]);
            }
            _ => panic!("expected Commands::Bagd"),
        }
        // --topic and --topics-json stay mutually exclusive through the fold.
        assert!(
            Cli::try_parse_from([
                "cerulion",
                "bagd",
                "--out",
                "/tmp/x.mcap",
                "--topic",
                "/a",
                "--topics-json",
                "/tmp/t.json",
            ])
            .is_err(),
            "--topic + --topics-json must conflict"
        );
    }

    // ─── CER-508 chunks 6+7: --peer-loss / --single-process / --trace-limit ───

    /// Helper: parse args and return the `GraphAction::Run` fields under test.
    fn parse_run(args: &[&str]) -> (Option<PeerLoss>, bool, u64, TimeSource) {
        let mut full = vec!["cerulion", "graph", "run", "g"];
        full.extend_from_slice(args);
        let cli = Cli::try_parse_from(full).expect("graph run args must parse");
        match cli.command {
            Commands::Graph {
                action:
                    GraphAction::Run {
                        peer_loss,
                        single_process,
                        trace_limit,
                        time_source,
                        ..
                    },
            } => (peer_loss, single_process, trace_limit, time_source),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `--peer-loss fail` / `--peer-loss continue` parse to the mirror enum.
    #[test]
    fn graph_run_peer_loss_parses_fail_and_continue() {
        let (p, ..) = parse_run(&["--peer-loss", "fail"]);
        assert_eq!(p, Some(PeerLoss::Fail));
        let (p, ..) = parse_run(&["--peer-loss", "continue"]);
        assert_eq!(p, Some(PeerLoss::Continue));
    }

    /// Bare `graph run g` leaves `--peer-loss` at `None` — the engine then
    /// consults the hidden env seam / the Continue default (`resolve_peer_loss`).
    /// The `None`-vs-`Some(Continue)` distinction is the PRECEDENCE contract:
    /// only an EXPLICIT flag beats the env seam.
    #[test]
    fn graph_run_peer_loss_defaults_to_none() {
        let (p, ..) = parse_run(&[]);
        assert_eq!(p, None, "absent flag must be None (env seam stays live)");
    }

    /// A garbage `--peer-loss` value is rejected at parse (value_enum).
    #[test]
    fn graph_run_peer_loss_garbage_rejected() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--peer-loss", "banana"])
                .is_err(),
            "--peer-loss only accepts continue|fail"
        );
    }

    /// `--single-process` parses (default false).
    #[test]
    fn graph_run_single_process_flag() {
        let (_, sp, ..) = parse_run(&["--single-process"]);
        assert!(sp);
        let (_, sp, ..) = parse_run(&[]);
        assert!(!sp, "default must be single_process = false");
    }

    /// `--trace-limit 50000` parses; the default is the engine's
    /// PRODUCTION_TRACE_LIMIT (single source — 100_000 today).
    #[test]
    fn graph_run_trace_limit_parses_and_defaults() {
        let (_, _, tl, _) = parse_run(&["--trace-limit", "50000"]);
        assert_eq!(tl, 50_000);
        let (_, _, tl, _) = parse_run(&[]);
        assert_eq!(
            tl,
            cerulion_cli_engine::graph_cmd::PRODUCTION_TRACE_LIMIT as u64,
            "default must be the engine's production trace cap"
        );
    }

    /// `--trace-limit 0` is REJECTED at parse (value_parser range 1..): the
    /// trace ring is a bounded observability window — no unbounded escape hatch.
    #[test]
    fn graph_run_trace_limit_zero_rejected() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--trace-limit", "0"]).is_err(),
            "--trace-limit 0 (unbounded) must be rejected at parse"
        );
    }

    /// The chunks-6+7 flags compose with `--time-source` (independent knobs;
    /// the External × multi-process rejection is an ENGINE dispatch decision,
    /// not a parse-level conflict).
    #[test]
    fn graph_run_new_flags_combo_with_time_source() {
        let (p, sp, tl, ts) = parse_run(&[
            "--peer-loss",
            "fail",
            "--single-process",
            "--trace-limit",
            "2000",
            "--time-source",
            "virtual",
        ]);
        assert_eq!(p, Some(PeerLoss::Fail));
        assert!(sp);
        assert_eq!(tl, 2_000);
        assert_eq!(ts, TimeSource::Virtual);
    }

    /// The PeerLoss mirror maps 1:1 onto the engine enum (both directions of
    /// the two-variant surface).
    #[test]
    fn peer_loss_mirror_maps_to_engine_enum() {
        assert_eq!(
            EnginePeerLossPolicy::from(PeerLoss::Continue),
            EnginePeerLossPolicy::Continue
        );
        assert_eq!(
            EnginePeerLossPolicy::from(PeerLoss::Fail),
            EnginePeerLossPolicy::Fail
        );
    }
}

#[cfg(test)]
mod topic_list_network_flag_tests {
    use super::*;

    /// CER-752: bare `topic list` (no flags) defaults to AUTOMAGIC remote
    /// discovery — `no_network` is false, no explicit locators. The `--network`
    /// flag is GONE (remote discovery is on by default).
    #[test]
    fn topic_list_bare_defaults_to_network_on() {
        let cli = Cli::try_parse_from(["cerulion", "topic", "list"])
            .expect("bare `topic list` must parse");
        match cli.command {
            Commands::Topic {
                action:
                    TopicAction::List {
                        no_network,
                        connect,
                        listen,
                        scan,
                    },
            } => {
                assert!(!no_network, "bare list must default to remote-discovery ON");
                assert!(connect.is_empty());
                assert!(listen.is_empty());
                assert!(
                    !scan,
                    "CER-760: the opt-in subnet sweep must be OFF by default"
                );
            }
            _ => panic!("expected Topic::List"),
        }
    }

    /// CER-760 rung 4: `topic list --scan` opts into the unicast subnet sweep;
    /// the flag is OFF on a bare `topic list` (pinned above). This is the sole
    /// producer of a `true` scan — the structural gate against the sweep firing
    /// unopted.
    #[test]
    fn topic_list_scan_flag_parses() {
        let cli = Cli::try_parse_from(["cerulion", "topic", "list", "--scan"])
            .expect("`topic list --scan` must parse");
        match cli.command {
            Commands::Topic {
                action: TopicAction::List { scan, .. },
            } => assert!(scan, "--scan must set scan=true"),
            _ => panic!("expected Topic::List"),
        }
    }

    /// CER-752: `topic list --no-network` skips the remote query.
    #[test]
    fn topic_list_no_network_flag_parses() {
        let cli = Cli::try_parse_from(["cerulion", "topic", "list", "--no-network"])
            .expect("`topic list --no-network` must parse");
        match cli.command {
            Commands::Topic {
                action: TopicAction::List { no_network, .. },
            } => assert!(no_network, "--no-network must set no_network=true"),
            _ => panic!("expected Topic::List"),
        }
    }

    /// CER-752: the removed `--network` flag is now an UNKNOWN argument (killed,
    /// not deprecated).
    #[test]
    fn topic_list_removed_network_flag_is_unknown() {
        assert!(
            Cli::try_parse_from(["cerulion", "topic", "list", "--network"]).is_err(),
            "the `--network` flag was removed — it must be an unknown argument now"
        );
    }

    /// CER-752: `--connect` / `--listen` accumulate in order and NO LONGER
    /// require any gating flag (they add to the default scouting session).
    #[test]
    fn topic_list_repeatable_locators_accumulate_without_gate() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "topic",
            "list",
            "--connect",
            "tcp/192.168.123.99:7447",
            "--connect",
            "tcp/192.168.123.100:7447",
            "--listen",
            "tcp/0.0.0.0:7447",
        ])
        .expect("repeated locators must parse without a gating flag");
        match cli.command {
            Commands::Topic {
                action:
                    TopicAction::List {
                        no_network,
                        connect,
                        listen,
                        scan,
                    },
            } => {
                assert!(!no_network);
                assert!(!scan, "no --scan ⇒ scan stays off");
                assert_eq!(
                    connect,
                    vec![
                        "tcp/192.168.123.99:7447".to_string(),
                        "tcp/192.168.123.100:7447".to_string()
                    ]
                );
                assert_eq!(listen, vec!["tcp/0.0.0.0:7447".to_string()]);
            }
            _ => panic!("expected Topic::List"),
        }
    }

    /// CER-752: a bare `--connect` (no gating flag) now parses fine — the
    /// `requires = "network"` coupling is GONE.
    #[test]
    fn topic_list_connect_alone_parses() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "topic",
            "list",
            "--connect",
            "tcp/192.168.123.99:7447",
        ])
        .expect("`--connect` alone must parse now that the gate is gone");
        match cli.command {
            Commands::Topic {
                action: TopicAction::List { connect, .. },
            } => assert_eq!(connect, vec!["tcp/192.168.123.99:7447".to_string()]),
            _ => panic!("expected Topic::List"),
        }
    }
}

/// CER-1257: verbs this CLI has REMOVED must not parse at all.
///
/// The precedent is `topic_list_removed_network_flag_is_unknown` above: a
/// killed surface is killed, not deprecated, and the pin that proves it is a
/// parse REFUSAL rather than an assertion about a message. A removed verb that
/// still parses is the failure this module exists to catch — clap answers a
/// declared variant itself, so the surface can go on existing (in `--help`
/// classification, in dispatch, in tab completion) long after the behaviour
/// behind it is gone.
#[cfg(test)]
mod removed_verb_tests {
    use super::*;

    /// CER-1257: `cerulion replay` is an unknown subcommand.
    ///
    /// CER-982 (founder ruling 65) removed the BEHAVIOUR and left a hidden
    /// `Replay` variant behind purely to DIAGNOSE the removal — it ran nothing
    /// and always exited 2, but it still parsed. The founder ruled on
    /// 2026-08-11 that the diagnosis goes too, so nothing named `replay` may
    /// reach the parser: re-execution is `cerulion bag play <bag> --resim all
    /// [--verify]`, and the old spelling now dies as clap's plain
    /// `unrecognized subcommand 'replay'` (exit 2 — the same code the tombstone
    /// returned, so no caller's contract moves).
    ///
    /// BOTH arms are load-bearing. The removed variant was declared
    /// `trailing_var_arg` + `allow_hyphen_values` specifically so that every
    /// old invocation shape reached it, so a bare bag path and a hyphenated
    /// flag shape entered by DIFFERENT clap paths — a half-deleted declaration
    /// (variant gone, or an `external_subcommand` catch-all added later) can
    /// leave one still parsing while the other refuses.
    ///
    /// BASELINE-VERIFIED: run against the pre-removal tree, each arm FAILS —
    /// `replay bag.mcap` and `replay bag.mcap --tolerance t.yaml` both parsed
    /// into the tombstone variant.
    #[test]
    fn removed_replay_verb_is_an_unknown_subcommand() {
        assert!(
            Cli::try_parse_from(["cerulion", "replay", "bag.mcap"]).is_err(),
            "`cerulion replay` was removed — it must be an unrecognized \
             subcommand now, not a parse into a tombstone variant"
        );
        assert!(
            Cli::try_parse_from(["cerulion", "replay", "bag.mcap", "--tolerance", "t.yaml"])
                .is_err(),
            "an old `cerulion replay` flag shape must be unrecognized too — the \
             removed variant swallowed trailing hyphenated args, so this arm is \
             what catches a half-deleted declaration"
        );
    }
}

#[cfg(test)]
mod graph_run_network_flag_tests {
    use super::*;

    /// CER-752: bare `graph run g` carries no kill-switch — the run takes the
    /// CER-752 default (an enabled `network:` block ⇒ Strict; no block ⇒
    /// Permissive).
    #[test]
    fn graph_run_without_network_flag_parses_none() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g"])
            .expect("bare `graph run g` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run { network, .. },
            } => assert!(network.is_none(), "absent flag must parse to None"),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `graph run g --network off` parses to `Some("off")` — the
    /// kill-switch.
    #[test]
    fn graph_run_network_off_parses() {
        let cli = Cli::try_parse_from(["cerulion", "graph", "run", "g", "--network", "off"])
            .expect("`graph run g --network off` must parse");
        match cli.command {
            Commands::Graph {
                action: GraphAction::Run { network, .. },
            } => assert_eq!(network.as_deref(), Some("off")),
            _ => panic!("expected Graph::Run"),
        }
    }

    /// `off` is the ONLY accepted value: `--network on` is a loud parse
    /// error listing the valid value (there is no `--network on` — an
    /// absent flag already means "honor the YAML").
    #[test]
    fn graph_run_network_on_rejected_at_parse() {
        let Err(err) = Cli::try_parse_from(["cerulion", "graph", "run", "g", "--network", "on"])
        else {
            panic!("--network on must be rejected at parse (only `off` exists)");
        };
        let msg = err.to_string();
        assert!(
            msg.contains("off"),
            "the rejection must list the valid value `off`; got: {msg}"
        );
    }

    /// A bare `--network` without a value is a parse error (the flag
    /// requires its MODE value — clap enforces it).
    #[test]
    fn graph_run_network_without_value_rejected_at_parse() {
        assert!(
            Cli::try_parse_from(["cerulion", "graph", "run", "g", "--network"]).is_err(),
            "bare --network (no value) must be a parse error"
        );
    }

    /// CER-752: `node run` mirrors `graph run`'s kill-switch — bare parse =
    /// None (the permissive default applies), `--network off` = Some("off"),
    /// any other value rejected at parse.
    #[test]
    fn node_run_network_off_mirrors_graph_run() {
        let cli = Cli::try_parse_from(["cerulion", "node", "run", "n"])
            .expect("bare `node run n` must parse");
        match cli.command {
            Commands::Node {
                action: NodeAction::Run { network, .. },
            } => assert!(
                network.is_none(),
                "absent flag = the permissive default applies"
            ),
            _ => panic!("expected Node::Run"),
        }

        let cli = Cli::try_parse_from(["cerulion", "node", "run", "n", "--network", "off"])
            .expect("`node run n --network off` must parse");
        match cli.command {
            Commands::Node {
                action: NodeAction::Run { network, .. },
            } => assert_eq!(network.as_deref(), Some("off")),
            _ => panic!("expected Node::Run"),
        }

        assert!(
            Cli::try_parse_from(["cerulion", "node", "run", "n", "--network", "on"]).is_err(),
            "`off` is the only accepted value for node run --network"
        );
    }
}

#[cfg(test)]
mod ros_attach_flag_tests {
    use super::*;

    /// The removed `ros` family's stub swallows EVERY old spelling — help
    /// flags included. Entelligence MAJOR on #727: clap answers `--help`
    /// BEFORE returning a parsed command, so without `disable_help_flag` on
    /// the stub, `cerulion ros --help` served clap's help page (exit 0)
    /// instead of the migration message `main` promises for the removed
    /// family. MUTANT: dropping `disable_help_flag` makes every arm below
    /// fail with a `DisplayHelp` error instead of a parsed stub.
    #[test]
    fn removed_ros_stub_swallows_help_flags_instead_of_serving_clap_help() {
        for argv in [
            vec!["cerulion", "ros", "--help"],
            vec!["cerulion", "ros", "-h"],
            vec!["cerulion", "ros", "attach", "--help"],
            vec!["cerulion", "ros", "attach", "--iface", "10.0.0.7"],
        ] {
            let expected_args: Vec<String> = argv[2..].iter().map(|s| s.to_string()).collect();
            let cli = match Cli::try_parse_from(&argv) {
                Ok(cli) => cli,
                Err(e) => panic!(
                    "`{}` must parse into the migration stub, never clap's \
                     own help/error (kind {:?})",
                    argv.join(" "),
                    e.kind()
                ),
            };
            match cli.command {
                Commands::Ros { args } => assert_eq!(
                    args,
                    expected_args,
                    "the stub must swallow the full old argv for `{}`",
                    argv.join(" ")
                ),
                _ => panic!("expected the Commands::Ros stub"),
            }
        }
    }

    /// CER-722 review fix 2 (pre-fix FAILING): non-finite / non-positive
    /// `ros2 attach --timeout` values are rejected AT PARSE TIME with a
    /// message naming the constraint — pre-fix they parsed fine and were
    /// silently coerced to 5.0 downstream.
    #[test]
    fn ros_attach_rejects_nonpositive_and_nonfinite_timeouts_at_parse() {
        // Pass-2 adds the OVERFLOW class ("2e19": finite + positive, but
        // `Duration::from_secs_f64` panics above ~5.8e11 s — pre-pass-2 it
        // parsed fine and panicked at runtime) and the bound edge ("3601").
        for bad in ["0", "-3", "nan", "inf", "-inf", "0.0", "2e19", "3601"] {
            let arg = format!("--timeout={bad}");
            // let-else (the `graph_run_network_on_rejected_at_parse` house
            // pattern) — `expect_err` would require `Cli: Debug`, which the
            // clap structs deliberately do not derive.
            let Err(err) = Cli::try_parse_from([
                "cerulion",
                "ros2",
                "attach",
                "--iface",
                "192.168.123.18",
                arg.as_str(),
            ]) else {
                panic!("--timeout={bad} must be rejected at parse");
            };
            let msg = err.to_string();
            assert!(
                msg.contains("finite number of seconds > 0 and <= 3600"),
                "the rejection must name BOTH bounds; got: {msg}"
            );
        }
    }

    /// The happy `ros2 attach` parse: a valid `--timeout` lands verbatim, the
    /// default is 5.0, and `--iface` is REQUIRED.
    #[test]
    fn ros_attach_timeout_parses_and_iface_is_required() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "ros2",
            "attach",
            "--iface",
            "192.168.123.18",
            "--timeout",
            "2.5",
        ])
        .expect("valid ros2 attach must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Attach { timeout, iface, .. },
            } => {
                assert_eq!(timeout, 2.5);
                assert_eq!(iface.to_string(), "192.168.123.18");
            }
            _ => panic!("expected Ros2::Attach"),
        }

        // Default window.
        let cli = Cli::try_parse_from(["cerulion", "ros2", "attach", "--iface", "10.0.0.7"])
            .expect("iface-only ros2 attach must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Attach { timeout, .. },
            } => assert_eq!(timeout, 5.0),
            _ => panic!("expected Ros2::Attach"),
        }

        // Pass-2: the upper bound is INCLUSIVE — exactly 3600 s parses.
        let cli = Cli::try_parse_from([
            "cerulion",
            "ros2",
            "attach",
            "--iface",
            "10.0.0.7",
            "--timeout",
            "3600",
        ])
        .expect("--timeout 3600 (the inclusive bound) must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Attach { timeout, .. },
            } => assert_eq!(timeout, 3600.0),
            _ => panic!("expected Ros2::Attach"),
        }

        // --iface is required (the load-bearing multi-homed-discovery knob).
        assert!(
            Cli::try_parse_from(["cerulion", "ros2", "attach"]).is_err(),
            "ros2 attach without --iface must be a parse error"
        );
    }

    /// CER-905 (founder ruling pt53), the INVERSION of the CER-725 finding-6
    /// pin: `ros2 attach` stages no visualization node on the robot, so the
    /// `--no-viz` opt-out is REMOVED rather than defaulted-on — a flag that
    /// skips staging is a misleading name for a no-op, and this repo kills
    /// misleading user-facing surface instead of re-documenting it.
    ///
    /// Removing a CLI flag is a user-visible breaking change, so the contract
    /// is that a stale invocation FAILS LOUDLY (clap's unknown-argument error)
    /// rather than being silently accepted and ignored. Both the removed flag
    /// and the never-shipped positive alias are pinned. Clap-level unit, no
    /// process spawn.
    #[test]
    fn ros_attach_rejects_the_removed_no_viz_flag() {
        // Anti-tautology control: the same invocation WITHOUT the flag parses,
        // so the rejection below is attributable to `--no-viz` alone and not to
        // some unrelated breakage in the `ros2 attach` argument set.
        let cli = Cli::try_parse_from(["cerulion", "ros2", "attach", "--iface", "10.0.0.7"])
            .expect("iface-only ros2 attach must parse");
        assert!(
            matches!(
                cli.command,
                Commands::Ros2 {
                    action: Ros2Action::Attach { .. }
                }
            ),
            "expected Ros2::Attach"
        );

        for removed in ["--no-viz", "--viz"] {
            // `Cli` is not `Debug`, so unwrap the Result by hand rather than
            // with `expect_err`.
            let err = match Cli::try_parse_from([
                "cerulion", "ros2", "attach", "--iface", "10.0.0.7", removed,
            ]) {
                Ok(_) => panic!(
                    "`{removed}` must be a LOUD parse error, never silently accepted \
                     (a removed flag that still parses would be ignored in silence)"
                ),
                Err(e) => e,
            };
            assert_eq!(
                err.kind(),
                clap::error::ErrorKind::UnknownArgument,
                "`{removed}` must fail as an unknown argument, got {:?}",
                err.kind()
            );
        }
    }

    /// CER-825 chunk 4: the viz flags parse with the daemon-client defaults —
    /// `--no-spawn`/`--serve-web`/`--detach` default OFF. `--release` and
    /// `--memory-limit` were REMOVED (the daemon is a prebuilt binary that owns
    /// its own build + the hosted-proxy memory; per-run verb flags could never
    /// reconfigure the shared long-lived daemon).
    #[test]
    fn viz_flags_parse_and_default() {
        // Defaults: topics-only ⇒ no_spawn/serve_web/detach off.
        let cli = Cli::try_parse_from(["cerulion", "viz", "/tf"]).expect("viz /tf must parse");
        match cli.command {
            Commands::Viz {
                topics,
                robot,
                connect,
                listen,
                no_spawn,
                serve_web,
                detach,
            } => {
                assert_eq!(topics, vec!["/tf".to_string()]);
                // CER-794 chunk 5: remote flags default empty/None.
                assert!(robot.is_none(), "--robot defaults None");
                assert!(
                    connect.is_empty() && listen.is_empty(),
                    "--connect/--listen default empty"
                );
                assert!(!no_spawn && !serve_web && !detach, "flags default OFF");
            }
            _ => panic!("expected Commands::Viz"),
        }
        // CER-825 chunk 4: --detach parses (the Studio handoff — attach + return).
        let cli = Cli::try_parse_from(["cerulion", "viz", "/tf", "--detach"])
            .expect("viz --detach parses");
        match cli.command {
            Commands::Viz { detach, .. } => assert!(detach, "--detach sets detach"),
            _ => panic!("expected Commands::Viz"),
        }
        // The remaining spawn flags parse together.
        let cli =
            Cli::try_parse_from(["cerulion", "viz", "/go2/cloud", "--serve-web", "--no-spawn"])
                .expect("viz with the spawn flags must parse");
        match cli.command {
            Commands::Viz {
                no_spawn,
                serve_web,
                ..
            } => assert!(no_spawn && serve_web),
            _ => panic!("expected Commands::Viz"),
        }
        // The REMOVED flags are rejected as UNKNOWN arguments (not merely some
        // other parse error — the flag is gone from the contract, so clap must not
        // recognize it at all).
        // (`Cli` is not `Debug`, so unwrap the ERROR via `.err()` rather than
        // `expect_err`, which would need the Ok type to be `Debug`.)
        let released = Cli::try_parse_from(["cerulion", "viz", "/tf", "--release"])
            .err()
            .expect("--release was removed");
        assert_eq!(
            released.kind(),
            clap::error::ErrorKind::UnknownArgument,
            "--release rejects as an UNKNOWN flag, got {:?}",
            released.kind()
        );
        let mem = Cli::try_parse_from(["cerulion", "viz", "/tf", "--memory-limit", "256MB"])
            .err()
            .expect("--memory-limit was removed");
        assert_eq!(
            mem.kind(),
            clap::error::ErrorKind::UnknownArgument,
            "--memory-limit rejects as an UNKNOWN flag, got {:?}",
            mem.kind()
        );
    }

    /// CER-794 D1 chunk 5: the remote-robot flags parse — `--robot NAME` plus
    /// repeatable `--connect`/`--listen` locators (mirroring `topic list`), with a
    /// pinned `TOPIC=SCHEMA` positional.
    #[test]
    fn viz_remote_robot_flags_parse() {
        let cli = Cli::try_parse_from([
            "cerulion",
            "viz",
            "--robot",
            "go2",
            "--connect",
            "tcp/192.168.123.99:7683",
            "--connect",
            "tcp/10.0.0.9:7447",
            "--listen",
            "tcp/0.0.0.0:7447",
            "/utlidar/cloud=sensor_msgs/PointCloud2",
        ])
        .expect("viz --robot with locators must parse");
        match cli.command {
            Commands::Viz {
                topics,
                robot,
                connect,
                listen,
                ..
            } => {
                assert_eq!(robot.as_deref(), Some("go2"));
                assert_eq!(
                    connect,
                    vec!["tcp/192.168.123.99:7683", "tcp/10.0.0.9:7447"],
                    "--connect is repeatable"
                );
                assert_eq!(listen, vec!["tcp/0.0.0.0:7447"]);
                assert_eq!(topics, vec!["/utlidar/cloud=sensor_msgs/PointCloud2"]);
            }
            _ => panic!("expected Commands::Viz"),
        }
    }

    /// CER-794 D1 c5 review (FIX 2): `--connect`/`--listen` `requires = "robot"` —
    /// the local viz path never reads these locators, so passing one WITHOUT
    /// `--robot` is a LOUD clap parse error (naming `--robot`), not a silent drop.
    #[test]
    fn viz_connect_or_listen_without_robot_is_a_loud_error() {
        // let-else (the house pattern) — `expect_err` would require `Cli: Debug`,
        // which the clap structs deliberately do not derive.
        // --connect without --robot: rejected.
        let Err(err) = Cli::try_parse_from([
            "cerulion",
            "viz",
            "/tf",
            "--connect",
            "tcp/192.168.123.99:7683",
        ]) else {
            panic!("--connect without --robot must be a parse error");
        };
        let msg = err.to_string();
        assert!(
            msg.contains("--robot"),
            "the error must name the required --robot flag:\n{msg}"
        );

        // --listen without --robot: rejected too.
        let Err(err) = Cli::try_parse_from(["cerulion", "viz", "--listen", "tcp/0.0.0.0:7447"])
        else {
            panic!("--listen without --robot must be a parse error");
        };
        assert!(
            err.to_string().contains("--robot"),
            "the error must name the required --robot flag:\n{err}"
        );

        // Control: WITH --robot both parse fine (the gate is satisfied).
        Cli::try_parse_from([
            "cerulion",
            "viz",
            "--robot",
            "go2",
            "--connect",
            "tcp/192.168.123.99:7683",
            "--listen",
            "tcp/0.0.0.0:7447",
        ])
        .expect("--connect/--listen WITH --robot must parse");
    }

    /// CER-794 D1 c5 review (FIX 2): the `--connect`/`--listen` help now states the
    /// requirement is ENFORCED (`Requires --robot`) and no longer the misleading
    /// pre-fix "Ignored without --robot" (which described a silent drop that is now
    /// a hard parse error).
    #[test]
    fn viz_connect_listen_help_states_requires_robot_not_ignored() {
        use clap::CommandFactory;
        let mut cmd = Cli::command();
        let viz = cmd
            .find_subcommand_mut("viz")
            .expect("the viz subcommand exists");
        let help = viz.render_long_help().to_string();
        let flat: String = help.split_whitespace().collect::<Vec<_>>().join(" ");
        assert!(
            flat.contains("Requires `--robot`"),
            "connect/listen help must state the requirement is enforced:\n{help}"
        );
        assert!(
            !flat.contains("Ignored without `--robot`"),
            "the misleading 'Ignored without --robot' footnote must be gone:\n{help}"
        );
    }

    /// CER-794 D1 c3 review (FIX 4): the `--serve-web` help enumerates EVERY
    /// condition under which it is ignored — crucially including `--no-spawn`,
    /// which the pre-fix text omitted. Rendered help wraps to terminal width, so
    /// whitespace is collapsed before matching (robust against clap's reflow).
    #[test]
    fn viz_serve_web_help_names_all_ignore_conditions_including_no_spawn() {
        use clap::CommandFactory;
        let mut cmd = Cli::command();
        let viz = cmd
            .find_subcommand_mut("viz")
            .expect("the viz subcommand exists");
        let help = viz.render_long_help().to_string();
        // Collapse every run of whitespace (including clap's line wraps) into one
        // space so the substring matches survive help reflow.
        let flat: String = help.split_whitespace().collect::<Vec<_>>().join(" ");
        assert!(
            flat.contains("Applies ONLY when the verb spawns a viewer"),
            "serve-web help must state the APPLIES condition:\n{help}"
        );
        // CER-825: under the daemon-client model the ignore conditions are
        // --no-spawn and an absent `rerun` binary (there is no reachable-attach /
        // env-set case anymore — the daemon hosts the proxy, the viewer connects).
        assert!(
            flat.contains("`--no-spawn`"),
            "serve-web help must name --no-spawn among the ignore conditions:\n{help}"
        );
        assert!(
            flat.contains("no `rerun` binary is found"),
            "serve-web help must name the absent-binary ignore condition:\n{help}"
        );
    }

    /// CER-722/CER-760: `--robot-name` parses to `Some(name)`; its ABSENCE
    /// defaults to `None` (the identity is then derived from the topics'
    /// shared namespace, else the hostname). Clap-level unit, no process spawn.
    #[test]
    fn ros_attach_robot_name_flag_parses_and_defaults_none() {
        // Absent: robot_name == None.
        let cli = Cli::try_parse_from(["cerulion", "ros2", "attach", "--iface", "10.0.0.7"])
            .expect("iface-only ros2 attach must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Attach { robot_name, .. },
            } => assert_eq!(robot_name, None, "robot_name defaults to None"),
            _ => panic!("expected Ros2::Attach"),
        }
        // Present: robot_name == Some(value).
        let cli = Cli::try_parse_from([
            "cerulion",
            "ros2",
            "attach",
            "--iface",
            "10.0.0.7",
            "--robot-name",
            "spot",
        ])
        .expect("ros2 attach --robot-name must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Attach { robot_name, .. },
            } => assert_eq!(robot_name.as_deref(), Some("spot")),
            _ => panic!("expected Ros2::Attach"),
        }
    }

    /// The global `--verbose` flag also accepts the `-v` short alias, and the
    /// two are equivalent; its ABSENCE defaults to `false`. Clap-level unit.
    #[test]
    fn verbose_flag_accepts_short_and_long_alias() {
        // Absent → false.
        let cli = Cli::try_parse_from(["cerulion", "ros2", "attach", "--iface", "10.0.0.7"])
            .expect("bare subcommand must parse");
        assert!(!cli.verbose, "verbose defaults to false");
        // `-v` → true.
        let short =
            Cli::try_parse_from(["cerulion", "-v", "ros2", "attach", "--iface", "10.0.0.7"])
                .expect("-v must parse");
        assert!(short.verbose, "-v sets verbose true");
        // `--verbose` → true (equivalent to `-v`).
        let long = Cli::try_parse_from([
            "cerulion",
            "--verbose",
            "ros2",
            "attach",
            "--iface",
            "10.0.0.7",
        ])
        .expect("--verbose must parse");
        assert!(long.verbose, "--verbose sets verbose true");
        assert_eq!(
            short.verbose, long.verbose,
            "-v and --verbose are equivalent"
        );
    }
}

#[cfg(test)]
mod verb_log_class_tests {
    use super::*;

    /// Parse a full argv and return the classified verb log class — the exact
    /// path `main` takes (`Cli::parse().command.log_verb_class()`).
    fn class_of(argv: &[&str]) -> VerbLogClass {
        Cli::try_parse_from(argv)
            .unwrap_or_else(|e| panic!("argv {argv:?} must parse: {e}"))
            .command
            .log_verb_class()
    }

    /// CER-836: one-shot introspection / management verbs default QUIET
    /// (`OneShot` → `cerulion=warn`). `topic echo`/`hz` LOOP but are still
    /// one-shot by the founder's ruling — their OUTPUT is the data, and the
    /// breadcrumbs are noise.
    #[test]
    fn one_shot_verbs_are_quiet() {
        let one_shot: &[&[&str]] = &[
            &["cerulion", "topic", "list"],
            &["cerulion", "topic", "info", "/t"],
            &["cerulion", "topic", "echo", "/t"],
            &["cerulion", "topic", "hz", "/t"],
            &["cerulion", "schema", "list"],
            &["cerulion", "schema", "info", "sensor_msgs/Image"],
            &["cerulion", "schema", "create", "Foo"],
            &["cerulion", "schema", "delete", "Foo"],
            &["cerulion", "node", "list"],
            &["cerulion", "node", "info", "camera"],
            &[
                "cerulion", "node", "create", "camera", "--policy", "external",
            ],
            &["cerulion", "node", "delete", "camera"],
            &["cerulion", "node", "build", "camera"],
            &["cerulion", "node", "stage", "camera"],
            &["cerulion", "graph", "create", "g"],
            &["cerulion", "graph", "validate", "g"],
            &["cerulion", "graph", "list"],
            &["cerulion", "graph", "levels", "g"],
            &["cerulion", "graph", "partition", "g"],
            &["cerulion", "workspace", "create", "ws"],
            &["cerulion", "workspace", "init"],
            &["cerulion", "trace", "inspect", "somedir"],
            &["cerulion", "clean"],
        ];
        for argv in one_shot {
            assert_eq!(
                class_of(argv),
                VerbLogClass::OneShot,
                "{argv:?} must be a QUIET one-shot verb"
            );
            assert!(
                class_of(argv).is_quiet_default(),
                "{argv:?} is_quiet_default() must be true"
            );
        }
    }

    /// CER-982 × CER-1053 chunk 9: `--strict-state` crossed the verb rename and
    /// parses on `bag play`, in both resim modes.
    ///
    /// The RESOLVER half (it reaches the engine in both modes) is pinned by
    /// `resim_cmd::tests::run_shaping_flags_are_legal_in_both_resim_modes`; this
    /// arm pins the half that one structurally cannot see — that clap parses the
    /// flag onto the variant at all, so a `main` destructure which dropped it
    /// would fail here rather than silently ignoring the operator's request.
    #[test]
    fn bag_play_parses_strict_state_in_both_resim_modes() {
        for (argv, want_verify) in [
            (
                vec![
                    "cerulion",
                    "bag",
                    "play",
                    "b.mcap",
                    "--resim",
                    "all",
                    "--strict-state",
                ],
                false,
            ),
            (
                vec![
                    "cerulion",
                    "bag",
                    "play",
                    "b.mcap",
                    "--resim",
                    "all",
                    "--verify",
                    "--strict-state",
                ],
                true,
            ),
        ] {
            let cli = Cli::try_parse_from(&argv).expect("must parse");
            let Commands::Bag {
                action:
                    BagAction::Play {
                        strict_state,
                        verify,
                        resim,
                        ..
                    },
            } = cli.command
            else {
                panic!("expected `bag play`");
            };
            assert!(strict_state, "{argv:?} must carry --strict-state");
            assert_eq!(verify, want_verify, "{argv:?}");
            assert_eq!(resim.as_deref(), Some("all"), "{argv:?}");
        }
    }

    /// CER-836: runtime / daemon verbs keep the `info` default (`LongRunning`).
    #[test]
    fn long_running_verbs_keep_info() {
        let long_running: &[&[&str]] = &[
            &["cerulion", "graph", "run", "g"],
            &["cerulion", "graph", "profile", "g"],
            &["cerulion", "node", "run", "camera"],
            // CER-982: `bag play --resim` is where the removed `cerulion
            // replay`'s re-execution went, so it inherits its log class — as
            // does plain playback, which was already `LongRunning`.
            &["cerulion", "bag", "play", "bag.mcap", "--resim", "all"],
            &["cerulion", "ros2", "attach", "--iface", "10.0.0.7"],
            &["cerulion", "viz"],
            &["cerulion", "connect", "robot1"],
        ];
        for argv in long_running {
            assert_eq!(
                class_of(argv),
                VerbLogClass::LongRunning,
                "{argv:?} must keep the `info` default"
            );
            assert!(
                !class_of(argv).is_quiet_default(),
                "{argv:?} is_quiet_default() must be false"
            );
        }
    }

    /// The hidden multi-process worker / gateway verbs run a runtime loop, so
    /// they keep `info` (they are spawned by `graph run`, never typed).
    #[test]
    fn hidden_worker_verbs_keep_info() {
        assert_eq!(
            class_of(&["cerulion", "graph", "run-worker", "--plan", "/tmp/p.json"]),
            VerbLogClass::LongRunning,
            "run-worker is a runtime process"
        );
        assert_eq!(
            class_of(&[
                "cerulion",
                "graph",
                "run-gateway",
                "--handoff",
                "/tmp/h.json"
            ]),
            VerbLogClass::LongRunning,
            "run-gateway is a runtime process"
        );
    }

    /// `-v/--verbose` is class-independent (it is applied on top of the class
    /// in `init_logging`), so the classification of a verb is the SAME with or
    /// without `-v` — the flag raises the level, it does not reclassify.
    #[test]
    fn verbose_flag_does_not_change_class() {
        assert_eq!(
            class_of(&["cerulion", "-v", "topic", "list"]),
            VerbLogClass::OneShot,
            "-v does not reclassify a one-shot verb"
        );
        assert_eq!(
            class_of(&["cerulion", "-v", "graph", "run", "g"]),
            VerbLogClass::LongRunning,
            "-v does not reclassify a long-running verb"
        );
    }
}

#[cfg(test)]
mod topic_echo_flag_tests {
    use super::*;

    /// Destructure a parsed `topic echo` command into its `truncate_length`.
    fn parse_echo_truncate(args: &[&str]) -> u64 {
        let mut argv = vec!["cerulion", "topic", "echo", "t"];
        argv.extend_from_slice(args);
        let cli = Cli::try_parse_from(argv).expect("`topic echo` must parse");
        match cli.command {
            Commands::Topic {
                action:
                    TopicAction::Echo {
                        truncate_length, ..
                    },
            } => truncate_length,
            _ => panic!("expected Topic::Echo"),
        }
    }

    /// CER-849: bare `topic echo <topic>` defaults `--truncate-length` to the
    /// engine's `DEFAULT_ECHO_TRUNCATE_LENGTH` (128) — one source of truth.
    #[test]
    fn topic_echo_bare_defaults_to_128() {
        assert_eq!(
            parse_echo_truncate(&[]),
            cerulion_cli_engine::topic_cmd::DEFAULT_ECHO_TRUNCATE_LENGTH as u64,
        );
    }

    /// CER-849: an explicit `--truncate-length N` overrides the default.
    #[test]
    fn topic_echo_truncate_length_parses() {
        assert_eq!(parse_echo_truncate(&["--truncate-length", "4"]), 4);
        assert_eq!(parse_echo_truncate(&["--truncate-length", "1"]), 1);
    }

    /// CER-849: `--truncate-length 0` is REJECTED at parse by the `range(1..)`
    /// value_parser. Asserting the error KIND is `ValueValidation` pins that the
    /// range check (not some unrelated parse failure) is what rejected it — the
    /// element render bound cannot be zero.
    #[test]
    fn topic_echo_truncate_length_zero_rejected() {
        // `Cli` does not derive `Debug`, so match (not `expect_err`) to reach the
        // error without requiring `T: Debug` on the Ok arm.
        match Cli::try_parse_from(["cerulion", "topic", "echo", "t", "--truncate-length", "0"]) {
            Ok(_) => panic!("--truncate-length 0 must be rejected at parse"),
            Err(err) => assert_eq!(
                err.kind(),
                clap::error::ErrorKind::ValueValidation,
                "range(1..) must be the rejecter (got {:?})",
                err.kind()
            ),
        }
    }

    /// CER-849: a NEGATIVE `--truncate-length` is REJECTED at parse. clap treats
    /// `-5` as an unexpected hyphen-led argument (it looks like a flag) and
    /// errors BEFORE the u64 value_parser / range check ever runs — a
    /// parse-structure rejection, not the `ValueValidation` the zero case hits.
    /// Either way it must be an error.
    #[test]
    fn topic_echo_truncate_length_negative_rejected() {
        assert!(
            Cli::try_parse_from(["cerulion", "topic", "echo", "t", "--truncate-length", "-5"])
                .is_err(),
            "a negative --truncate-length must be rejected at parse"
        );
    }

    /// The `(run, topics)` a `bag record` argv parses to, or `None` when the
    /// argv is not a `bag record` at all.
    fn parse_bag_record(argv: &[&str]) -> Option<(Option<String>, Vec<String>)> {
        let cli = Cli::try_parse_from(argv).ok()?;
        match cli.command {
            Commands::Bag {
                action: BagAction::Record { run, topics, .. },
            } => Some((run, topics)),
            _ => None,
        }
    }

    /// CER-981 C2: `--run` must not SWALLOW the first positional topic.
    ///
    /// The verb's whole point is that a run and a topic list COMPOSE — attach to
    /// a run for its graph/env/identity, record a hand-named subset of it — and
    /// with a bare `num_args = 0..=1` that composition was unspellable: clap
    /// bound `/topic` as the RUN NAME, leaving zero topics, so the operator got
    /// a "no such run" refusal naming their own topic. `require_equals` is what
    /// separates the two, and all three shapes are pinned together because the
    /// fix is exactly a re-partition of one argv space.
    #[test]
    fn cer981_bare_run_does_not_swallow_the_first_positional_topic() {
        let (run, topics) = parse_bag_record(&["cerulion", "bag", "record", "--run", "/telemetry"])
            .expect("`bag record --run /telemetry` must parse");
        assert_eq!(
            run.as_deref(),
            Some(""),
            "a bare --run must take its default_missing_value (the SOLE-run target), never the \
             next word"
        );
        assert_eq!(
            topics,
            vec!["/telemetry".to_string()],
            "the word after a bare --run is a TOPIC — this composition is the one the verb \
             documents and it must survive parsing"
        );
    }

    /// The other half: `--run=NAME` still NAMES a run, and topics beside it are
    /// still topics. Without this arm the fix above is satisfied by an arg that
    /// takes no value at all.
    #[test]
    fn cer981_run_equals_names_a_run_and_still_composes_with_topics() {
        let (run, topics) =
            parse_bag_record(&["cerulion", "bag", "record", "--run=perception", "/a", "/b"])
                .expect("`bag record --run=perception /a /b` must parse");
        assert_eq!(
            run.as_deref(),
            Some("perception"),
            "--run=<RUN> is how a run is named"
        );
        assert_eq!(topics, vec!["/a".to_string(), "/b".to_string()]);
    }

    /// A bare `--run` with NO topics is still the sole-run attach — the shape
    /// `bag record --run` is documented as, and the one `default_missing_value`
    /// exists for.
    #[test]
    fn cer981_bare_run_alone_is_the_sole_run_target() {
        let (run, topics) =
            parse_bag_record(&["cerulion", "bag", "record", "--run"]).expect("`--run` must parse");
        assert_eq!(run.as_deref(), Some(""));
        assert!(
            topics.is_empty(),
            "nothing follows the flag, so nothing is a topic"
        );
    }
}

#[cfg(test)]
mod ros2_run_dispatch_tests {
    use super::*;

    /// Parse argv and destructure the `Ros2` action into (is_launch, args).
    /// This pins the clap FALLBACK path only — the primary dispatch is
    /// `main`'s raw-argv intercept, which never parses at all (that is what
    /// makes a LEADING hyphenated token forwardable; the e2e pins it).
    fn parse_ros2(argv: &[&str]) -> (bool, Vec<String>) {
        let cli = Cli::try_parse_from(argv).expect("argv must parse");
        match cli.command {
            Commands::Ros2 {
                action: Ros2Action::Run { args },
            } => (false, args),
            Commands::Ros2 {
                action: Ros2Action::Launch { args },
            } => (true, args),
            _ => panic!("expected Commands::Ros2"),
        }
    }

    /// `ros2 run`: everything after the verb token is collected verbatim.
    #[test]
    fn ros2_run_collects_args_verbatim() {
        let (is_launch, args) = parse_ros2(&[
            "cerulion",
            "ros2",
            "run",
            "demo_nodes_cpp",
            "talker",
            "--ros-args",
            "-r",
            "chatter:=c2",
        ]);
        assert!(!is_launch);
        assert_eq!(
            args,
            [
                "demo_nodes_cpp",
                "talker",
                "--ros-args",
                "-r",
                "chatter:=c2"
            ]
        );
    }

    /// `ros2 launch`: the package form and launch arguments pass through
    /// untouched — nothing here knows what a launch file is.
    #[test]
    fn ros2_launch_collects_package_form_and_launch_args() {
        let (is_launch, args) = parse_ros2(&[
            "cerulion",
            "ros2",
            "launch",
            "moveit2_tutorials",
            "demo.launch.py",
            "use_rviz:=false",
        ]);
        assert!(is_launch);
        assert_eq!(
            args,
            ["moveit2_tutorials", "demo.launch.py", "use_rviz:=false"]
        );
    }

    /// Hyphenated tokens AFTER a positional are collected raw
    /// (trailing_var_arg + allow_hyphen_values) — clap never interprets them.
    #[test]
    fn ros2_post_token_hyphenated_args_are_collected_raw() {
        let (_, args) = parse_ros2(&[
            "cerulion",
            "ros2",
            "launch",
            "demo.launch.py",
            "--show-args",
        ]);
        assert_eq!(args, ["demo.launch.py", "--show-args"]);
    }

    /// Bare `cerulion ros2` requires a subcommand (clap's exit-2 usage error
    /// — the one usage class the wrappers own).
    #[test]
    fn ros2_without_action_is_a_usage_error() {
        assert!(
            Cli::try_parse_from(["cerulion", "ros2"]).is_err(),
            "ros2 requires an action"
        );
    }
}