oxo-flow-cli 0.27.0

#![forbid(unsafe_code)]
//! oxo-flow CLI — Bioinformatics pipeline engine.
//!
//! Provides subcommands for running, validating, and managing workflows.

mod background;
pub mod banner;
pub mod commands;
mod logging;

use crate::commands::ai_status::{ai_setup_command, ai_status_command, ai_test_command};
use crate::commands::batch::batch_command;
use crate::commands::clean::clean_command;
use crate::commands::cluster::cluster_command;
use crate::commands::completions::handle_completions;
use crate::commands::info::info_command;
use crate::commands::infra::{env_command, handle_license};
use crate::commands::output::{
    GraphFormat, MetroArg, handle_diff, handle_export, handle_graph, handle_report,
};
use crate::commands::project::{init_command, template_command};
use crate::commands::provenance::provenance_verify_command;
use crate::commands::publish::publish_command;
use crate::commands::quality::{format_command, lint_command, touch_command, validate_command};
use crate::commands::run::{
    debug_command, dry_run_command, handle_status, resume_command, run_command,
};
use crate::commands::verify_data::verify_data_command;
use crate::commands::web::ServeMode;
use anyhow::Result;
use clap::{CommandFactory, FromArgMatches, Parser, Subcommand};
use std::path::{Path, PathBuf};

/// oxo-flow — A Rust-native bioinformatics pipeline engine.
///
/// Build, validate, and execute reproducible bioinformatics workflows
/// with first-class environment management and clinical-grade reporting.
#[derive(Parser, Debug)]
#[command(
    name = "oxo-flow",
    version,
    about = "A Rust-native bioinformatics pipeline engine",
    long_about = "oxo-flow is a high-performance, modular bioinformatics pipeline engine\n\
                   built from first principles in Rust. It supports conda, mamba, pixi,\n\
                   docker, singularity, venv, and environment-modules backends (plus the\n\
                   system environment) with DAG-based execution.",
    after_help = "Get started:\n  \
        oxo-flow init my-pipeline && cd my-pipeline   # scaffold an example project\n  \
        oxo-flow run my-pipeline.oxoflow               # execute it\n  \
        oxo-flow template \"RNA-seq with STAR\" -o pipeline.oxoflow   # generate a workflow (add --ai for LLM authoring)\n\n\
        Docs: https://traitome.github.io/oxo-flow/"
)]
pub struct Cli {
    #[command(subcommand)]
    command: Commands,

    /// Enable verbose (debug-level) logging.
    #[arg(global = true, short = 'v', long)]
    verbose: bool,

    /// Suppress non-essential output (errors only).
    #[arg(global = true, long)]
    quiet: bool,

    /// Disable colored output. Also respects the NO_COLOR environment variable.
    #[arg(global = true, long)]
    no_color: bool,

    /// Output machine-readable JSON to stdout (human-readable logs stay on stderr).
    /// Implemented by run, dry-run, validate, lint, test, status, batch, info,
    /// schema, verify-data, license, and ai; every other command rejects the
    /// flag instead of silently ignoring it.
    #[arg(global = true, long)]
    json: bool,
}

#[derive(Subcommand, Debug)]
pub enum Commands {
    /// Execute a workflow.
    Run {
        #[arg(
            value_name = "WORKFLOW",
            help = "Path to the .oxoflow workflow file, or a repository to run \
                    (gh:owner/repo[@ref], owner/repo[@ref], *.git URL, or file://dir)"
        )]
        workflow: Option<PathBuf>,
        #[arg(
            short = 'j',
            long,
            default_value = "1",
            help = "Maximum number of concurrent jobs"
        )]
        jobs: usize,
        #[arg(short = 'k', long, help = "Continue execution when a job fails")]
        keep_going: bool,
        #[arg(short = 'd', long, help = "Working directory for execution")]
        workdir: Option<PathBuf>,
        #[arg(
            long,
            value_name = "PATH",
            help = "Write the run log to PATH instead of the default .oxo-flow/logs/oxo-flow.log (previous logs rotate to PATH.1 .. PATH.9)"
        )]
        log_file: Option<PathBuf>,
        #[arg(
            short = 't',
            long,
            help = "Run only specific target rules (repeatable, prefix matching)"
        )]
        target: Vec<String>,
        #[arg(
            long,
            help = "Run one include module and the producers of its declared inputs (repeatable; unions with --target). Module names are the include's `name` field or its file stem"
        )]
        module: Vec<String>,
        #[arg(
            short = 'r',
            long,
            default_value = "0",
            help = "Number of times to retry failed jobs"
        )]
        retry: u32,
        #[arg(
            long,
            default_value = "0",
            help = "Timeout per job in seconds (0 = disabled), or a duration like 1h/30m"
        )]
        timeout: String,
        #[arg(long, help = "Resume only failed rules from a previous run")]
        resume_failed: bool,
        #[arg(
            long,
            help = "Execution profile name (loaded from profiles/<NAME>.toml or profiles/<NAME>.oxoflow next to the workflow; fills in config keys the workflow does not set)"
        )]
        profile: Option<String>,
        #[arg(
            long,
            default_value = "0",
            help = "Maximum CPU threads available for execution (0 = auto-detect)"
        )]
        max_threads: u32,
        #[arg(
            long,
            default_value = "0",
            help = "Maximum memory in MB available for execution (0 = auto-detect)"
        )]
        max_memory: u64,
        #[arg(long, help = "Skip environment setup (assume environments are ready)")]
        skip_env_setup: bool,
        #[arg(long, help = "Skip automatic reference/index building")]
        skip_ref_build: bool,
        #[arg(
            long,
            help = "Directory for caching environment setup state (reclaim stale entries with `oxo-flow env gc`)"
        )]
        cache_dir: Option<PathBuf>,
        #[arg(long, help = "Track output file checksums for later verification")]
        provenance: bool,
        #[arg(long, help = "Execute from a published .tar.zst bundle")]
        bundle: Option<PathBuf>,
        #[arg(
            long = "yes",
            help = "Skip the confirmation prompt when running from a bundle (required in non-interactive sessions: CI, scripts, redirected input, or --json)"
        )]
        yes: bool,
        #[arg(
            long = "arg",
            value_name = "KEY=VALUE",
            help = "Set a workflow config value (overrides `[config]` defaults). Repeatable."
        )]
        args: Vec<String>,
        #[arg(
            value_name = "KEY=VALUE",
            trailing_var_arg = true,
            allow_hyphen_values = true,
            help = "Direct config overrides: KEY=VALUE or --KEY=VALUE; --KEY VALUE (space form) is accepted for declared `[config]` keys — unknown --flags are rejected"
        )]
        config_overrides: Vec<String>,
        /// Enable AI error recovery on rule failure.
        #[arg(long)]
        ai_recover: bool,
        /// Maximum AI attempts for recovery when a call fails (default: 1;
        /// only a FAILED call is retried).
        #[arg(long = "ai-max-retries", value_name = "N")]
        ai_max_retries: Option<u32>,
        /// Filter to a subset of samples: `first:N` or explicit names
        /// (repeatable, comma-separated)
        #[arg(
            long = "samples",
            value_name = "LIST",
            help = "Sample selection: '@path' replaces the workflow's samples from a samplesheet, '+@path' appends (same-name groups merge); names, first:N (pilot), or ready (complete inputs) filter to a subset. Repeatable, comma-separated."
        )]
        samples_filter: Vec<String>,
        /// Re-execute every rule selected for this run even if outputs are
        /// up to date. Checkpoint records for rules outside this run are kept.
        #[arg(
            long,
            help = "Force re-execution of this run's rules (ignore up-to-date checks)"
        )]
        rerun: bool,
        /// Skip the automatic report snapshot after the run.
        #[arg(long, help = "Skip the automatic report snapshot after the run")]
        no_report_snapshot: bool,
        /// Queue cap for cluster runs; overrides the profile's
        /// `[cluster] max_submitted` for this invocation.
        #[arg(
            long,
            value_name = "N",
            help = "Cluster jobs in flight at once (overrides the profile's max_submitted)"
        )]
        max_submitted: Option<usize>,
        /// Detach execution into a background process: the foreground
        /// invocation spawns a detached child that re-runs this command
        /// without the flag, writes its pid to .oxo-flow/background.pid,
        /// prints a summary on stderr, and exits 0 immediately. Monitor
        /// with `oxo-flow status`, the run log, or report snapshots; stop
        /// with `kill <pid>` (the pid is in the summary and the pid file).
        #[arg(
            long,
            help = "Detach execution into a background process (see long help)"
        )]
        background: bool,
        /// Proceed despite protected-tree drift (issue #935): `\[engine]`
        /// protected_data_trees baselines differ from the current tree.
        /// Without this flag `run` refuses to execute.
        #[arg(long)]
        allow_protected_drift: bool,
    },
    /// Resume an interrupted workflow from a checkpoint.
    Resume {
        #[arg(
            value_name = "CHECKPOINT",
            help = "Path to the checkpoint file (.oxo-flow/checkpoint.json)"
        )]
        checkpoint: PathBuf,
        #[arg(
            short = 'j',
            long,
            default_value = "1",
            help = "Maximum number of concurrent jobs"
        )]
        jobs: usize,
        /// Enable AI error recovery on rule failure.
        #[arg(long)]
        ai_recover: bool,
        /// Maximum AI attempts for recovery when a call fails (default: 1;
        /// only a FAILED call is retried).
        #[arg(long = "ai-max-retries", value_name = "N")]
        ai_max_retries: Option<u32>,
        #[arg(
            short = 'k',
            long,
            help = "Continue execution when a job fails (same semantics as `run`)"
        )]
        keep_going: bool,
        #[arg(
            long,
            default_value = "0",
            help = "Timeout per job in seconds (0 = disabled), or a duration like 1h/30m"
        )]
        timeout: String,
        #[arg(
            short = 'd',
            long,
            help = "Working directory to resume in (default: the one recorded in the checkpoint)"
        )]
        workdir: Option<PathBuf>,
        /// Skip the automatic report snapshot after the resumed run.
        #[arg(
            long,
            help = "Skip the automatic report snapshot after the resumed run"
        )]
        no_report_snapshot: bool,
        /// Detach the resumed run into a background process: the foreground
        /// invocation spawns a detached child that re-runs `resume` without
        /// the flag, writes its pid to .oxo-flow/background.pid, prints a
        /// summary on stderr, and exits 0 immediately (see `run
        /// --background`).
        #[arg(long)]
        background: bool,
    },
    /// Preview execution without running any commands (alias: plan).
    #[command(alias = "plan")]
    DryRun {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: Option<PathBuf>,
        #[arg(
            short = 't',
            long,
            help = "Run only specific target rules (repeatable, prefix matching)"
        )]
        target: Vec<String>,
        #[arg(
            long,
            help = "Run one include module and the producers of its declared inputs (repeatable; unions with --target). Module names are the include's `name` field or its file stem"
        )]
        module: Vec<String>,
        /// Enable AI-powered analysis of the workflow.
        #[arg(long)]
        ai: bool,
        /// Maximum AI attempts for the analysis when a call fails (default:
        /// 1; only a FAILED call is retried).
        #[arg(long = "ai-max-retries", value_name = "N")]
        ai_max_retries: Option<u32>,
        /// Filter to a subset of samples: `first:N` or explicit names
        /// (repeatable, comma-separated).
        #[arg(
            long = "samples",
            value_name = "LIST",
            help = "Sample selection: '@path' replaces the workflow's samples from a samplesheet, '+@path' appends (same-name groups merge); names, first:N (pilot), or ready (complete inputs) filter to a subset. Repeatable, comma-separated."
        )]
        samples_filter: Vec<String>,
        #[arg(
            short = 'd',
            long,
            help = "Working directory to resolve paths against (default: the workflow file's directory)"
        )]
        workdir: Option<PathBuf>,
        /// Execution profile name, loaded from `profiles/<NAME>.toml` — the
        /// SAME merge semantics as `run` (profile values fill in missing
        /// config keys), so the preview matches what run would execute.
        #[arg(long)]
        profile: Option<String>,
        /// Skip automatic reference/index building (assume pre-built) —
        /// mirrors the run flag; the preview lists required builds otherwise.
        #[arg(long)]
        skip_ref_build: bool,
        /// Directory for caching environment setup state — mirrors the run
        /// flag so run→dry-run transcription is 1:1 (issue #441); the
        /// preview reads no env cache, so it is accepted and ignored.
        #[arg(long, value_name = "PATH")]
        cache_dir: Option<PathBuf>,
        /// Set a workflow config value (overrides `[config]` defaults).
        /// Repeatable — same forms as `run` (issue #77 parity).
        #[arg(
            long = "arg",
            value_name = "KEY=VALUE",
            help = "Set a workflow config value (overrides `[config]` defaults). Repeatable."
        )]
        args: Vec<String>,
        /// Direct config overrides in the same trailing positional form
        /// `run` accepts: KEY=VALUE or --KEY=VALUE; --KEY VALUE (space
        /// form) for declared `[config]` keys — unknown --flags are rejected.
        #[arg(
            value_name = "KEY=VALUE",
            trailing_var_arg = true,
            allow_hyphen_values = true,
            help = "Direct config overrides: KEY=VALUE or --KEY=VALUE; --KEY VALUE (space form) is accepted for declared `[config]` keys — unknown --flags are rejected"
        )]
        config_overrides: Vec<String>,
        /// Force re-execution of this run's rules (ignore up-to-date
        /// checks) — previews exactly what `run --rerun` would execute.
        #[arg(long)]
        rerun: bool,
        /// Resume only failed rules from a previous run — mirrors the run
        /// flag's execution set.
        #[arg(long)]
        resume_failed: bool,
    },
    /// Validate a .oxoflow workflow file.
    Validate {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(
            long,
            help = "Validate as a sub-workflow fragment (skips DAG and input-existence checks)"
        )]
        as_include: bool,
        /// Enable AI-powered semantic validation.
        #[arg(long)]
        ai: bool,
    },
    /// Derive catalog metadata from a workflow file.
    Info {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(
            long,
            value_name = "FORMAT",
            help = "Output format: json (default, machine-readable on stdout) or text (human-readable summary)"
        )]
        format: Option<String>,
    },
    /// Initialize a new workflow project.
    Init {
        #[arg(value_name = "NAME", help = "Project name (no path separators)")]
        name: String,
        #[arg(short = 'd', long, help = "Target directory")]
        dir: Option<PathBuf>,
        #[arg(
            long,
            help = "Overwrite an existing workflow file (its contents are lost)"
        )]
        force: bool,
    },
    /// Generate a workflow from a predefined template or via AI.
    Template {
        #[arg(
            value_name = "TEMPLATE",
            help = "Template name or natural-language description (with --ai)"
        )]
        template: Option<String>,
        #[arg(
            short = 'o',
            long,
            help = "Output file or directory (a trailing slash forces a directory)"
        )]
        output: Option<PathBuf>,
        /// Enable AI-powered workflow generation from natural language.
        #[arg(long)]
        ai: bool,
        /// URL(s) to use as reference material for AI generation.
        #[arg(long = "from-url", value_name = "URL")]
        from_url: Vec<String>,
        /// File(s) to use as reference material for AI generation.
        #[arg(long = "from-file", value_name = "PATH")]
        from_file: Vec<PathBuf>,
        /// Maximum AI correction rounds (overrides config).
        #[arg(long = "ai-max-retries", value_name = "N")]
        ai_max_retries: Option<u32>,
        /// Generation team profile: `compact` (one agent + engine gates,
        /// default) or `full` (adds the task contract, the deterministic
        /// Curator brief, and an independent review pass with a bounded
        /// fix). Overrides `[ai] team_profile` in the config.
        #[arg(long = "ai-team-profile", value_name = "PROFILE")]
        ai_team_profile: Option<String>,
        /// Fresh-draw attempts for the whole generation (best-of-N: up to N
        /// independent tries; the first one passing the deterministic gates
        /// wins, so later attempts are only paid when earlier ones fail).
        #[arg(long = "ai-attempts", value_name = "N", default_value_t = 1)]
        ai_attempts: u32,
    },
    /// AI status, test, setup, and workflow explanation.
    ///
    /// Run without args for quick status.
    /// Use 'ai test' for comprehensive self-test.
    /// Use 'ai setup' for interactive wizard.
    /// Use `ai explain <workflow>` for a three-layer explanation
    /// (overview → per-step detail → scientific review).
    #[command(name = "ai")]
    Ai {
        #[arg(
            value_name = "ACTION",
            help = "Action to run: 'test' (comprehensive self-test), 'setup' (interactive wizard), or 'explain' (workflow explanation); omit for a quick status"
        )]
        action: Option<String>,

        /// Workflow file to explain (with action 'explain').
        #[arg(value_name = "WORKFLOW")]
        workflow: Option<PathBuf>,

        /// Explain a single rule by name (with action 'explain').
        #[arg(long, value_name = "RULE")]
        step: Option<String>,

        /// Explanation depth (with action 'explain'). Default: beginner
        /// prose that defines jargon; 'expert' is parameter-level and
        /// efficiency-focused.
        #[arg(long, value_enum, default_value_t = crate::commands::ai_explain::ExplainLevel::Beginner)]
        level: crate::commands::ai_explain::ExplainLevel,

        /// Machine-readable JSON output (with action 'explain'): the
        /// deterministic skeleton plus model-written prose fields.
        #[arg(long)]
        json: bool,
    },

    /// Output the workflow DAG for visualization.
    Graph {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(
            short = 'f',
            long,
            value_enum,
            default_value_t = GraphFormat::Ascii,
            help = "Output format: ascii, dot, dot-clustered, tree, mermaid, metro"
        )]
        format: GraphFormat,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
        #[arg(
            long = "expanded",
            help = "Show the DAG after wildcard/sample/scatter expansion (the actual runtime DAG)"
        )]
        expanded: bool,
        /// Station zoom for the metro format (default: rule): 'rule' keeps
        /// one station per rule; 'process' collapses chain-connected
        /// same-tool rules into tool-named stations (one stop per analysis
        /// step, like a metro map); 'module' emits one station per module
        /// section (publication/overview tier).
        #[arg(long, value_enum)]
        granularity: Option<MetroArg>,
    },
    /// Show execution status and per-rule timings from a checkpoint file.
    Status {
        #[arg(
            value_name = "CHECKPOINT",
            help = "Path to the checkpoint file (default: <workdir>/.oxo-flow/checkpoint.json)"
        )]
        checkpoint: Option<PathBuf>,
        /// Resolve the default checkpoint against this run directory
        /// (same semantics as `run --workdir`); ignored with an explicit
        /// CHECKPOINT path.
        #[arg(long, value_name = "DIR")]
        workdir: Option<PathBuf>,
        /// Show per-rule wall-clock times and total runtime (slowest first).
        #[arg(long)]
        timing: bool,
        #[arg(
            short = 'n',
            long,
            default_value = "10",
            requires = "timing",
            help = "Maximum number of rules to show in the --timing view"
        )]
        limit: usize,
    },
    /// Pull a published bundle from a remote source.
    Pull {
        #[arg(
            value_name = "URL",
            help = "Bundle URL (gh:owner/repo@tag, https://, or file://)"
        )]
        url: String,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
    },
    /// Inspect and manage workflow configuration.
    Config {
        #[command(subcommand)]
        action: ConfigAction,
    },
    /// Compare two .oxoflow workflow files and show differences.
    Diff {
        #[arg(value_name = "WORKFLOW_A", help = "First workflow file to compare")]
        workflow_a: PathBuf,
        #[arg(value_name = "WORKFLOW_B", help = "Second workflow file to compare")]
        workflow_b: PathBuf,
    },
    /// Debug a workflow: show expanded commands after variable substitution.
    Debug {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(
            short = 'r',
            long = "rule",
            help = "Show the expanded command for this rule only"
        )]
        rule_name: Option<String>,
        /// Enable AI-powered command explanation.
        #[arg(long)]
        ai: bool,
    },
    /// Clean workflow outputs and temporary files.
    Clean {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(short = 'n', long, help = "Preview which outputs would be deleted")]
        dry_run: bool,
        #[arg(
            long,
            help = "Actually delete outputs (without this flag, clean only previews)"
        )]
        force: bool,
        #[arg(
            long,
            help = "Remove orphaned transform chunk directories (.oxo-flow/chunks)"
        )]
        orphans: bool,
        #[arg(
            short = 'd',
            long,
            help = "Working directory for .oxo-flow artifacts (default: the workflow file's directory)"
        )]
        workdir: Option<PathBuf>,
    },
    /// Manage software environments.
    Env {
        #[command(subcommand)]
        action: EnvAction,
    },
    /// Reformat a .oxoflow file into canonical TOML form.
    Format {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
        #[arg(long, help = "Only check formatting, don't write")]
        check: bool,
    },
    /// Run best-practice linting checks on a .oxoflow file.
    Lint {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(long, help = "Treat warnings as errors (non-zero exit on any warning)")]
        strict: bool,
        /// Enable AI-powered semantic linting.
        #[arg(long)]
        ai: bool,
    },
    /// Mark workflow outputs as up-to-date without re-executing rules.
    Touch {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(short = 'r', long = "rule", help = "Rule names to touch")]
        rules: Vec<String>,
        #[arg(
            short = 'd',
            long,
            help = "Working directory the outputs live in (default: the workflow file's directory)"
        )]
        workdir: Option<PathBuf>,
        #[arg(
            long = "arg",
            value_name = "KEY=VALUE",
            help = "Config override (KEY=VALUE or --KEY VALUE); touch validates output paths against overridden values"
        )]
        args: Vec<String>,
        #[arg(
            value_name = "OVERRIDES",
            trailing_var_arg = true,
            allow_hyphen_values = true,
            help = "Config overrides as trailing KEY=VALUE pairs"
        )]
        config_overrides: Vec<String>,
    },
    /// Generate reports from workflow execution.
    Report {
        #[arg(
            value_name = "WORKFLOW",
            help = "Path to the .oxoflow workflow file (auto-discovered when omitted)"
        )]
        workflow: Option<PathBuf>,
        #[arg(
            short = 'f',
            long,
            help = "Output format: html, json, md, pdf, pdf-command (default: html, or inferred from the -o extension)"
        )]
        format: Option<String>,
        #[arg(short = 'o', long, help = "Output file path ('-' for stdout)")]
        output: Option<PathBuf>,
        #[arg(
            long = "checkpoint",
            value_name = "PATH",
            help = "Path to checkpoint file (default: .oxo-flow/checkpoint.json)"
        )]
        checkpoint_path: Option<PathBuf>,
        #[arg(
            long = "ai",
            help = "AI result interpretation — plain-language summary of execution outcomes, caveats, and next steps (stderr + report section)"
        )]
        ai: bool,
        #[arg(
            short = 'd',
            long,
            help = "Working directory to look for .oxo-flow in (default: the workflow file's directory)"
        )]
        workdir: Option<PathBuf>,
        #[arg(
            long,
            help = "Reproducible output: pin the generation timestamp (SOURCE_DATE_EPOCH or the Unix epoch) so identical state yields byte-identical reports"
        )]
        ci: bool,
        #[arg(long, help = "Omit the generation timestamp from the report")]
        no_timestamps: bool,
        #[arg(
            long,
            help = "Fail (exit 2) when the checkpoint is missing or the report template fails to render"
        )]
        strict: bool,
        #[arg(long, help = "List available report sections and exit")]
        list_sections: bool,
        #[arg(
            long = "run",
            value_name = "DIR",
            conflicts_with = "workflow",
            help = "Workdir of a previous run: the workflow and checkpoint are auto-discovered there"
        )]
        run_dir: Option<PathBuf>,
        #[arg(long, help = "Failure-focused report: diagnosis first")]
        failed: bool,
        #[arg(
            long,
            help = "Template-only report — ignore execution data (no checkpoint required)"
        )]
        plan: bool,
        #[arg(
            long = "init-template",
            help = "Write the built-in report template to ./report-template.tera and exit"
        )]
        init_template: bool,
        #[arg(
            long = "list-templates",
            help = "List available report templates and exit"
        )]
        list_templates: bool,
        #[arg(
            long = "r-data",
            value_name = "DIR",
            help = "Write R-friendly TSV files (sample_table.tsv, metrics.tsv) to DIR"
        )]
        r_data: Option<PathBuf>,
        #[arg(
            long = "diff",
            value_name = "CHECKPOINT",
            help = "Model-level diff of this report's checkpoint against another checkpoint (stderr, terminal-highlighted)"
        )]
        diff: Option<PathBuf>,
        #[arg(
            long = "acct",
            value_name = "PATH",
            help = "Import SLURM sacct-style CSV accounting (JobID,JobName,State,Elapsed,CPUTime,MaxRSS) into a Resource Accounting section"
        )]
        acct: Option<PathBuf>,
        #[arg(
            long = "versions-yml",
            value_name = "PATH",
            help = "Write an nf-core-style versions.yml of declared software (docker image:tag, module tool/version, env-file hashes) to PATH ('-' for stdout) and exit; diff it in CI to catch undeclared dependency drift"
        )]
        versions_yml: Option<PathBuf>,
    },
    /// Start the web interface server.
    Serve {
        /// Server operation mode: personal, team, or hpc.
        #[arg(long, value_enum, default_value_t = ServeMode::Personal, env = "OXO_FLOW_MODE")]
        mode: ServeMode,
        #[arg(
            long,
            default_value = "127.0.0.1",
            env = "OXO_FLOW_HOST",
            help = "Address to bind"
        )]
        host: String,
        #[arg(
            short = 'p',
            long,
            default_value = "8080",
            env = "OXO_FLOW_PORT",
            help = "Port to listen on"
        )]
        port: u16,
        #[arg(
            long,
            default_value = "/",
            env = "OXO_FLOW_BASE_PATH",
            help = "Base URL path for the web interface"
        )]
        base_path: String,
        /// Open the web interface in the default browser on startup
        /// (desktop-app experience).
        #[arg(long = "open", env = "OXO_FLOW_OPEN_BROWSER")]
        open_browser: bool,
    },
    /// Generate shell completions for oxo-flow.
    Completions {
        #[arg(value_enum, help = "Shell to generate completions for")]
        shell: clap_complete::Shell,
    },
    /// Export a workflow to a container definition or standalone TOML.
    Export {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(short = 'f', long, default_value = "docker", help = "Output format")]
        format: String,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
    },
    /// Manage cluster job submission and monitoring.
    Cluster {
        #[command(subcommand)]
        action: ClusterAction,
    },
    /// Execute a command template in parallel across multiple items.
    Batch {
        #[arg(
            value_name = "TEMPLATE",
            help = "Shell command template with {item} placeholder"
        )]
        template: String,
        #[arg(
            value_name = "ITEMS",
            help = "Files or items to process (glob patterns supported)"
        )]
        items: Vec<String>,
        #[arg(
            short = 'j',
            long,
            default_value = "1",
            help = "Maximum number of concurrent jobs"
        )]
        jobs: usize,
        #[arg(short = 'x', long, help = "Stop on the first failed item")]
        stop_on_error: bool,
        #[arg(short = 'f', long, help = "Read items from a file (one per line)")]
        file: Option<PathBuf>,
        #[arg(long = "json-output", help = "Output results as formatted JSON")]
        json_output: bool,
        #[arg(
            short = 'n',
            long,
            help = "Preview the commands without executing them"
        )]
        dry_run: bool,
        #[arg(short = 'd', long, help = "Working directory for execution")]
        workdir: Option<PathBuf>,
        #[arg(short = 'e', long, help = "Environment to run each item in")]
        environment: Option<String>,
        #[arg(long, help = "Generate a .oxoflow workflow file from the template")]
        generate_workflow: bool,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
    },
    /// Verify output file integrity using stored checksums.
    Provenance {
        #[command(subcommand)]
        action: ProvenanceAction,
    },
    /// Verify a data tree against a vendor md5 sidecar (md5sum format).
    VerifyData {
        #[arg(value_name = "TREE", help = "Directory to verify")]
        tree: PathBuf,
        #[arg(
            long,
            value_name = "MD5_FILE",
            help = "md5 sidecar to verify against (md5sum text format)"
        )]
        from: PathBuf,
    },
    /// Output the JSON Schema for the .oxoflow format.
    Schema,
    /// Run a workflow in test mode, validating and verifying outputs.
    Test {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(long, help = "File whose existence is verified after the test run")]
        output: Option<PathBuf>,
        #[arg(
            long,
            help = "Execute the workflow (default only validates and verifies outputs)"
        )]
        run: bool,
        #[arg(
            short = 'j',
            long,
            default_value = "1",
            help = "Maximum number of concurrent jobs"
        )]
        jobs: usize,
        /// Filter to a subset of samples: `@path`/`+@path` sheet
        /// operations, `first:N`, explicit names, or `ready` — the SAME
        /// unified selection `run`/`dry-run` use (the handler forwards the
        /// specs to both, so the help text mirrors their wording).
        #[arg(
            long = "samples",
            value_name = "LIST",
            help = "Sample selection: '@path' replaces the workflow's samples from a samplesheet, '+@path' appends (same-name groups merge); names, first:N (pilot), or ready (complete inputs) filter to a subset. Repeatable, comma-separated."
        )]
        samples_filter: Vec<String>,
        /// Run deep health checks (script files, environment definition files,
        /// system-backend binaries, reference data) after the basic steps.
        #[arg(
            long,
            help = "Run deep checks: script files, env YAML files, backend binaries, reference data"
        )]
        deep: bool,
        #[arg(
            short = 'd',
            long,
            help = "Working directory for the test run (default: the workflow file's directory)"
        )]
        workdir: Option<PathBuf>,
        /// Execution profile for the --run step (same merge semantics as
        /// `run`); the validation/dry-run steps use it for their preview.
        #[arg(long)]
        profile: Option<String>,
        #[arg(
            short = 't',
            long,
            help = "Run only specific target rules (repeatable, prefix matching) — applies to the --run step"
        )]
        target: Vec<String>,
        #[arg(
            long,
            default_value = "0",
            help = "Timeout per job in seconds (0 = disabled) — applies to the --run step"
        )]
        timeout: String,
        #[arg(
            long,
            default_value = "0",
            help = "Number of times to retry failed jobs — applies to the --run step"
        )]
        retry: u32,
        #[arg(
            short = 'k',
            long,
            help = "Continue execution when a job fails — applies to the --run step"
        )]
        keep_going: bool,
    },
    /// Publish a workflow with its environment files into a bundle.
    Publish {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(short = 'o', long, help = "Output file path")]
        output: Option<PathBuf>,
        #[arg(long, help = "Generate conda lockfiles for reproducible environments")]
        with_lockfiles: bool,
        #[arg(
            long = "format",
            help = "Bundle archive format: tar.zst (default) or tar.gz"
        )]
        format: Option<String>,
    },
    /// Verify or display license status.
    License {
        /// Path to license file to verify (optional; checks current status if omitted)
        #[arg(value_name = "LICENSE_PATH", help = "License file path")]
        path: Option<PathBuf>,
    },
}

#[derive(Subcommand, Debug)]
pub enum EnvAction {
    /// List environments declared by a workflow (or available backends without one).
    List {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: Option<PathBuf>,
    },
    /// Check whether a workflow's environments are ready to use.
    Check {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: Option<PathBuf>,
    },
    /// Create a new environment from a spec file.
    Create {
        #[arg(
            value_name = "SPEC",
            help = "Environment spec file (.yaml/.yml/.toml/.lock), or a description with --ai"
        )]
        spec: PathBuf,
        #[arg(short = 'n', long, help = "Environment or profile name")]
        name: Option<String>,
        #[arg(
            long = "ai",
            help = "Generate the environment spec from a natural-language description (SPEC is the description)"
        )]
        ai: bool,
        #[arg(
            long = "backend",
            value_name = "BACKEND",
            help = "Environment backend: conda (YAML) or pixi (TOML). Default: inferred from the spec's extension, or conda with --ai"
        )]
        backend: Option<String>,
    },
    /// Inspect and clean up the shared environment store (issue #904).
    ///
    /// OPT-IN: set OXO_FLOW_ENV_STORE to a non-empty path to activate the
    /// store. When enabled, pixi/venv environments materialize once per
    /// content key under the store root and workflow trees hold symlinks
    /// into it. GC never deletes a store dir that a checkpoint in the
    /// CURRENT workflow directory still references (the #893 lesson) —
    /// runs in other worktrees are protected only by the age gate, so
    /// check --dry-run output before a real run when several worktrees
    /// share one store.
    Gc {
        /// Only delete unreferenced store dirs untouched for at least this
        /// many days (default: 90). Store dirs referenced by checkpoints in
        /// other workdirs are NOT seen by GC — they are protected only by
        /// this age gate, so keep it well above the interval between runs
        /// in any workdir sharing the store.
        #[arg(long, value_name = "DAYS", default_value_t = 90)]
        min_age_days: u64,
        /// Show what would be deleted without touching anything.
        #[arg(long)]
        dry_run: bool,
    },
}

#[derive(Subcommand, Debug)]
pub enum ConfigAction {
    /// Show a workflow's declared configuration values.
    Show {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
    },
    /// Summarize a workflow's configuration (alias: check).
    #[command(alias = "check")]
    Stats {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
    },
    /// Get a single configuration key's value.
    Get {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(value_name = "KEY", help = "Config key")]
        key: String,
    },
}

#[derive(Subcommand, Debug)]
pub enum ClusterAction {
    /// Generate and submit job scripts for a workflow to the cluster.
    Submit {
        #[arg(value_name = "WORKFLOW", help = "Path to the .oxoflow workflow file")]
        workflow: PathBuf,
        #[arg(
            short = 'b',
            long,
            help = "Cluster backend: slurm, pbs, sge, or lsf (defaults to $OXO_FLOW_CLUSTER_BACKEND, then slurm)"
        )]
        backend: Option<String>,
        #[arg(short = 'q', long, help = "Cluster queue or partition name")]
        queue: Option<String>,
        #[arg(short = 'a', long, help = "Cluster billing account")]
        account: Option<String>,
        #[arg(
            long,
            value_name = "SPEC",
            help = "Wall-time limit for every job (24h, 2d, or 24:00:00); a rule's time_limit wins"
        )]
        walltime: Option<String>,
        #[arg(
            long = "extra-arg",
            value_name = "ARG",
            allow_hyphen_values = true,
            help = "Extra scheduler argument, passed through verbatim (repeatable)"
        )]
        extra_args: Vec<String>,
        #[arg(
            short = 'o',
            long,
            default_value = "cluster_scripts",
            help = "Directory for generated job scripts"
        )]
        output: PathBuf,
        #[arg(
            short = 't',
            long,
            help = "Run only specific target rules (repeatable, prefix matching)"
        )]
        target: Vec<String>,
        #[arg(
            long,
            help = "Run one include module and the producers of its declared inputs (repeatable; unions with --target). Module names are the include's `name` field or its file stem"
        )]
        module: Vec<String>,
        #[arg(long, help = "Generate scripts without submitting")]
        dry_run: bool,
        /// Generate job scripts with dependency support and a wrapper script
        #[arg(long, help = "Generate job scripts with dependency support")]
        with_dependencies: bool,
    },
    /// Show the status of submitted cluster jobs. Without job IDs, list the
    /// user's jobs the scheduler currently reports.
    Status {
        #[arg(
            short = 'b',
            long,
            help = "Cluster backend: slurm, pbs, sge, or lsf (defaults to $OXO_FLOW_CLUSTER_BACKEND, then slurm)"
        )]
        backend: Option<String>,
        #[arg(value_name = "JOB_IDS", help = "One or more job IDs")]
        job_ids: Vec<String>,
    },
    /// Cancel submitted cluster jobs.
    Cancel {
        #[arg(
            short = 'b',
            long,
            help = "Cluster backend: slurm, pbs, sge, or lsf (defaults to $OXO_FLOW_CLUSTER_BACKEND, then slurm)"
        )]
        backend: Option<String>,
        #[arg(value_name = "JOB_IDS", help = "One or more job IDs")]
        job_ids: Vec<String>,
    },
    /// Fetch logs for a submitted cluster job.
    Logs {
        #[arg(
            short = 'b',
            long,
            help = "Cluster backend: slurm, pbs, sge, or lsf (defaults to $OXO_FLOW_CLUSTER_BACKEND, then slurm)"
        )]
        backend: Option<String>,
        #[arg(value_name = "JOB_ID", help = "Cluster job ID")]
        job_id: String,
    },
}

#[derive(Subcommand, Debug)]
pub enum ProvenanceAction {
    /// Verify output file checksums from a checkpoint or provenance file.
    Verify {
        #[arg(value_name = "CHECKPOINT_PATH", help = "Path to the checkpoint file")]
        checkpoint: PathBuf,
    },
}

/// Emit the single `--json` document for `test`.
///
/// The command runs several sub-commands; each used to print its own JSON
/// document, so `test --json` produced 3-4 concatenated documents that no
/// parser accepts (audit finding). The steps now run with their JSON output
/// suppressed and this summary is the command's only document.
fn emit_test_json(
    json: bool,
    workflow: &Path,
    status: &str,
    steps: &[serde_json::Value],
    reports: &serde_json::Map<String, serde_json::Value>,
) {
    if !json {
        return;
    }
    let mut output = serde_json::json!({
        "command": "test",
        "workflow": workflow.display().to_string(),
        "status": status,
        "steps": steps,
    });
    if !reports.is_empty() {
        output["reports"] = serde_json::Value::Object(reports.clone());
    }
    println!("{}", serde_json::to_string_pretty(&output).unwrap());
}

/// One `steps[]` entry for the `test --json` document.
fn test_step_json(name: &str, passed: bool) -> serde_json::Value {
    serde_json::json!({
        "step": name,
        "status": if passed { "passed" } else { "failed" },
    })
}

/// Commands whose `--json` document is implemented.
///
/// The global flag is accepted by every subcommand, but most have no JSON
/// output — they used to ignore it and exit 0 with human text (or nothing)
/// on stdout, which scripts read as a malformed document (audit finding).
/// Unsupported commands now reject the flag instead; `report` additionally
/// gets a pointer at its own `-f json`.
fn json_supported(command: &Commands) -> bool {
    match command {
        Commands::Run { .. }
        | Commands::DryRun { .. }
        | Commands::Validate { .. }
        | Commands::Lint { .. }
        | Commands::Test { .. }
        | Commands::Status { .. }
        | Commands::Batch { .. }
        | Commands::Info { .. }
        | Commands::Schema
        | Commands::License { .. }
        | Commands::Ai { .. } => true,
        Commands::Provenance { action } => matches!(action, ProvenanceAction::Verify { .. }),
        Commands::VerifyData { .. } => true,
        _ => false,
    }
}

/// Migration hint for commands removed in v0.13 (issue #76).
///
/// clap would otherwise answer `oxo-flow history` with a bare
/// "unrecognized subcommand" (and `watch` even suggested unrelated
/// commands); scripts and muscle memory deserve the actual migration
/// path (issue #142 H2). Returns the hint when the first positional
/// argument names a removed command.
fn legacy_command_hint() -> Option<String> {
    let mut args = std::env::args_os().skip(1);
    // Global flags are all boolean and take no values, so skipping them
    // by name finds the subcommand position deterministically.
    for arg in args.by_ref() {
        let s = arg.to_string_lossy();
        match s.as_ref() {
            "-v" | "--verbose" | "-q" | "--quiet" | "--json" | "--no-color" | "-h" | "--help"
            | "-V" | "--version" => continue,
            "history" => {
                return Some(
                    "'history' was removed in v0.13 — 'oxo-flow status --timing' shows run \
                     history with wall times."
                        .into(),
                );
            }
            "package" => {
                return Some(
                    "'package' was removed in v0.13 — use 'oxo-flow export' (compose support \
                     included)."
                        .into(),
                );
            }
            "profile" => {
                return Some(
                    "'profile' was removed in v0.13 — use 'run --profile <NAME>' (or dry-run) \
                     instead."
                        .into(),
                );
            }
            "watch" => {
                return Some(
                    "'watch' was removed in v0.13 with no replacement — poll 'oxo-flow status' \
                     (or 'status --timing') instead."
                        .into(),
                );
            }
            _ => return None,
        }
    }
    None
}

#[tokio::main]
async fn main() -> Result<()> {
    // Removed commands (issue #76) get a migration hint instead of clap's
    // bare "unrecognized subcommand" (issue #142 H2).
    if let Some(hint) = legacy_command_hint() {
        eprintln!("{hint}");
        eprintln!("Run 'oxo-flow --help' to see the current command set.");
        std::process::exit(1);
    }

    // clap prints -h/--help during parsing, so pick the banner variant up
    // front: colors only on an interactive terminal, and never when
    // NO_COLOR is set or --no-color was passed.
    let use_color = std::io::IsTerminal::is_terminal(&std::io::stdout())
        && std::env::var_os("NO_COLOR").is_none()
        // args_os, not args: clap itself works on OsString and never
        // panics on non-UTF-8 arguments (bioinformatics paths included).
        && !std::env::args_os().any(|arg| arg == "--no-color");
    let matches = {
        let mut command = Cli::command();
        // Platform config file (oxo-flow.web.toml / OXO_FLOW_CONFIG)
        // supplies the lowest-precedence defaults for `serve` — the same
        // tier the standalone web binary honors (flag > env > file >
        // built-in). The desktop app ships the CLI, so it must resolve
        // the file too.
        if let Some(cfg) = oxo_flow_web::config::load() {
            if let Some(mode) = cfg.server.mode {
                let mode = clap::builder::OsStr::from(mode);
                command = command
                    .mut_subcommand("serve", |c| c.mut_arg("mode", |a| a.default_value(mode)));
            }
            if let Some(host) = cfg.server.host {
                let host = clap::builder::OsStr::from(host);
                command = command
                    .mut_subcommand("serve", |c| c.mut_arg("host", |a| a.default_value(host)));
            }
            if let Some(port) = cfg.server.port {
                let port = clap::builder::OsStr::from(port.to_string());
                command = command
                    .mut_subcommand("serve", |c| c.mut_arg("port", |a| a.default_value(port)));
            }
            if let Some(base_path) = cfg.server.base_path {
                let base_path = clap::builder::OsStr::from(base_path);
                command = command.mut_subcommand("serve", |c| {
                    c.mut_arg("base_path", |a| a.default_value(base_path))
                });
            }
        }
        command = command.help_template(if use_color {
            banner::HELP_TEMPLATE
        } else {
            banner::HELP_TEMPLATE_PLAIN
        });
        if !use_color {
            // Also disable clap's own help styling (bold headings etc.).
            command = command.color(clap::ColorChoice::Never);
        }
        command.get_matches()
    };
    let cli = Cli::from_arg_matches(&matches)?;

    // One colour policy for every output surface: `--no-color`, NO_COLOR,
    // and background runs (whose stderr is the run log, issue #194 A3).
    // The tracing layer used to key off the redirect variable alone, so
    // `--no-color`/NO_COLOR still painted ANSI escapes into redirected
    // stderr (audit finding).
    // #541: stderr is where the narrative goes — a redirected stderr must
    // not receive ANSI escapes even when stdout is still a TTY (the exact
    // population `2> err.log` serves). Both the colored crate override and
    // the tracing subscriber's with_ansi read this.
    use std::io::IsTerminal;
    let color_disabled = cli.no_color
        || std::env::var_os("NO_COLOR").is_some()
        || std::env::var_os("OXO_FLOW_STDERR_ALREADY_REDIRECTED").is_some()
        || !std::io::stderr().is_terminal();
    if color_disabled {
        colored::control::set_override(false);
    }

    // ── `--json` support gate (audit finding) ───────────────────────────
    // The global flag is accepted everywhere but only a subset of commands
    // emit a JSON document; the rest silently ignored it. Fail fast with
    // the command's name instead of exiting 0 with unusable stdout.
    if cli.json && !json_supported(&cli.command) {
        let name = matches.subcommand_name().unwrap_or("this command");
        let hint = match &cli.command {
            Commands::Report { .. } => " — `report` selects JSON with `-f json`",
            _ => "",
        };
        anyhow::bail!(
            "`oxo-flow {name}` does not support --json yet{hint}.\n  \
             Drop --json to run it normally, or use a command whose JSON \
             document is implemented (run, dry-run, validate, lint, test, \
             status, batch, info, schema, license, ai, provenance verify, \
             verify-data)."
        );
    }
    // A background run never executes the workflow in the foreground
    // process, so there is no run summary to write: reject the combination
    // rather than exit 0 with zero bytes on stdout (audit finding).
    if cli.json
        && let Commands::Run {
            background: true, ..
        } = &cli.command
    {
        anyhow::bail!(
            "`run --background` cannot be combined with --json: the foreground \
             invocation only launches the detached child, so no run summary \
             exists to emit.\n  Drop --json, or run in the foreground to get \
             the machine-readable summary."
        );
    }

    let default_level = if cli.quiet {
        "error"
    } else if cli.verbose {
        "debug"
    } else {
        "info"
    };
    // The archived run log (`.oxo-flow/logs/oxo-flow.log`) must always carry
    // the structured execution events (workflow_started / rule_completed /
    // …) — they are the machine-readable audit trail (#194 B3), not debug
    // chatter. A user's `RUST_LOG=warn` must not silently strip them, so the
    // `execution_event` target is pinned to `info` ahead of any env filter.
    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(default_level));
    let env_filter = env_filter.add_directive(
        "execution_event=info"
            .parse()
            .expect("static execution_event directive parses"),
    );
    tracing_subscriber::fmt()
        .with_env_filter(env_filter)
        .with_target(false)
        .with_ansi(!color_disabled)
        // Logs go to stderr so machine-readable stdout (graph DOT output,
        // --json, pipes into dot/other tools) stays clean.
        .with_writer(crate::logging::TeeWriter)
        .init();

    // Suppress banner in quiet mode
    crate::commands::set_quiet_mode(cli.quiet);

    match cli.command {
        Commands::Run {
            workflow,
            jobs,
            keep_going,
            workdir,
            log_file,
            target,
            module,
            retry,
            timeout,
            resume_failed,
            profile,
            max_threads,
            max_memory,
            skip_env_setup,
            skip_ref_build,
            cache_dir,
            provenance,
            bundle,
            yes,
            args,
            config_overrides,
            ai_recover,
            ai_max_retries,
            samples_filter,
            rerun,
            no_report_snapshot,
            max_submitted,
            background,
            allow_protected_drift,
        } => {
            // ── Background runs (issue #158) ─────────────────────────────
            // The foreground process never executes the workflow: it spawns
            // a DETACHED child that re-runs the same command without
            // --background (same argv, so --bundle/--profile/--samples/…
            // pass through verbatim), writes the child's pid, prints a
            // one-line summary on stderr, and exits 0. The child then runs
            // the normal flow — checkpoint, workdir lock, and resume
            // semantics are unchanged. Bundle confirmation still applies in
            // the child; a detached child cannot prompt, so bundle runs
            // need --yes (fail-closed, same as any non-interactive session).
            if background {
                let bg_workdir = crate::background::background_workdir_for_run(
                    workflow.as_deref(),
                    workdir.as_deref(),
                    bundle.as_deref(),
                )?;
                let log_path =
                    crate::background::resolve_log_path(&bg_workdir, log_file.as_deref());
                let args = crate::background::strip_background_flag(std::env::args_os());
                crate::background::launch_in_background(&args, &bg_workdir, &log_path)?;
                std::process::exit(0);
            }
            use anyhow::Context as _;
            use colored::Colorize as _;
            #[allow(unused_imports)]
            use std::io::BufRead as _;
            let (wf, wd) = if let Some(bundle_path) = bundle {
                let (extracted_wf, extracted_dir) =
                    crate::commands::bundle::extract_and_verify_bundle(&bundle_path)?;
                // Respect explicit -d flag; otherwise use extracted dir
                let effective_wd = workdir.unwrap_or_else(|| extracted_dir.clone());

                // Confirmation gate for bundle execution (remote code safety).
                // After checksum verification, print what's about to run and
                // require explicit confirmation unless --yes is set.
                // The manifest describes the JUST-VERIFIED extracted bundle —
                // a user-supplied -d directory may hold a stale manifest.
                if !yes {
                    let manifest_path =
                        crate::commands::bundle::find_manifest_in_dir(&extracted_dir)?;
                    let manifest_json = std::fs::read_to_string(&manifest_path)
                        .context("failed to read bundle manifest")?;
                    let manifest: serde_json::Value =
                        serde_json::from_str(&manifest_json).context("failed to parse manifest")?;

                    eprintln!();
                    eprintln!("{}", "Bundle Verification Complete".bold().green());
                    eprintln!(
                        "  Workflow: {}",
                        manifest["workflow"].as_str().unwrap_or("unknown")
                    );
                    eprintln!(
                        "  Format:   {}",
                        manifest["format"].as_str().unwrap_or("unknown")
                    );
                    eprintln!(
                        "  Version:  {}",
                        manifest["oxo_flow_version"].as_str().unwrap_or("unknown")
                    );
                    if let Some(resources) = manifest.get("resources")
                        && let Some(recommendations) = resources.get("recommendations")
                    {
                        eprintln!("  Resources:");
                        if let Some(t) = recommendations["min_threads"].as_u64() {
                            eprintln!("    Min threads: {}", t.to_string().cyan());
                        }
                        if let Some(m) = recommendations["min_memory_mb"].as_u64() {
                            eprintln!(
                                "    Min memory:  {} MB ({:.1} GB)",
                                m.to_string().cyan(),
                                m as f64 / 1024.0
                            );
                        }
                        if let Some(g) = recommendations["min_gpu"].as_u64()
                            && g > 0
                        {
                            eprintln!("    Min GPU:     {}", g.to_string().cyan());
                        }
                    }
                    eprintln!("  Source:   {}", bundle_path.display());

                    let can_prompt = crate::commands::bundle::can_prompt_for_confirmation(
                        cli.json,
                        std::io::IsTerminal::is_terminal(&std::io::stderr()),
                        std::io::IsTerminal::is_terminal(&std::io::stdin()),
                    );
                    if !can_prompt {
                        // The freshly extracted dir will never run — clean it
                        // up instead of leaking /tmp/oxo-bundle-*.
                        let _ = std::fs::remove_dir_all(&extracted_dir);
                        anyhow::bail!(
                            "Running a bundle requires confirmation, and this session cannot prompt for it. \
                             Use --yes to confirm in CI, scripts, or with --json.\n\
                             Bundle: {}",
                            bundle_path.display()
                        );
                    }

                    eprintln!();
                    eprint!("  {} Proceed with execution? [y/N] ", "⚠".yellow());
                    use std::io::Write as _;
                    std::io::stderr().flush().ok();
                    let mut input = String::new();
                    std::io::stdin().read_line(&mut input)?;
                    if !input.trim().eq_ignore_ascii_case("y")
                        && !input.trim().eq_ignore_ascii_case("yes")
                    {
                        let _ = std::fs::remove_dir_all(&extracted_dir);
                        anyhow::bail!("execution cancelled by user");
                    }
                }

                (Some(extracted_wf), Some(effective_wd))
            } else {
                (workflow, workdir)
            };
            // Merge direct `KEY=VALUE` / `--KEY VALUE` overrides with `--arg KEY=VALUE`
            // (backward compat). Later entries win on duplicate keys.
            let mut merged_args = config_overrides;
            merged_args.extend(args);
            run_command(
                wf,
                jobs,
                keep_going,
                wd,
                log_file,
                target,
                module,
                retry,
                timeout,
                resume_failed,
                profile,
                max_threads,
                max_memory,
                skip_env_setup,
                skip_ref_build,
                cache_dir,
                provenance,
                cli.json,
                merged_args,
                ai_recover,
                ai_max_retries,
                samples_filter,
                rerun,
                no_report_snapshot,
                max_submitted,
                None, // checkpoint_override (run derives the workdir default)
                allow_protected_drift,
            )
            .await?
        }
        Commands::Resume {
            checkpoint,
            jobs,
            ai_recover,
            ai_max_retries,
            keep_going,
            timeout,
            workdir,
            no_report_snapshot,
            background,
        } => {
            // ── Background resume (issue #158) ───────────────────────────
            // Same detach contract as `run --background`; the workdir comes
            // from the checkpoint when --workdir is not given, exactly as
            // `resume_command` resolves it.
            if background {
                let bg_workdir = crate::background::background_workdir_for_resume(
                    &checkpoint,
                    workdir.as_deref(),
                )?;
                let log_path = crate::background::resolve_log_path(&bg_workdir, None);
                let args = crate::background::strip_background_flag(std::env::args_os());
                crate::background::launch_in_background(&args, &bg_workdir, &log_path)?;
                std::process::exit(0);
            }
            resume_command(
                checkpoint,
                jobs,
                ai_recover,
                ai_max_retries,
                keep_going,
                timeout,
                workdir,
                no_report_snapshot,
            )
            .await?
        }
        Commands::DryRun {
            workflow,
            target,
            module,
            ai,
            ai_max_retries,
            samples_filter,
            workdir,
            profile,
            skip_ref_build,
            cache_dir,
            args,
            config_overrides,
            rerun,
            resume_failed,
        } => {
            // Same merge rule as `run`: trailing KEY=VALUE positionals first,
            // --arg entries after (later entries win on duplicate keys).
            let mut merged_args = config_overrides;
            merged_args.extend(args);
            dry_run_command(
                workflow,
                target,
                module,
                cli.verbose,
                cli.json,
                ai,
                ai_max_retries,
                samples_filter,
                workdir,
                profile,
                skip_ref_build,
                cache_dir,
                merged_args,
                rerun,
                resume_failed,
            )
            .await?
        }
        Commands::Validate {
            workflow,
            as_include,
            ai,
        } => {
            validate_command(workflow, as_include, cli.json, ai).await?;
        }
        Commands::Info { workflow, format } => info_command(workflow, format)?,
        Commands::Init { name, dir, force } => init_command(name, dir, force)?,
        Commands::Template {
            template,
            output,
            ai,
            from_url,
            from_file,
            ai_max_retries,
            ai_team_profile,
            ai_attempts,
        } => {
            // CLI flag wins over `[ai] team_profile` in the resolved config.
            let team_profile = match ai_team_profile.as_deref() {
                Some(flag) => Some(
                    flag.parse::<oxo_flow_ai::agent::team::TeamProfile>()
                        .map_err(|e| anyhow::anyhow!(e))?,
                ),
                None => oxo_flow_ai::AI.config().ok().and_then(|c| c.team_profile),
            };
            template_command(
                template,
                output,
                ai,
                from_url,
                from_file,
                crate::commands::project::TemplateAiOptions {
                    max_retries: ai_max_retries,
                    team_profile,
                    attempts: ai_attempts,
                },
            )
            .await?
        }
        Commands::Ai {
            action,
            workflow,
            step,
            level,
            json,
        } => {
            // The global --json and the action's own flag are equivalent:
            // `ai explain` is the only action with a JSON document, and a
            // user may place the global flag before the subcommand.
            let json = json || cli.json;
            // #541: `ai --json` is whitelisted as a JSON command, so an
            // action-less invocation answers with the provider status
            // document instead of a usage error. (With an explicit action
            // the flag keeps its per-action meaning.)
            if action.is_none() && json && workflow.is_none() && step.is_none() {
                let provider = oxo_flow_ai::provider::create_provider_from_env();
                let name = provider.name();
                let doc = serde_json::json!({
                    "command": "ai",
                    "provider": name,
                    "model": provider.model(),
                    "configured": !matches!(name, "noop" | "disabled"),
                    "hint": "use `oxo-flow ai explain <workflow.oxoflow>` for AI analysis",
                });
                println!("{}", serde_json::to_string_pretty(&doc)?);
                return Ok(());
            }
            // --step/--level/--json/workflow only make sense with 'explain'.
            let explain_args = workflow.is_some() || step.is_some() || json;
            match action.as_deref() {
                Some("explain") => {
                    let Some(workflow) = workflow else {
                        anyhow::bail!(
                            "'ai explain' requires a workflow file:\n  oxo-flow ai explain <workflow.oxoflow> [--step <rule>] [--level beginner|expert] [--json]"
                        );
                    };
                    crate::commands::ai_explain::ai_explain_command(
                        &workflow,
                        step.as_deref(),
                        level,
                        json,
                    )
                    .await?
                }
                Some("test") => {
                    if explain_args {
                        anyhow::bail!("'ai test' takes no workflow/--step/--json arguments");
                    }
                    ai_test_command().await?
                }
                Some("setup") => {
                    if explain_args {
                        anyhow::bail!("'ai setup' takes no workflow/--step/--json arguments");
                    }
                    ai_setup_command().await?
                }
                None => {
                    if explain_args {
                        anyhow::bail!(
                            "workflow/--step/--json require the 'explain' action:\n  oxo-flow ai explain <workflow.oxoflow>"
                        );
                    }
                    ai_status_command().await?
                }
                Some(other) => {
                    anyhow::bail!(
                        "unknown ai action '{other}' — expected one of: test, setup, explain"
                    )
                }
            }
        }
        Commands::Graph {
            workflow,
            format,
            output,
            expanded,
            granularity,
        } => handle_graph(workflow, format, output, expanded, granularity)?,
        Commands::Status {
            checkpoint,
            workdir,
            timing,
            limit,
        } => handle_status(checkpoint, workdir, cli.json, timing, limit).await?,
        Commands::Pull { url, output } => crate::commands::pull::pull_command(&url, output).await?,
        Commands::Config { action } => crate::commands::infra::handle_config(action)?,
        Commands::Diff {
            workflow_a,
            workflow_b,
        } => handle_diff(workflow_a, workflow_b)?,
        Commands::Debug {
            workflow,
            rule_name,
            ai,
        } => debug_command(workflow, rule_name, ai).await?,
        Commands::Clean {
            workflow,
            dry_run,
            force,
            orphans,
            workdir,
        } => clean_command(workflow, dry_run, force, orphans, workdir)?,
        Commands::Env { action } => env_command(action).await?,
        Commands::Format {
            workflow,
            output,
            check,
        } => format_command(workflow, output, check)?,
        Commands::Lint {
            workflow,
            strict,
            ai,
        } => lint_command(workflow, strict, cli.json, ai).await?,
        Commands::Touch {
            workflow,
            rules,
            workdir,
            args,
            config_overrides,
        } => {
            let mut cli_args = config_overrides;
            cli_args.extend(args);
            touch_command(workflow, rules, workdir, cli_args)?
        }
        Commands::Report {
            workflow,
            format,
            output,
            checkpoint_path,
            ai,
            workdir,
            ci,
            no_timestamps,
            strict,
            list_sections,
            run_dir,
            failed,
            plan,
            init_template,
            list_templates,
            r_data,
            diff,
            acct,
            versions_yml,
        } => {
            handle_report(crate::commands::output::ReportArgs {
                workflow,
                format,
                output,
                checkpoint_path,
                ai,
                workdir,
                ci,
                no_timestamps,
                strict,
                list_sections,
                run_dir,
                failed,
                plan,
                init_template,
                list_templates,
                r_data,
                diff,
                acct,
                versions_yml,
            })
            .await?
        }
        Commands::Serve {
            mode,
            host,
            port,
            base_path,
            open_browser,
        } => crate::commands::web::handle_serve(mode, host, port, base_path, open_browser).await?,
        Commands::Completions { shell } => handle_completions(shell)?,
        Commands::Export {
            workflow,
            format,
            output,
        } => handle_export(workflow, format, output)?,
        Commands::Cluster { action } => cluster_command(action).await?,
        Commands::Batch {
            template,
            items,
            jobs,
            stop_on_error,
            file,
            json_output: json,
            dry_run,
            workdir,
            environment,
            generate_workflow,
            output,
        } => {
            batch_command(
                template,
                items,
                jobs,
                stop_on_error,
                file,
                // The global --json and the legacy --json-output both drive
                // JSON output (alignment audit 2026-08-14): the global flag
                // was silently ignored here before.
                cli.json || json,
                dry_run,
                workdir,
                environment,
                generate_workflow,
                output,
            )
            .await?
        }
        Commands::Provenance { action } => match action {
            ProvenanceAction::Verify { checkpoint } => {
                provenance_verify_command(checkpoint, cli.json)?
            }
        },
        Commands::VerifyData { tree, from } => verify_data_command(tree, from, cli.json)?,
        Commands::Schema => {
            let schema = include_str!("../schema/oxoflow-v1.schema.json");
            println!("{schema}");
        }
        Commands::Test {
            workflow,
            output,
            run,
            jobs,
            samples_filter,
            deep,
            workdir,
            profile,
            target,
            timeout,
            retry,
            keep_going,
        } => {
            use colored::Colorize;
            eprintln!(
                "{} Running test suite for {}\n",
                "🧪".bold(),
                workflow.display()
            );
            // `test --json` owns stdout: every step runs with its own JSON
            // output suppressed (their human output is on stderr) and this
            // handler emits ONE document describing all steps — it used to
            // print 3-4 concatenated documents (audit finding).
            let mut steps: Vec<serde_json::Value> = Vec::new();
            // Sub-documents of steps that have one (currently deep-check):
            // the aggregate stays a single document, but no detail is lost.
            let mut reports = serde_json::Map::new();
            macro_rules! step {
                ($name:literal, $call:expr) => {{
                    let result = $call;
                    steps.push(test_step_json($name, result.is_ok()));
                    if let Err(e) = result {
                        emit_test_json(cli.json, &workflow, "failed", &steps, &reports);
                        return Err(e);
                    }
                }};
            }
            // 1. Validate
            eprintln!("{} Validation...", "1.".bold());
            step!(
                "validate",
                validate_command(workflow.clone(), false, false, false).await
            );
            // 2. Lint
            eprintln!("{} Lint...", "2.".bold());
            step!(
                "lint",
                lint_command(workflow.clone(), false, false, false).await
            );
            // 3. Dry-run
            eprintln!("{} Dry-run...", "3.".bold());
            step!(
                "dry-run",
                dry_run_command(
                    Some(workflow.clone()),
                    target.clone(),
                    Vec::new(), // module
                    cli.verbose,
                    false,
                    false,
                    None,
                    samples_filter.clone(),
                    workdir.clone(),
                    profile.clone(),
                    false,
                    None,
                    vec![],
                    false,
                    false,
                )
                .await
            );
            // 4. Optional: deep health checks (issue #64)
            if deep {
                eprintln!("{} Deep checks...", "4.".bold());
                match crate::commands::quality::run_deep_check(&workflow, workdir.as_deref()) {
                    Ok((report, document)) => {
                        steps.push(test_step_json("deep-check", report.passed));
                        reports.insert("deep-check".to_string(), document);
                        if !report.passed {
                            // Same verdict as `deep-check`: D-errors fail.
                            emit_test_json(cli.json, &workflow, "failed", &steps, &reports);
                            std::process::exit(1);
                        }
                    }
                    Err(e) => {
                        steps.push(test_step_json("deep-check", false));
                        emit_test_json(cli.json, &workflow, "failed", &steps, &reports);
                        return Err(e);
                    }
                }
            }
            // 5. Optional: run with --run flag
            if run {
                eprintln!("{} Execution...", if deep { "5." } else { "4." }.bold());
                step!(
                    "run",
                    run_command(
                        Some(workflow.clone()),
                        jobs,
                        keep_going,      // keep_going
                        workdir.clone(), // workdir
                        None,            // log_file (default path)
                        target.clone(),  // target
                        Vec::new(),      // module
                        retry,           // retry
                        timeout.clone(), // timeout
                        false,           // resume_failed
                        profile.clone(), // profile
                        0,               // max_threads
                        0,               // max_memory
                        false,           // skip_env_setup
                        false,           // skip_ref_build
                        None,            // cache_dir
                        false,           // provenance
                        false,           // json (the single test document is emitted here)
                        vec![],          // cli_args
                        false,           // ai_recover
                        None,            // ai_max_retries
                        samples_filter.clone(),
                        false, // rerun (test mode: normal up-to-date checks)
                        false, // no_report_snapshot (test mode keeps the standard run behavior)
                        None,  // max_submitted (cluster queue cap — test keeps the profile's)
                        None,  // checkpoint_override (test derives the workdir default)
                        false, // allow_protected_drift (test honors the protection gate)
                    )
                    .await
                );
            }
            // Optional: verify output file existence
            if let Some(output_path) = output {
                if output_path.exists() {
                    eprintln!(
                        "{} Output file exists: {}",
                        "✓".green().bold(),
                        output_path.display()
                    );
                    steps.push(serde_json::json!({
                        "step": "output",
                        "status": "passed",
                    }));
                } else {
                    eprintln!(
                        "{} Output file not found: {}",
                        "✗".red().bold(),
                        output_path.display()
                    );
                    steps.push(serde_json::json!({
                        "step": "output",
                        "status": "failed",
                    }));
                    emit_test_json(cli.json, &workflow, "failed", &steps, &reports);
                    std::process::exit(1);
                }
            }
            emit_test_json(cli.json, &workflow, "passed", &steps, &reports);
            eprintln!("\n{} All checks passed.", "✓".green().bold());
        }
        Commands::Publish {
            workflow,
            output,
            with_lockfiles,
            format,
        } => publish_command(workflow, output, with_lockfiles, format)?,
        Commands::License { path } => handle_license(path, cli.json)?,
    }

    Ok(())
}