aion-cli 0.31.0

//! The `aion` command line: workflow operations over gRPC and the embedded
//! Aion workflow server (`aion server`).

use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::time::Duration;

use aion_client::{Client, ClientBuilder, ClientError, ListRequest};
use aion_core::{Payload, WorkflowStatus};
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use serde_json::Value;

use crate::output::{
    AcknowledgementOutput, QueryOutput, RetireOutput, describe_output, pause_output, print_json,
    reopen_output, resume_output, to_value,
};
use crate::payload::{
    json_payload, parse_run_id, parse_status, parse_workflow_id, payload_to_json,
};
use crate::start::{start_workflow, start_workflow_handle};

mod awl;
mod awl_doc;
mod awl_guide;
mod awl_recipe;
mod awl_scaffold;
mod boot_narration;
mod check_deterministic;
mod console;
mod deploy;
mod dev;
mod generate;
mod handover;
mod harness;
mod history;
mod input;
mod inspect;
mod launch;
mod new;
mod one_motion;
mod output;
mod package;
mod payload;
mod render;
mod server;
mod server_restart;
mod server_status;
mod server_stop;
mod settings;
mod start;
mod store_harden;
mod tail;
mod update;
mod update_index;
mod worker_agent;
mod worker_awl;
mod worker_managed;
mod worker_shell;
mod worker_surface;

const DEFAULT_ENDPOINT: &str = "127.0.0.1:50051";
const DEFAULT_NAMESPACE: &str = "default";
const DEFAULT_SUBJECT: &str = "cli-user";
const QUERY_DEADLINE: Duration = Duration::from_secs(5);

/// Aion workflow command line.
#[derive(Debug, Parser)]
#[command(
    name = "aion",
    version,
    about = "Operate Aion workflows over gRPC and run the Aion workflow server. \
             Bare `aion` starts the server in the background and opens the ops console."
)]
struct Cli {
    /// gRPC server endpoint a client dials. A missing scheme is interpreted as
    /// http://. Not a bind address: `aion server` takes `--grpc-address`.
    ///
    /// Deliberately has no clap `default_value`, so the difference between
    /// "the operator named an endpoint" and "nobody said" survives parsing.
    /// `aion server` needs that difference to refuse a flag it cannot honour;
    /// a default applied here would erase it before anyone could ask.
    #[arg(long, global = true)]
    endpoint: Option<String>,
    /// Workflow namespace used for operations.
    #[arg(long, global = true, default_value = DEFAULT_NAMESPACE)]
    namespace: String,
    /// Caller subject metadata sent to the server.
    #[arg(long, global = true, default_value = DEFAULT_SUBJECT)]
    subject: String,
    /// Bearer token for authenticated servers. Overrides the `AION_TOKEN`
    /// environment variable; when both are absent the development
    /// header paths apply.
    #[arg(long, global = true)]
    token: Option<String>,
    /// Print formatted, human-readable JSON.
    #[arg(long, global = true)]
    pretty: bool,
    /// Bare `aion` only: run the launched server in this terminal instead of
    /// the background.
    #[arg(long)]
    foreground: bool,
    /// Bare `aion` only: do not open the ops console in a browser.
    #[arg(long)]
    no_open: bool,
    /// Absent means the bare launcher: start the server in the background,
    /// wait until it is live, report in plain language, open the console.
    #[command(subcommand)]
    command: Option<Command>,
}

#[derive(Debug, Subcommand)]
enum Command {
    /// Run the Aion workflow server (HTTP, gRPC, WebSocket, remote workers).
    ///
    /// Loads configuration from `--config`, the environment, and command-line
    /// overrides, serves until a termination signal arrives, then drains
    /// in-flight activities gracefully before exiting. The control verbs
    /// `aion server stop` and `aion server status` act on a server already
    /// running under this machine's Aion home.
    Server(server::ServerCommand),
    /// Stop the server running under this machine's Aion home.
    ///
    /// The front-door spelling of `aion server stop`, and the SAME code path:
    /// bare `aion` launches a server, so `aion stop` stops one. It verifies
    /// the recorded incarnation before signalling, waits out the drain, and
    /// reports the drain outcome by name — at any stage of the server's life,
    /// including one still booting.
    Stop(crate::server_stop::StopArgs),
    /// Install a version of Aion and hand the running server over to it.
    ///
    /// Resolves the target (the latest stable release, or `--version`),
    /// installs it with `cargo install` into cargo's own root, and — when a
    /// server is running on this home — drains it and starts the freshly
    /// installed binary in its place, narrating the boot until it serves.
    /// Reports both identities. An install that fails touches nothing running.
    Update(crate::update::UpdateArgs),
    /// Operate a store's data directory directly, with no server running.
    ///
    /// `aion store harden` brings every file under the store to its private
    /// mode in one walk — the one-time step for a store created before files
    /// were private from their own creation, or one loosened by hand.
    Store(store_harden::StoreCommand),
    /// The instant authoring loop: watch a workflow project and hot-load on save.
    ///
    /// Watches the project's `src/` tree and, on every save, rebuilds and
    /// type-checks it through the external `gleam` binary, repackages it into a
    /// content-hash-versioned `.aion`, and hot-loads that version into the
    /// running server named by `--endpoint` — with no engine restart. A run
    /// already in-flight keeps executing on the immutable version it started
    /// with; only fresh runs pick up the reload. The `gleam` binary path is
    /// required (`--gleam-path`); there is no default. An optional
    /// `--debounce-ms` coalesces an editor's multi-event save into one rebuild.
    Dev(dev::DevArgs),
    /// Author AWL workflow documents (rev-2 surface): check, format, emit
    /// Gleam, derive JSON Schema, and read the language's own guide.
    ///
    /// Runs entirely locally and never connects to a server. `check` parses
    /// and typechecks a document, printing compiler-style
    /// `<file>:<line>:<column>: error: <message>` diagnostics to stderr;
    /// `fmt` rewrites the document in place with the canonical printer (the
    /// printer is the formatter — there is no check mode); `emit` lowers a
    /// checked document to a Gleam workflow module (the stopgap execution
    /// target); `schema` derives draft 2020-12 JSON Schema from the same
    /// checked document; `guide` prints what any word of the language means.
    ///
    /// The documents behind this surface, all in the repository:
    /// `docs/authoring/AWL.md` introduces the language and the loop;
    /// `docs/assistant/grounding/AWL-REFERENCE.md` is the complete
    /// reference every `guide` entry cites by section;
    /// `docs/assistant/grounding/AWL-AUTHORING.md` walks from an
    /// objective to a checked document;
    /// `docs/assistant/grounding/WORKERS.md` says what actually serves a
    /// declared action; `docs/assistant/grounding/COMMANDS.md` covers
    /// every command in this loop with its failure modes; and
    /// `docs/workers/DECLARED-COMMANDS.md` must be read before writing a
    /// `run "…"` body, because there is no sandbox.
    Awl {
        #[command(subcommand)]
        command: awl::AwlCommand,
    },
    /// Run worker processes supplied by the Aion distribution.
    Worker {
        #[command(subcommand)]
        command: WorkerCommand,
    },
    /// Operate a running Aion server or package workflows locally.
    #[command(flatten)]
    Client(ClientCommand),
}

#[derive(Debug, Subcommand)]
enum WorkerCommand {
    /// Serve strict manifest-wired shell commands on an Aion task queue.
    Shell(worker_shell::ShellArgs),
    /// Serve the queue a WORKER DOCUMENT declares — a `.awl` file whose
    /// opening declaration is `worker <queue>` rather than `workflow <name>`.
    ///
    /// The document is the whole wiring: it names the queue, declares each
    /// action's typed contract, and carries the command that serves it. A
    /// workflow document is refused by name, as is a document that does not
    /// check cleanly — with the checker's own diagnostics, at the desk,
    /// rather than as a remote refusal after dialling.
    ///
    /// The advertised schemas are derived through the same seam the deploy
    /// path uses, so this worker's advertisement and the server's stored
    /// contract for the queue are two independent derivations of two
    /// documents; the server's worker-contract admission is what holds them
    /// equal, and refuses a worker document that has drifted from the
    /// workflow that calls it.
    ///
    /// Every action runs through the same executor the server uses for a
    /// workflow document's declared bodies: argv parsed once, each parameter
    /// substituted as one whole argument, no shell, the host environment
    /// CLEARED except `PATH`, and process-group containment on cancellation.
    /// The document's capture selector decides only what a successful
    /// command's result is — the outcome record for a bare `run`, trimmed
    /// stdout for `run text`, decoded stdout for `run json`.
    ///
    /// `--identity` is derived from the queue as `<queue>-awl-worker` when
    /// absent. Every other connection setting must be supplied by a flag or
    /// its environment variable, and every missing one is named at once.
    ///
    /// `--check` prints the exact surface this worker would advertise, with
    /// the command each action runs, then exits without connecting.
    Awl(worker_awl::AwlWorkerArgs),
    /// Serve an agent harness compiled into this binary on the queue an `.awl`
    /// document declares — an agent worker without writing a crate.
    ///
    /// The document is authoritative and is the only source of the advertised
    /// schemas: an agent activity's output is decided by the agent, so there is
    /// no Rust type to derive one from. The task queue is the `worker` block's
    /// name — derived when the document declares exactly one block, selected with
    /// `--task-queue` when it declares several — and every action on that queue
    /// whose body the document does NOT carry is served. An action WITH a body is
    /// run by the server from the deployed contract, so no worker serves it.
    /// Serving is all-or-nothing: a partial advertisement is admitted on some
    /// actions and then dispatched one it never declared, so a queue is served
    /// whole or the verb refuses, by name.
    ///
    /// The worker dials LIMINAL candidate addresses (`--liminal-address`,
    /// repeatable for redial candidates), not the global gRPC `--endpoint`: the
    /// agent-harness seam exists only on the liminal worker transport. Every
    /// connection setting comes from a flag or from the named environment
    /// variable, the flag wins, and nothing is invented — a missing setting is
    /// reported by name, with all of them listed at once.
    ///
    /// The LAUNCH is declared in the document, in the `worker` block's `harness`
    /// section, and is not a flag: `kind` (`acp` or `norn`), `concurrency`, the
    /// three reconnect settings, `env_pass`, and the per-kind settings — `command`
    /// (absolute), `cwd` (absolute), `permission` (`deny` or `allow-once`, no
    /// default, because it decides whether an unattended agent may write files and
    /// run commands) and `exit_grace` for `acp`; the optional `binary` for `norn`;
    /// optional `args` for either. Nothing is inferred: the checker refuses a
    /// section missing any field of ITS DECLARED KIND's required set, by name, at
    /// `aion awl check`. A `kind` this binary was not built with is refused here,
    /// naming what the build does carry.
    ///
    /// `env_pass` is the agent subprocess's WHOLE environment: the document names
    /// the variables, this machine supplies their values, and a variable that is
    /// not named is absent in the agent whatever this worker's own environment
    /// carries. Nothing is inherited.
    ///
    /// `--check` prints the exact action surface this worker would advertise, the
    /// resolved harness, and the intervention controls that harness advertises,
    /// then exits without connecting. Those descriptors are the whole admission
    /// decision on a queue carrying a deployed contract, so reading them
    /// beforehand turns a remote contract-mismatch refusal into something
    /// checkable at the desk.
    Agent(Box<worker_agent::AgentArgs>),
    /// Report the managed worker fleet the SERVER supervises.
    ///
    /// Every worker is printed with BOTH its durable desired state and its live
    /// actual state, because either alone is the misleading half: "desired
    /// running, state stopped" is the condition an operator most needs to see.
    /// A server with no `[worker_supervision]` section reports
    /// `commissioned: false` and the remedy rather than an empty, calm-looking
    /// list.
    Status,
    /// Start a managed worker: record the intent durably, then supervise it.
    ///
    /// Idempotent — a worker already supervised is reported as it is, without a
    /// second process.
    Start {
        /// Worker deployment name (the primary key used at `/worker-deployments`).
        name: String,
    },
    /// Stop a managed worker. Terminal: it is not restarted until started.
    ///
    /// The server reports success only once it has PROVEN the worker's process
    /// group empty, so a successful stop is an OS-level fact rather than a
    /// bookkeeping change.
    Stop {
        /// Worker deployment name.
        name: String,
    },
    /// Restart a managed worker, leaving its desired state at running.
    Restart {
        /// Worker deployment name.
        name: String,
    },
}

#[derive(Debug, Subcommand)]
enum ClientCommand {
    /// Scaffold a new Aion workflow project from a template.
    ///
    /// Creates `<name>/` in the current directory. The default `awl`
    /// template emits a checked, runnable AWL document (`<name>.awl`), a
    /// ready development server config (`aion.toml`), a `.gitignore`, and a
    /// README with the exact next steps — no build manifest and no worker
    /// required. The five Gleam templates instead emit a buildable Gleam
    /// workflow with a typed entry function, `workflow.toml`, and JSON
    /// Schemas. Runs entirely locally and refuses to write into a non-empty
    /// directory.
    ///
    /// Templates: `awl` (default; typed input and output, three steps, one
    /// `if`/`else` decision), `hello-world` (minimal Gleam start →
    /// complete), `approval-flow` (signal vs durable timeout race plus a
    /// status query), `saga` (worker-served activities, workflow-driven
    /// retries, and refund compensation), `agent` (a durable agent loop —
    /// scout → act → verify → signal-gated human review, parameterised by
    /// prompts; `--worker rust` required), and `dev-pipeline` (the durable
    /// dev pipeline; `--worker rust` required). `--worker rust` additionally
    /// scaffolds a `worker/` crate serving the template's activities.
    New(new::NewArgs),
    /// Generate every derived artifact from a package's authored Gleam types.
    ///
    /// Runs locally and drives the Gleam toolchain (types-first, ADR-014):
    /// reads the boundary types declared in `src/<package>_io.gleam` via
    /// `gleam export package-interface`, then writes the codecs module, the
    /// emitted `schemas/*.json` artifacts, and — when the package declares
    /// activities via `src/<package>_activities.gleam` `manifest()` — the
    /// typed `activity.new` wrappers, the per-tier worker module, the remote
    /// wire-compat golden, and the `workflow.toml` activities list. All
    /// do-not-edit and derived, so a type or activity is declared once
    /// instead of hand-mirrored across files. Types outside the supported
    /// subset fail loudly naming the type and field; see
    /// docs/authoring/codegen.md. With `--check`, regenerates everything in
    /// memory and exits non-zero if any on-disk generated file differs (CI
    /// drift gate).
    Generate {
        /// Workflow project root containing `gleam.toml`, `workflow.toml`,
        /// and `src/<package>_io.gleam`.
        #[arg(default_value = ".")]
        path: PathBuf,
        /// Verify every generated file on disk is current instead of writing;
        /// exits non-zero naming the drifted file (CI gate).
        #[arg(long)]
        check: bool,
    },
    /// Statically check a workflow project for determinism violations.
    ///
    /// Runs entirely locally: reads `workflow.toml` and each declared
    /// workflow's `src/<entry_module>.gleam`, and flags any wall-clock or
    /// entropy call reachable from workflow code — the determinism boundary as
    /// a CI gate. Exits non-zero, naming every offending call and the
    /// `workflow.now()` / `workflow.random()` substitute to use, when any
    /// violation is found.
    Check {
        /// Workflow project root containing `workflow.toml` and `src/`.
        #[arg(default_value = ".")]
        path: PathBuf,
        /// Run the determinism gate. Required: `check` performs the
        /// determinism analysis only.
        #[arg(long)]
        deterministic: bool,
    },
    /// Emit a valid input skeleton derived from a workflow's input type.
    ///
    /// Runs entirely locally: reads the workflow's `input_schema` from
    /// `workflow.toml` — the same JSON Schema the input codec is generated
    /// from — and prints a structurally-valid JSON document so an input is
    /// never hand-written from scratch. Required properties carry type-shaped
    /// placeholders and optional properties are omitted; no values are
    /// defaulted.
    Input {
        /// Workflow type (the workflow's `entry_module`).
        workflow_type: String,
        /// Workflow project root containing `workflow.toml` and `schemas/`.
        #[arg(default_value = ".")]
        path: PathBuf,
    },
    /// Package an already-built Gleam workflow project into `.aion` archives.
    ///
    /// Runs entirely locally: reads `workflow.toml` in the project root and
    /// never connects to a server. Pass `--build` to run `gleam build` first.
    Package {
        /// Workflow project root containing `workflow.toml`.
        #[arg(default_value = ".")]
        path: PathBuf,
        /// Archive output path, resolved against the current directory.
        /// Only valid when the project declares exactly one workflow.
        #[arg(long)]
        out: Option<PathBuf>,
        /// Run `gleam build` in the project before packaging.
        #[arg(long)]
        build: bool,
    },
    /// Deploy a `.aion` archive or direct-compile and deploy an `.awl` file (operator API).
    ///
    /// Requires the server's `[deploy]` surface to be enabled and the caller
    /// to hold the deploy grant (`deploy` token claim, or the
    /// `x-aion-deploy` development header the CLI sends automatically).
    Deploy {
        /// Path to a complete `.aion` archive or an `.awl` workflow document.
        archive: PathBuf,
    },
    /// Compile, deploy, start, and await an AWL workflow in one motion.
    ///
    /// The input is validated locally against the compiled workflow contract
    /// before any deploy connection is made. The global `--namespace` remains
    /// authoritative for the start operation.
    Run {
        /// Path to an `.awl` workflow document.
        file: PathBuf,
        /// JSON input payload, validated before deployment.
        #[arg(long)]
        input: String,
        /// Optional bound on how long this command waits for the run's terminal
        /// event, written as an AWL duration literal (`30s`, `5m`, `3h`, `2d`).
        ///
        /// There is NO default: absent, the wait is bounded only by the
        /// document's own `timeout` declaration, which the engine enforces as a
        /// deadline. A document declaring no timeout and no `--timeout` here
        /// waits indefinitely. Reaching this bound stops the WAIT, never the
        /// run — the workflow continues on the server.
        #[arg(long, value_parser = one_motion::parse_await_timeout)]
        timeout: Option<Duration>,
    },
    /// List every loaded workflow version with its routing flag (operator API).
    Versions {
        /// Show only versions of this workflow type.
        #[arg(long)]
        workflow_type: Option<String>,
    },
    /// Re-point routing for a workflow type at an already-loaded version
    /// (rollback / roll-forward; operator API).
    Route {
        /// Logical workflow type.
        workflow_type: String,
        /// Content hash of the already-loaded target version.
        content_hash: String,
    },
    /// Unload a non-routed, unpinned workflow version (operator API).
    Unload {
        /// Logical workflow type.
        workflow_type: String,
        /// Content hash of the version to unload.
        content_hash: String,
    },
    /// Operate workflows on a running Aion server.
    #[command(flatten)]
    Remote(RemoteCommand),
}

#[derive(Debug, Subcommand)]
enum RemoteCommand {
    /// Start a workflow execution.
    Start {
        /// Workflow type registered with the server.
        workflow_type: String,
        /// JSON input payload for the workflow.
        #[arg(long, default_value = "null")]
        input: String,
        /// Read the JSON input payload from a UTF-8 file instead of `--input`.
        #[arg(long, conflicts_with = "input")]
        input_file: Option<PathBuf>,
        /// Operator-facing display name for this workflow (#211).
        ///
        /// A LABEL shown beside the id, recorded durably on the start. It is
        /// not an address: no command takes a name as a target, so you still
        /// operate on the workflow by its id. Omit to start it unnamed.
        ///
        /// You name the run you are starting, but the label reads back per
        /// WORKFLOW: the recorded attribute carries no run id, so a
        /// continue-as-new successor inherits it and `aion describe` returns
        /// the most recently recorded name for any run of the workflow.
        #[arg(long = "name")]
        display_name: Option<String>,
    },
    /// Send a signal to the latest run of a workflow.
    Signal {
        /// Workflow identifier.
        workflow_id: String,
        /// Signal name.
        signal_name: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
        /// JSON signal payload.
        #[arg(long, default_value = "null")]
        payload: String,
    },
    /// Query the latest run of a workflow.
    Query {
        /// Workflow identifier.
        workflow_id: String,
        /// Query name.
        query_name: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
        /// JSON arguments handed to the workflow's query handler. The default
        /// `null` is the canonical "no arguments" document.
        #[arg(long, default_value = "null")]
        arguments: String,
    },
    /// Request workflow cancellation.
    Cancel {
        /// Workflow identifier.
        workflow_id: String,
        /// Cancellation reason.
        #[arg(long, default_value = "cancelled by aion")]
        reason: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
    },
    /// Retire a workloop: run its declared `retire` body, record the
    /// retirement, and withdraw the loop from the cadence sweep.
    ///
    /// This is a loop's DECLARED way to stop, and it is not failure. `cancel`
    /// records a cancellation and never runs the wind-down the document
    /// declares; there is deliberately no flag to skip that wind-down, because
    /// whether it runs is the document's decision, not the operator's.
    Retire {
        /// Workloop identifier.
        workflow_id: String,
        /// Reason recorded on the retirement, verbatim, in the loop's history.
        #[arg(long, default_value = "retired by aion")]
        reason: String,
    },
    /// Reopen a terminal-reopenable (Failed or Cancelled) run and re-drive it.
    Reopen {
        /// Workflow identifier.
        workflow_id: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
    },
    /// Pause a running workflow, durably holding new activity dispatch.
    Pause {
        /// Workflow identifier.
        workflow_id: String,
        /// Optional pause reason.
        #[arg(long, default_value = "")]
        reason: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
    },
    /// Resume a paused workflow, releasing the dispatch hold.
    Resume {
        /// Workflow identifier.
        workflow_id: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
    },
    /// List workflow executions, newest start first, as one page
    /// `{items, next_cursor, count}`. Without `--limit` every page is walked
    /// and the whole result is printed as one page.
    List {
        /// Optional workflow status filter. Accepted values: running, completed, failed, cancelled, timed-out, continued-as-new, paused.
        #[arg(long, value_parser = parse_status)]
        status: Option<WorkflowStatus>,
        /// Rows per page. Omit to walk every page.
        #[arg(long)]
        limit: Option<u32>,
        /// The `next_cursor` of a previous page under the same filter.
        #[arg(long, requires = "limit")]
        cursor: Option<String>,
    },
    /// Describe a workflow execution: summary facts, run identity, and the
    /// history head sequence. The full event history is `aion inspect`.
    Describe {
        /// Workflow identifier.
        workflow_id: String,
        /// Specific run identifier. Defaults to the latest run when omitted.
        #[arg(long)]
        run_id: Option<String>,
        /// Show raw payload byte arrays instead of decoded payloads.
        #[arg(long)]
        raw: bool,
    },
    /// Tail this run and its direct child legs as one interleaved transcript.
    Tail(tail::TailArgs),
    /// Time-travel over a recorded run's event-store oplog.
    ///
    /// Reads the run's existing history over the same `describe` read every
    /// other command uses — no debug-only log and no second store. For each
    /// recorded event it prints the workflow-visible state projection and the
    /// recorded `now()` for that step; `random()` is surfaced as a run-level
    /// draw-ordinal projection (deterministic, never recorded, with a
    /// workflow-code-dependent per-step count), not a per-step value. On a
    /// non-determinism fault it prints the exact divergent command (expected
    /// vs found at the sequence) the resolver already computed.
    ///
    /// With `--from <seq> --mock <json>` it runs a what-if re-run from the
    /// chosen event with a mocked outcome via the existing replay path and
    /// reports the resulting path. The what-if is entirely in-memory and
    /// read-only; it never writes the event log. `--from` and `--mock` are
    /// required together — a what-if never defaults its fork point or outcome.
    /// `--mock` shape: `{"kind":"activity_completed","result":<json>}`,
    /// `{"kind":"activity_failed","message":"..."}`,
    /// `{"kind":"child_completed","result":<json>}`,
    /// `{"kind":"child_failed","message":"..."}`,
    /// `{"kind":"signal_delivered","payload":<json>}`, or
    /// `{"kind":"timer_fired"}`.
    Inspect {
        /// Workflow identifier.
        workflow_id: String,
        /// Specific run identifier. Defaults to the latest recorded run.
        #[arg(long)]
        run_id: Option<String>,
        /// Fork the what-if re-run from this recorded event sequence.
        #[arg(long)]
        from: Option<u64>,
        /// Mocked outcome JSON for the what-if fork (see the command help).
        #[arg(long)]
        mock: Option<String>,
    },
}

/// Every launch setting that moved out of `aion worker agent` and into the worker
/// document's `harness` section, with the field that replaced it.
///
/// This is a REFUSAL that teaches, never a compat shim: not one of these is accepted,
/// and the process still exits on clap's own argument error. What it adds is the
/// sentence clap cannot know — that the setting still exists, in a different file, under
/// a different name. Without it an operator reads "unexpected argument
/// `--acp-permission`" and has no way to tell a deleted setting from a typo.
const MIGRATED_LAUNCH_FLAGS: [(&str, &str); 14] = [
    ("--harness", "kind"),
    ("--concurrency", "concurrency"),
    ("--reconnect-initial-backoff", "reconnect_initial_backoff"),
    ("--reconnect-max-backoff", "reconnect_max_backoff"),
    ("--reconnect-max-attempts", "reconnect_max_attempts"),
    ("--acp-command", "command"),
    ("--acp-arg", "args"),
    ("--acp-cwd", "cwd"),
    ("--acp-permission", "permission"),
    ("--acp-env", "env_pass"),
    ("--acp-exit-grace", "exit_grace"),
    ("--norn-binary", "binary"),
    ("--norn-arg", "args"),
    ("--norn-unset-env", "env_pass"),
];

/// The subcommand path whose launch flags were deleted.
///
/// The advice is scoped to THIS verb and no other, which is not a nicety: four of the
/// deleted spellings — `--concurrency` and the three `--reconnect-*` — are LIVE, REQUIRED
/// flags on `aion worker awl` (see [`crate::worker_awl::args`], and the start command
/// `docs/workers/AWL-WORKER-DOCUMENTS.md` publishes). An operator who mistypes that
/// documented line must not be told those flags are deleted and sent to write a `harness`
/// section — a worker-family document is FORBIDDEN to carry one, because every action in
/// it has a declared body and there is nothing for an agent to serve. That advice would
/// send them to a refusal, from a refusal, for a flag that was never wrong.
const MIGRATED_FLAG_VERB: [&str; 2] = ["worker", "agent"];

/// Whether `argv` invokes the verb whose launch flags were deleted.
///
/// Matched as the adjacent token pair, which is what a subcommand path looks like on a
/// raw argv: `aion [global flags] worker agent …`. Deliberately not read off clap's
/// rendered usage line — the advice stays a pure function of the argv, so it can be
/// tested without a parser in the loop and cannot change meaning when clap rewords
/// itself.
fn invokes_migrated_flag_verb(argv: &[String]) -> bool {
    argv.windows(MIGRATED_FLAG_VERB.len())
        .any(|pair| pair == MIGRATED_FLAG_VERB)
}

/// The extra sentence a deleted launch flag earns, when `argv` invokes `worker agent`
/// AND carries one.
///
/// Read off the RAW argv rather than off clap's error, because clap reports the first
/// unexpected argument and an operator pasting an old runbook line has typically
/// supplied several — naming one of six would send them round the loop six times.
///
/// `=`-joined forms (`--norn-arg=--account`) are matched on their flag half, because
/// that is the shape market-mirror's own runbook used.
fn migrated_flag_advice(argv: &[String]) -> Option<String> {
    if !invokes_migrated_flag_verb(argv) {
        return None;
    }
    let supplied: Vec<&(&str, &str)> = MIGRATED_LAUNCH_FLAGS
        .iter()
        .filter(|(flag, _)| {
            argv.iter().any(|argument| {
                argument == flag
                    || argument
                        .split_once('=')
                        .is_some_and(|(head, _)| head == *flag)
            })
        })
        .collect();
    if supplied.is_empty() {
        return None;
    }
    let moved = supplied
        .iter()
        .map(|(flag, field)| format!("  {flag}  ->  `{field}` in the document's `harness` section"))
        .collect::<Vec<_>>()
        .join("\n");
    Some(format!(
        "\nThese are not unknown arguments — they are DELETED ones. An agent worker's launch \
         is declared in the `.awl` document it serves, in the `harness` section of its \
         `worker` block, and there is no flag for any of it:\n\n{moved}\n\nIf you typed this \
         command: write the section into the document, then re-run without those flags. \
         `aion awl check <FILE.awl>` names every setting the section is still missing, scoped \
         to the kind the section declares.\n\nIf you did NOT type this command, a managed-worker \
         deployment replayed it. A `builtin` worker-deployment record stores the argv it was \
         created with, verbatim and durably, and the supervisor replays exactly that on every \
         restart — so a record minted before the launch settings moved will fail this way on \
         every restart until its budget is spent, and editing the document will not change it. \
         The record is the thing to fix: add the `harness` section to the document, then \
         re-mint the deployment (`PUT /worker-deployments/<name>`, or the ops console's deploy \
         form) so its stored argv no longer carries these flags."
    ))
}

/// The parsed command line, or the exit code to leave on.
///
/// Wraps [`Cli::try_parse`] rather than calling [`clap::Parser::parse`] for one reason:
/// clap renders and exits by itself, and this binary has a sentence clap cannot know —
/// that some of the arguments it calls "unexpected" are DELETED, and where the settings
/// they carried live now. clap's own rendering and its own exit code are preserved
/// exactly; the advice is written after them, never instead of them.
fn parse_or_refuse() -> Result<Cli, ExitCode> {
    match Cli::try_parse() {
        Ok(cli) => Ok(cli),
        Err(error) => {
            // Collected BEFORE printing, so a closed stderr cannot lose the advice while
            // keeping the error.
            let advice = migrated_flag_advice(&std::env::args().collect::<Vec<_>>());
            // clap writes its own error to the right stream and this is the end of the
            // process either way, so a failed write has nowhere left to be reported.
            drop(error.print());
            if let Some(advice) = advice {
                eprintln!("{advice}");
            }
            Err(ExitCode::from(u8::try_from(error.exit_code()).unwrap_or(2)))
        }
    }
}

#[tokio::main]
async fn main() -> ExitCode {
    let cli = match parse_or_refuse() {
        Ok(cli) => cli,
        Err(code) => return code,
    };
    if let Some(refusal) = launcher_flag_refusal(&cli) {
        eprintln!("{refusal}");
        return ExitCode::from(2);
    }
    match cli.command {
        // Bare `aion`: the launcher. Start the server in the background (or
        // this terminal with `--foreground`), wait until it answers its own
        // liveness probe, report in plain language, open the ops console.
        None => {
            // The launcher starts a server, so the same refusal applies: a
            // dial address is not a bind address (see the Server arm below).
            if let Some(refusal) = server_endpoint_refusal(cli.endpoint.as_deref()) {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            launch::run(cli.foreground, cli.no_open).await
        }
        // The server owns its own reporting contract: tracing-logged errors
        // and the operations exit codes (2 = config, 130 = forced).
        Some(Command::Server(ref command)) => server_command(command, &cli).await,
        // The front-door spellings. Each routes to the SAME run function its
        // `aion server …` long form does, so the two spellings cannot drift
        // apart in behaviour, flags, or output.
        Some(Command::Stop(ref args)) => {
            if let Some(refusal) =
                control_verb_endpoint_refusal("stop", "stop", cli.endpoint.as_deref())
            {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            server_stop::run(args)
        }
        Some(Command::Update(ref args)) => {
            if let Some(refusal) =
                control_verb_endpoint_refusal("update", "update", cli.endpoint.as_deref())
            {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            update::run(args).await
        }
        Some(Command::Store(ref command)) => match command.command {
            store_harden::StoreSubcommand::Harden(ref args) => store_harden::harden(args),
        },
        Some(Command::Dev(ref args)) => {
            // The dev loop runs locally and drives the running server over the
            // same operator deploy RPC `aion deploy` uses; it reports through
            // the one client rendering contract.
            let result = dev::run(args, deploy_target(&cli))
                .await
                .and_then(|value| print_json(&value, cli.pretty));
            match result {
                Ok(()) => ExitCode::SUCCESS,
                Err(error) => {
                    eprintln!("{}", render::render_error(&error));
                    ExitCode::FAILURE
                }
            }
        }
        // The AWL authoring commands own their own compiler-style reporting
        // contract: diagnostics to stderr, a one-line summary to stdout.
        Some(Command::Awl { ref command }) => awl::run(command),
        Some(Command::Worker { ref command }) => {
            let result = match command {
                WorkerCommand::Shell(args) => worker_shell::run(args, &cli.endpoint()).await,
                // The AWL worker dials the global gRPC endpoint, exactly as
                // the shell worker does, and normalizes a scheme-less one.
                WorkerCommand::Awl(args) => worker_awl::run(args, &cli.endpoint()).await,
                // The agent worker registers into the namespace the global flag
                // names, and dials liminal candidates rather than `--endpoint`.
                WorkerCommand::Agent(args) => worker_agent::run(args, &cli.namespace).await,
                // The managed-fleet verbs are operator commands against a
                // RUNNING server, not worker processes this invocation hosts.
                // They print the server's own report as JSON, through the same
                // printer every other read uses.
                WorkerCommand::Status => worker_managed::status(&deploy_target(&cli))
                    .await
                    .and_then(|value| print_json(&value, cli.pretty)),
                WorkerCommand::Start { name } => worker_managed::start(&deploy_target(&cli), name)
                    .await
                    .and_then(|value| print_json(&value, cli.pretty)),
                WorkerCommand::Stop { name } => worker_managed::stop(&deploy_target(&cli), name)
                    .await
                    .and_then(|value| print_json(&value, cli.pretty)),
                WorkerCommand::Restart { name } => {
                    worker_managed::restart(&deploy_target(&cli), name)
                        .await
                        .and_then(|value| print_json(&value, cli.pretty))
                }
            };
            match result {
                Ok(()) => ExitCode::SUCCESS,
                Err(error) => {
                    eprintln!("{}", render::render_error(&error));
                    ExitCode::FAILURE
                }
            }
        }
        Some(Command::Client(ref command)) => run_client_command(&cli, command).await,
    }
}

async fn run_client_command(cli: &Cli, command: &ClientCommand) -> ExitCode {
    let result = run(cli, command).await.and_then(|value| {
        // `tail` owns a human transcript reporting contract, just as AWL owns
        // its compiler-style contract. Do not append JSON after the transcript.
        if matches!(command, ClientCommand::Remote(RemoteCommand::Tail(_))) {
            Ok(())
        } else {
            print_json(&value, cli.pretty)
        }
    });
    match result {
        Ok(()) => ExitCode::SUCCESS,
        Err(error) => {
            report_client_error(&error);
            ExitCode::FAILURE
        }
    }
}

/// Dispatches the parsed client command, connecting to the server only for
/// remote commands. `package` is local-only and never constructs a client;
/// the deploy subcommands drive the operator `DeployService` directly and
/// never touch the caller SDK.
async fn run(cli: &Cli, command: &ClientCommand) -> Result<Value> {
    match command {
        ClientCommand::New(args) => new::run(args),
        ClientCommand::Generate { path, check } => generate::run(path, *check),
        ClientCommand::Check {
            path,
            deterministic,
        } => check_deterministic::run(path, *deterministic),
        ClientCommand::Input {
            workflow_type,
            path,
        } => input::run(path, workflow_type),
        ClientCommand::Package { path, out, build } => package::run(path, out.as_deref(), *build),
        ClientCommand::Deploy { archive } => {
            if one_motion::is_awl(archive) {
                let (_, _, bytes) = one_motion::package_file(archive)?;
                deploy::deploy_bytes(&deploy_target(cli), bytes).await
            } else {
                deploy::deploy(&deploy_target(cli), archive).await
            }
        }
        ClientCommand::Run {
            file,
            input,
            timeout,
        } => {
            let prepared = one_motion::prepare_run(file, input)?;
            deploy::deploy_bytes(&deploy_target(cli), prepared.archive).await?;
            let client = build_client(cli).await?;
            let handle =
                start_workflow_handle(&client, &prepared.workflow_name, prepared.input).await?;
            one_motion::await_result(handle, &prepared.workflow_name, *timeout).await
        }
        ClientCommand::Versions { workflow_type } => {
            deploy::versions(&deploy_target(cli), workflow_type.as_deref()).await
        }
        ClientCommand::Route {
            workflow_type,
            content_hash,
        } => deploy::route(&deploy_target(cli), workflow_type, content_hash).await,
        ClientCommand::Unload {
            workflow_type,
            content_hash,
        } => deploy::unload(&deploy_target(cli), workflow_type, content_hash).await,
        ClientCommand::Remote(command) => {
            let client = build_client(cli).await?;
            execute(&client, command, cli).await
        }
    }
}

fn deploy_target(cli: &Cli) -> deploy::DeployTarget {
    deploy::DeployTarget::new(
        cli.endpoint(),
        cli.subject.clone(),
        deploy::resolve_token(cli.token.clone()),
    )
}

fn report_client_error(error: &anyhow::Error) {
    if let Some(diagnostic) = error.downcast_ref::<aion_awl::CompileError>() {
        eprintln!("{diagnostic}");
    } else if let Some(refusal) = error.downcast_ref::<one_motion::InputValidationError>() {
        eprintln!("{refusal}");
    } else {
        eprintln!("{}", render::render_error(error));
    }
}

async fn build_client(cli: &Cli) -> Result<Client> {
    ClientBuilder::new(cli.endpoint())
        .with_namespace(cli.namespace.clone())
        .with_subject(cli.subject.clone())
        .with_authorized_namespaces([cli.namespace.clone()])
        .build()
        .await
        .context("failed to connect to Aion server")
}

async fn execute(client: &Client, command: &RemoteCommand, cli: &Cli) -> Result<Value> {
    match command {
        RemoteCommand::Start {
            workflow_type,
            input,
            input_file,
            display_name,
        } => {
            let input = resolve_start_input(input, input_file.as_deref())?;
            start_workflow(client, workflow_type, input, display_name.clone()).await
        }
        RemoteCommand::Signal {
            workflow_id,
            signal_name,
            run_id,
            payload,
        } => signal_workflow(client, workflow_id, signal_name, run_id.as_deref(), payload).await,
        RemoteCommand::Query {
            workflow_id,
            query_name,
            run_id,
            arguments,
        } => {
            query_workflow(
                client,
                workflow_id,
                query_name,
                run_id.as_deref(),
                arguments,
            )
            .await
        }
        RemoteCommand::Cancel {
            workflow_id,
            reason,
            run_id,
        } => cancel_workflow(client, workflow_id, reason, run_id.as_deref()).await,
        RemoteCommand::Retire {
            workflow_id,
            reason,
        } => retire_workloop(client, workflow_id, reason).await,
        RemoteCommand::Reopen {
            workflow_id,
            run_id,
        } => reopen_workflow(client, workflow_id, run_id.as_deref()).await,
        RemoteCommand::Pause {
            workflow_id,
            reason,
            run_id,
        } => pause_workflow(client, workflow_id, reason, run_id.as_deref()).await,
        RemoteCommand::Resume {
            workflow_id,
            run_id,
        } => resume_workflow(client, workflow_id, run_id.as_deref()).await,
        RemoteCommand::List {
            status,
            limit,
            cursor,
        } => list_workflows(client, *status, *limit, cursor.clone()).await,
        RemoteCommand::Describe {
            workflow_id,
            run_id,
            raw,
        } => describe_workflow(client, workflow_id, run_id.as_deref(), *raw).await,
        RemoteCommand::Tail(args) => tail::dispatch(client, args, cli).await,
        RemoteCommand::Inspect {
            workflow_id,
            run_id,
            from,
            mock,
        } => {
            inspect::run(
                client,
                workflow_id,
                run_id.as_deref(),
                *from,
                mock.as_deref(),
            )
            .await
        }
    }
}

/// Resolve the start input into a JSON payload: the contents of `--input-file`
/// when given (read as UTF-8), otherwise the inline `--input` string. The two
/// flags are mutually exclusive at parse time (`conflicts_with`), so a file is
/// only ever read when `--input` was not passed. Validating here lets a
/// malformed-JSON error name the flag the payload actually came from.
fn resolve_start_input(input: &str, input_file: Option<&Path>) -> Result<Payload> {
    match input_file {
        Some(path) => {
            let raw = std::fs::read_to_string(path)
                .with_context(|| format!("failed to read --input-file {}", path.display()))?;
            json_payload(&raw)
                .with_context(|| format!("invalid --input-file {} JSON", path.display()))
        }
        None => json_payload(input).context("invalid --input JSON"),
    }
}

async fn signal_workflow(
    client: &Client,
    workflow_id: &str,
    signal_name: &str,
    run_id: Option<&str>,
    payload: &str,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let payload = json_payload(payload).context("invalid --payload JSON")?;
    client
        .signal(
            &workflow_id,
            run_id.as_ref(),
            signal_name.to_owned(),
            payload,
        )
        .await
        .context("failed to signal workflow")?;
    to_value(AcknowledgementOutput {
        workflow_id: &workflow_id.to_string(),
        accepted: true,
    })
}

async fn query_workflow(
    client: &Client,
    workflow_id: &str,
    query_name: &str,
    run_id: Option<&str>,
    arguments: &str,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let arguments = json_payload(arguments).context("invalid query arguments")?;
    let payload = client
        .query(
            &workflow_id,
            run_id.as_ref(),
            query_name.to_owned(),
            arguments,
            QUERY_DEADLINE,
        )
        .await
        .context("failed to query workflow")?;
    to_value(QueryOutput {
        result: payload_to_json(&payload)?,
    })
}

async fn cancel_workflow(
    client: &Client,
    workflow_id: &str,
    reason: &str,
    run_id: Option<&str>,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    client
        .cancel(&workflow_id, run_id.as_ref(), reason.to_owned())
        .await
        .context("failed to cancel workflow")?;
    to_value(AcknowledgementOutput {
        workflow_id: &workflow_id.to_string(),
        accepted: true,
    })
}

async fn retire_workloop(client: &Client, workflow_id: &str, reason: &str) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let recorded = client
        .retire_workloop(&workflow_id, reason.to_owned())
        .await
        .context("failed to retire workloop")?;
    to_value(RetireOutput {
        workflow_id: &workflow_id.to_string(),
        retired: true,
        reason: &recorded,
    })
}

async fn reopen_workflow(
    client: &Client,
    workflow_id: &str,
    run_id: Option<&str>,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let outcome = client
        .reopen(&workflow_id, run_id.as_ref())
        .await
        .map_err(|error| reopen_error(&workflow_id.to_string(), error))?;
    to_value(reopen_output(&workflow_id.to_string(), &outcome))
}

/// Render a reopen failure as a typed, operator-facing message — never a raw
/// transport error. `InvalidState` names the precondition ("not reopenable"),
/// `NotFound` is a plain "not found"; everything else keeps the client detail.
fn reopen_error(workflow_id: &str, error: ClientError) -> anyhow::Error {
    match error {
        ClientError::InvalidState { detail } => {
            anyhow::anyhow!(
                "workflow {workflow_id} is not reopenable: {}",
                detail.message
            )
        }
        ClientError::NotFound { .. } => anyhow::anyhow!("workflow {workflow_id} not found"),
        other => anyhow::anyhow!("failed to reopen workflow {workflow_id}: {other}"),
    }
}

async fn pause_workflow(
    client: &Client,
    workflow_id: &str,
    reason: &str,
    run_id: Option<&str>,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let outcome = client
        .pause(&workflow_id, run_id.as_ref(), reason.to_owned())
        .await
        .map_err(|error| pause_error(&workflow_id.to_string(), error))?;
    to_value(pause_output(&workflow_id.to_string(), &outcome))
}

async fn resume_workflow(
    client: &Client,
    workflow_id: &str,
    run_id: Option<&str>,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let outcome = client
        .resume(&workflow_id, run_id.as_ref())
        .await
        .map_err(|error| resume_error(&workflow_id.to_string(), error))?;
    to_value(resume_output(&workflow_id.to_string(), &outcome))
}

/// Render a pause failure as a typed, operator-facing message (#204).
/// `InvalidState` names the precondition (the run's actual status); `NotFound`
/// is a plain "not found"; everything else keeps the client detail.
fn pause_error(workflow_id: &str, error: ClientError) -> anyhow::Error {
    match error {
        ClientError::InvalidState { detail } => {
            anyhow::anyhow!(
                "workflow {workflow_id} cannot be paused: {}",
                detail.message
            )
        }
        ClientError::NotFound { .. } => anyhow::anyhow!("workflow {workflow_id} not found"),
        other => anyhow::anyhow!("failed to pause workflow {workflow_id}: {other}"),
    }
}

/// Render a resume failure as a typed, operator-facing message (#204).
fn resume_error(workflow_id: &str, error: ClientError) -> anyhow::Error {
    match error {
        ClientError::InvalidState { detail } => {
            anyhow::anyhow!(
                "workflow {workflow_id} cannot be resumed: {}",
                detail.message
            )
        }
        ClientError::NotFound { .. } => anyhow::anyhow!("workflow {workflow_id} not found"),
        other => anyhow::anyhow!("failed to resume workflow {workflow_id}: {other}"),
    }
}

/// The transfer chunk a full walk of the list is fetched in. Not a cap on
/// the result — the walk continues until the server hands back no cursor —
/// only the size of each round trip.
const LIST_WALK_CHUNK: u32 = 500;

async fn list_workflows(
    client: &Client,
    status: Option<WorkflowStatus>,
    limit: Option<u32>,
    cursor: Option<String>,
) -> Result<Value> {
    let request = |cursor: Option<String>, limit: u32| ListRequest {
        filter: aion_core::WorkflowListFilter {
            statuses: status.into_iter().collect(),
            ..aion_core::WorkflowListFilter::default()
        },
        sort: aion_core::WorkflowSort {
            field: aion_core::WorkflowSortField::StartedAt,
            direction: aion_core::SortDirection::Desc,
        },
        cursor,
        limit,
        request_id: None,
    };
    if let Some(limit) = limit {
        let page = client
            .list(request(cursor, limit))
            .await
            .context("failed to list workflows")?;
        return to_value(page);
    }
    let mut walked = client
        .list(request(None, LIST_WALK_CHUNK))
        .await
        .context("failed to list workflows")?;
    while let Some(next) = walked.next_cursor.take() {
        let page = client
            .list(request(Some(next), LIST_WALK_CHUNK))
            .await
            .context("failed to list workflows")?;
        walked.items.extend(page.items);
        walked.next_cursor = page.next_cursor;
        walked.count = page.count;
        walked.provenance = page.provenance;
    }
    to_value(walked)
}

async fn describe_workflow(
    client: &Client,
    workflow_id: &str,
    run_id: Option<&str>,
    raw: bool,
) -> Result<Value> {
    let workflow_id = parse_workflow_id(workflow_id)?;
    let run_id = run_id.map(parse_run_id).transpose()?;
    let description = client
        .describe(&workflow_id, run_id.as_ref())
        .await
        .context("failed to describe workflow")?;
    describe_output(
        description.summary,
        description.run_id,
        description.history_head_seq,
        description.terminal_event,
        description.provenance,
        raw,
    )
}

/// The `aion server` command family: the control verbs (stop/status, local
/// by contract) and the run-the-server default. Split from `main` so the
/// dispatch stays a table; every refusal doctrine ruling lives here with the
/// arm it governs.
async fn server_command(command: &server::ServerCommand, cli: &Cli) -> ExitCode {
    match &command.command {
        // The control verbs act on a server already running under this
        // home; they never start one and never read the run flags
        // (`args_conflicts_with_subcommands` guarantees none were given).
        // They are also LOCAL: each reads this machine's AION_HOME pid
        // record and signals or probes that pid, so a named `--endpoint`
        // cannot be honoured — and silently dropping it would land the
        // action on this box while the operator believes it went to the
        // named one. Refuse, same doctrine as the run arm below.
        Some(server::ServerSubcommand::Stop(args)) => {
            if let Some(refusal) =
                control_verb_endpoint_refusal("stop", "stop", cli.endpoint.as_deref())
            {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            server_stop::run(args)
        }
        Some(server::ServerSubcommand::Status(args)) => {
            if let Some(refusal) =
                control_verb_endpoint_refusal("status", "report status", cli.endpoint.as_deref())
            {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            server_status::run(args).await
        }
        Some(server::ServerSubcommand::Restart(args)) => {
            if let Some(refusal) =
                control_verb_endpoint_refusal("restart", "restart", cli.endpoint.as_deref())
            {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            server_restart::run(args).await
        }
        None => {
            let args = &command.run;
            // `--endpoint` is a client dial address. It was accepted here and
            // silently dropped, so an operator who passed it to move the bind
            // got a server on the default port instead — which, on a box
            // already running one, means a collision or a second process
            // reaching for state the operator thought they had moved away
            // from. Refuse: a bind address the operator named and did not get
            // is not a thing to start up and hope about.
            if let Some(refusal) = server_endpoint_refusal(cli.endpoint.as_deref()) {
                eprintln!("{refusal}");
                return ExitCode::from(2);
            }
            // Report the agent harness composed into this binary at the root
            // (§3A.3): the default build ships Norn out-of-box; a
            // `--no-default-features` build reports none. The worker drives
            // whichever harness is compiled in through the neutral trait.
            harness::announce_composed_harness();
            // `--open` is a CLI-only nicety: resolve the served URL from the
            // effective config and, once the listener is reachable, open it in
            // the browser. Best-effort — a failure to open never blocks the run.
            if args.open() {
                server::spawn_browser_open(&args.clone().into());
            }
            aion_server::run(args.clone().into()).await
        }
    }
}

/// The refusal for `--endpoint` on a control verb, `None` when no endpoint
/// was named.
///
/// `aion server stop`/`status` are local actions: they read this machine's
/// `AION_HOME` pid record and signal or probe that process. An endpoint names
/// a server a CLIENT dials, which these verbs can never honour — and a
/// destructive verb that silently ignores where the operator pointed it lands
/// on the wrong machine. Same doctrine as [`server_endpoint_refusal`]: refuse
/// a flag the verb cannot honour, with wording that says the verb is local
/// rather than that it cannot bind.
///
/// `verb` is the subcommand token (it renders inside `` `aion server {verb}` ``);
/// `action` is the rendered verb PHRASE ("stop", "report status") — the token
/// is not always an English verb, and "refusing to status" is not a sentence
/// an operator should meet at 3am.
fn control_verb_endpoint_refusal(
    verb: &str,
    action: &str,
    endpoint: Option<&str>,
) -> Option<String> {
    endpoint.map(|endpoint| {
        format!(
            "refusing to {action}: `--endpoint {endpoint}` names a server a client dials, and \
             `aion server {verb}` is local — it reads this machine's AION_HOME pid record and \
             acts on that process. To operate a remote server, run the verb on that machine \
             (its home decides which server answers)."
        )
    })
}

/// The refusal `aion server` owes an operator who passed `--endpoint`, or
/// `None` when they did not.
///
/// Separate from the dispatch arm so the message itself is testable: a refusal
/// that names the wrong replacement flag is barely better than the silence it
/// replaced.
fn server_endpoint_refusal(endpoint: Option<&str>) -> Option<String> {
    endpoint.map(|endpoint| {
        format!(
            "refusing to start: `--endpoint {endpoint}` names an address a CLIENT dials, and \
             `aion server` cannot bind it. Use `--grpc-address <ADDR>` (or \
             AION_SERVER_GRPC_ADDRESS, or [server].grpc_address) to choose the gRPC listener, \
             and `--listen-address <ADDR>` for HTTP."
        )
    })
}

/// The refusal for launcher-only flags combined with a subcommand, `None`
/// when the invocation is coherent.
///
/// `--foreground` and `--no-open` configure the bare `aion` launcher; every
/// subcommand already has its own answer to both questions (`aion server` IS
/// foreground, and `aion server --open` opts INTO the browser), so accepting
/// them here would mean silently ignoring what the operator asked for.
fn launcher_flag_refusal(cli: &Cli) -> Option<String> {
    if cli.command.is_none() || (!cli.foreground && !cli.no_open) {
        return None;
    }
    let named = match (cli.foreground, cli.no_open) {
        (true, true) => "`--foreground` and `--no-open` configure",
        (true, false) => "`--foreground` configures",
        // The guard above established at least one flag is set.
        (false, _) => "`--no-open` configures",
    };
    Some(format!(
        "refusing to start: {named} the bare `aion` launcher and cannot be combined with a \
         subcommand. `aion server` already runs in the foreground, and `aion server --open` \
         opts into the browser."
    ))
}

impl Cli {
    /// The address a client dials: what the operator named, or the default, and
    /// **always carrying a scheme**.
    ///
    /// The default lives here rather than in the clap attribute so that
    /// [`Self::endpoint`] being `None` still means "nobody said", which is the
    /// only way `aion server` can tell a flag it must refuse from one that was
    /// never passed.
    ///
    /// Normalisation lives here too, and that is the load-bearing part. It used
    /// to live at each dial site: every client verb wrapped this accessor in
    /// [`normalize_endpoint`] and `aion worker shell` did not. Since
    /// [`DEFAULT_ENDPOINT`] is scheme-less, a shell worker started without an
    /// explicit `--endpoint http://…` never connected to anything. Adding the
    /// missing wrapper at that one site would have left the shape that produced
    /// it — one rule, "a dial address carries a scheme", known in as many places
    /// as there are callers and agreed only by whoever remembered. There is now
    /// no un-normalised dial address for a call site to reach for.
    ///
    /// [`server_endpoint_refusal`] still reads the raw `Option` field directly,
    /// deliberately and for the opposite reason: it needs "nobody said" and the
    /// operator's literal text, neither of which survives normalisation. It
    /// refuses; it never dials.
    fn endpoint(&self) -> String {
        normalize_endpoint(self.endpoint.as_deref().unwrap_or(DEFAULT_ENDPOINT))
    }
}

fn normalize_endpoint(endpoint: &str) -> String {
    if endpoint.contains("://") {
        endpoint.to_owned()
    } else {
        format!("http://{endpoint}")
    }
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};

    use aion_core::Payload;
    use aion_server::config::CliOverrides;
    use clap::{CommandFactory, Parser};
    use serde_json::{Value, json};

    use crate::{payload::json_payload, tail};

    use super::{
        Cli, ClientCommand, Command, DEFAULT_ENDPOINT, RemoteCommand, awl,
        control_verb_endpoint_refusal, normalize_endpoint, payload_to_json, resolve_start_input,
        server_endpoint_refusal,
    };

    const WORKFLOW_ID: &str = "00000000-0000-0000-0000-000000000001";
    const RUN_ID: &str = "00000000-0000-0000-0000-000000000002";

    #[test]
    fn start_reads_input_payload_from_file() -> anyhow::Result<()> {
        let dir = tempfile::tempdir()?;
        let path = dir.path().join("payload.json");
        std::fs::write(&path, r#"{"order_id":42}"#)?;

        let cli = Cli::try_parse_from([
            std::ffi::OsStr::new("aion"),
            std::ffi::OsStr::new("start"),
            std::ffi::OsStr::new("order_saga"),
            std::ffi::OsStr::new("--input-file"),
            path.as_os_str(),
        ])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Start {
            input,
            input_file,
            ..
        }))) = cli.command
        else {
            anyhow::bail!("expected start command");
        };
        // The inline default is untouched; the file contents win.
        assert_eq!(input, "null");
        assert_eq!(input_file.as_deref(), Some(path.as_path()));
        let payload = resolve_start_input(&input, input_file.as_deref())?;
        assert_eq!(payload_to_json(&payload)?, json!({ "order_id": 42 }));
        Ok(())
    }

    #[test]
    fn start_without_input_file_keeps_inline_input() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "start", "order_saga"])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Start {
            input,
            input_file,
            ..
        }))) = cli.command
        else {
            anyhow::bail!("expected start command");
        };
        assert!(input_file.is_none());
        let payload = resolve_start_input(&input, input_file.as_deref())?;
        assert_eq!(payload_to_json(&payload)?, Value::Null);
        Ok(())
    }

    /// #211: `--name` sets the run's operator-facing label, and omitting it
    /// leaves the run unnamed. The flag is spelled `--name` while the field is
    /// `display_name`, so this pins the spelling an operator actually types.
    #[test]
    fn start_reads_the_display_name_flag() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "start",
            "order_saga",
            "--name",
            "Nightly settlement",
        ])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Start {
            display_name, ..
        }))) = cli.command
        else {
            anyhow::bail!("expected start command");
        };
        assert_eq!(display_name.as_deref(), Some("Nightly settlement"));

        let cli = Cli::try_parse_from(["aion", "start", "order_saga"])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Start {
            display_name, ..
        }))) = cli.command
        else {
            anyhow::bail!("expected start command");
        };
        assert_eq!(display_name, None, "an unnamed start selects no label");
        // A whitespace-only `--name` is refused, but the refusal lives in the
        // client SDK's start-option validation (`aion-client`'s
        // `a_blank_display_name_is_refused`), which every CLI start goes
        // through — clap itself has no opinion on a blank string, and a test
        // here that only re-asserted the string would measure nothing.
        Ok(())
    }

    #[test]
    fn start_rejects_input_and_input_file_together() {
        assert!(
            Cli::try_parse_from([
                "aion",
                "start",
                "order_saga",
                "--input",
                "{}",
                "--input-file",
                "payload.json",
            ])
            .is_err()
        );
    }

    #[test]
    fn start_missing_input_file_error_names_the_path() -> anyhow::Result<()> {
        let path = Path::new("/definitely/not/there/payload.json");
        let Err(error) = resolve_start_input("null", Some(path)) else {
            anyhow::bail!("missing file must error");
        };
        assert!(
            format!("{error:#}").contains("/definitely/not/there/payload.json"),
            "error should name the path: {error:#}"
        );
        Ok(())
    }

    #[test]
    fn start_malformed_input_file_json_error_names_the_flag_and_path() -> anyhow::Result<()> {
        let dir = tempfile::tempdir()?;
        let path = dir.path().join("payload.json");
        std::fs::write(&path, "{not json")?;

        let Err(error) = resolve_start_input("null", Some(&path)) else {
            anyhow::bail!("malformed file JSON must error");
        };
        let message = format!("{error:#}");
        assert!(
            message.contains(&format!("invalid --input-file {} JSON", path.display())),
            "error should name --input-file and the path: {message}"
        );
        assert!(
            !message.contains("invalid --input JSON"),
            "error must not blame the inline flag: {message}"
        );
        Ok(())
    }

    #[test]
    fn start_malformed_inline_input_json_error_names_the_flag() -> anyhow::Result<()> {
        let Err(error) = resolve_start_input("{not json", None) else {
            anyhow::bail!("malformed inline JSON must error");
        };
        let message = format!("{error:#}");
        assert!(
            message.contains("invalid --input JSON"),
            "error should name --input: {message}"
        );
        Ok(())
    }

    #[test]
    fn describe_accepts_optional_run_id() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "describe", WORKFLOW_ID, "--run-id", RUN_ID])?;

        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Describe {
            workflow_id,
            run_id,
            raw,
        }))) = cli.command
        else {
            anyhow::bail!("expected describe command");
        };
        assert_eq!(workflow_id, WORKFLOW_ID);
        assert_eq!(run_id.as_deref(), Some(RUN_ID));
        assert!(!raw);
        Ok(())
    }

    #[test]
    fn tail_accepts_parent_id_without_any_child_id() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "tail", WORKFLOW_ID, "--no-follow"])?;

        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Tail(args)))) = cli.command
        else {
            anyhow::bail!("expected tail command");
        };
        assert_eq!(args.workflow_id, WORKFLOW_ID);
        assert_eq!(args.run_id, None);
        assert_eq!(args.http_endpoint, tail::default_http_endpoint());
        assert!(args.no_follow);
        Ok(())
    }

    #[test]
    fn describe_accepts_raw_payload_flag() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "describe", WORKFLOW_ID, "--raw"])?;

        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Describe { raw, .. }))) =
            cli.command
        else {
            anyhow::bail!("expected describe command");
        };
        assert!(raw);
        Ok(())
    }

    #[test]
    fn reopen_parses_workflow_id_and_optional_run_id() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "reopen", WORKFLOW_ID])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Reopen {
            workflow_id,
            run_id,
        }))) = cli.command
        else {
            anyhow::bail!("expected reopen command");
        };
        assert_eq!(workflow_id, WORKFLOW_ID);
        assert!(run_id.is_none());

        let cli = Cli::try_parse_from(["aion", "reopen", WORKFLOW_ID, "--run-id", RUN_ID])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Reopen { run_id, .. }))) =
            cli.command
        else {
            anyhow::bail!("expected reopen command");
        };
        assert_eq!(run_id.as_deref(), Some(RUN_ID));
        Ok(())
    }

    #[test]
    fn reopen_help_documents_the_subcommand() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(reopen) = command.find_subcommand_mut("reopen") else {
            anyhow::bail!("reopen subcommand should be registered");
        };
        let help = reopen.render_long_help().to_string();
        assert!(help.contains("--run-id"));
        assert!(help.contains("reopen"));
        Ok(())
    }

    #[test]
    fn new_parses_name_template_and_worker() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "new",
            "my_flow",
            "--template",
            "saga",
            "--worker",
            "rust",
        ])?;
        let Some(Command::Client(ClientCommand::New(_))) = cli.command else {
            anyhow::bail!("expected new command");
        };

        // The template defaults to awl and --worker is optional. The parsed
        // default itself is pinned in `new::scaffold`'s tests, where the
        // argument fields are visible.
        let cli = Cli::try_parse_from(["aion", "new", "my_flow"])?;
        let Some(Command::Client(ClientCommand::New(_))) = cli.command else {
            anyhow::bail!("expected new command");
        };

        // Unknown templates and worker languages are parse errors.
        assert!(Cli::try_parse_from(["aion", "new", "x", "--template", "nope"]).is_err());
        assert!(Cli::try_parse_from(["aion", "new", "x", "--worker", "cobol"]).is_err());
        Ok(())
    }

    #[test]
    fn new_help_documents_templates_and_refusal() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(new) = command.find_subcommand_mut("new") else {
            anyhow::bail!("new subcommand should be registered");
        };
        let help = new.render_long_help().to_string();

        assert!(help.contains("--template"));
        assert!(help.contains("awl"));
        assert!(
            help.contains("[default: awl]"),
            "help must render the awl default: {help}"
        );
        assert!(help.contains("hello-world"));
        assert!(help.contains("approval-flow"));
        assert!(help.contains("saga"));
        assert!(help.contains("--worker"));
        assert!(help.contains("non-empty directory"));
        Ok(())
    }

    #[test]
    fn package_defaults_to_current_directory_without_build_or_out() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "package"])?;

        let Some(Command::Client(ClientCommand::Package { path, out, build })) = cli.command else {
            anyhow::bail!("expected package command");
        };
        assert_eq!(path, Path::new("."));
        assert!(out.is_none());
        assert!(!build);
        Ok(())
    }

    #[test]
    fn package_accepts_path_out_and_build() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "package",
            "examples/hello-world",
            "--out",
            "dist/hello.aion",
            "--build",
        ])?;

        let Some(Command::Client(ClientCommand::Package { path, out, build })) = cli.command else {
            anyhow::bail!("expected package command");
        };
        assert_eq!(path, Path::new("examples/hello-world"));
        assert_eq!(out.as_deref(), Some(Path::new("dist/hello.aion")));
        assert!(build);
        Ok(())
    }

    #[test]
    fn package_help_documents_local_only_behaviour() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(package) = command.find_subcommand_mut("package") else {
            anyhow::bail!("package subcommand should be registered");
        };
        let help = package.render_long_help().to_string();

        assert!(help.contains("--build"));
        assert!(help.contains("--out"));
        assert!(help.contains("never connects to a server"));
        Ok(())
    }

    #[test]
    fn check_parses_path_and_requires_the_deterministic_flag() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "check", "examples/order-saga", "--deterministic"])?;
        let Some(Command::Client(ClientCommand::Check {
            path,
            deterministic,
        })) = cli.command
        else {
            anyhow::bail!("expected check command");
        };
        assert_eq!(path, Path::new("examples/order-saga"));
        assert!(deterministic);

        // Path defaults to the current directory; the flag defaults off.
        let cli = Cli::try_parse_from(["aion", "check"])?;
        let Some(Command::Client(ClientCommand::Check {
            path,
            deterministic,
        })) = cli.command
        else {
            anyhow::bail!("expected check command");
        };
        assert_eq!(path, Path::new("."));
        assert!(!deterministic);
        Ok(())
    }

    #[test]
    fn awl_check_and_fmt_parse_the_file_path() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "awl", "check", "flows/report.awl"])?;
        let Some(Command::Awl {
            command: awl::AwlCommand::Check { file },
        }) = cli.command
        else {
            anyhow::bail!("expected awl check command");
        };
        assert_eq!(file, Path::new("flows/report.awl"));

        let cli = Cli::try_parse_from(["aion", "awl", "fmt", "flows/report.awl"])?;
        let Some(Command::Awl {
            command: awl::AwlCommand::Fmt { file },
        }) = cli.command
        else {
            anyhow::bail!("expected awl fmt command");
        };
        assert_eq!(file, Path::new("flows/report.awl"));
        Ok(())
    }

    #[test]
    fn awl_commands_require_a_file_and_fmt_has_no_check_mode() {
        // The file argument is mandatory for both commands.
        assert!(Cli::try_parse_from(["aion", "awl", "check"]).is_err());
        assert!(Cli::try_parse_from(["aion", "awl", "fmt"]).is_err());
        // House rule: the printer is the formatter — no `--check` mode.
        assert!(Cli::try_parse_from(["aion", "awl", "fmt", "f.awl", "--check"]).is_err());
    }

    #[test]
    fn awl_emit_parses_the_file_path_and_optional_output() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "awl", "emit", "flows/report.awl"])?;
        let Some(Command::Awl {
            command:
                awl::AwlCommand::Emit {
                    file,
                    output,
                    target,
                },
        }) = cli.command
        else {
            anyhow::bail!("expected awl emit command");
        };
        assert_eq!(file, Path::new("flows/report.awl"));
        assert_eq!(output, None);
        // The default target is Gleam (byte-identical to the pre-`--target`
        // behaviour).
        assert_eq!(target, awl::EmitTarget::Gleam);

        let cli = Cli::try_parse_from([
            "aion",
            "awl",
            "emit",
            "flows/report.awl",
            "-o",
            "flows/report.beam",
            "--target",
            "beam",
        ])?;
        let Some(Command::Awl {
            command:
                awl::AwlCommand::Emit {
                    file,
                    output,
                    target,
                },
        }) = cli.command
        else {
            anyhow::bail!("expected awl emit command");
        };
        assert_eq!(file, Path::new("flows/report.awl"));
        assert_eq!(output, Some(PathBuf::from("flows/report.beam")));
        assert_eq!(target, awl::EmitTarget::Beam);
        Ok(())
    }

    #[test]
    fn awl_emit_requires_a_file() {
        assert!(Cli::try_parse_from(["aion", "awl", "emit"]).is_err());
    }

    #[test]
    fn input_parses_workflow_type_and_path() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "input", "order_saga", "examples/order-saga"])?;
        let Some(Command::Client(ClientCommand::Input {
            workflow_type,
            path,
        })) = cli.command
        else {
            anyhow::bail!("expected input command");
        };
        assert_eq!(workflow_type, "order_saga");
        assert_eq!(path, Path::new("examples/order-saga"));

        // The path defaults to the current directory.
        let cli = Cli::try_parse_from(["aion", "input", "order_saga"])?;
        let Some(Command::Client(ClientCommand::Input { path, .. })) = cli.command else {
            anyhow::bail!("expected input command");
        };
        assert_eq!(path, Path::new("."));
        Ok(())
    }

    #[test]
    fn check_help_documents_the_determinism_gate() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(check) = command.find_subcommand_mut("check") else {
            anyhow::bail!("check subcommand should be registered");
        };
        let help = check.render_long_help().to_string();
        assert!(help.contains("--deterministic"));
        assert!(help.contains("wall-clock"));
        assert!(help.contains("Exits non-zero"));
        Ok(())
    }

    #[test]
    fn inspect_parses_workflow_run_from_and_mock() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "inspect", WORKFLOW_ID])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Inspect {
            workflow_id,
            run_id,
            from,
            mock,
        }))) = cli.command
        else {
            anyhow::bail!("expected inspect command");
        };
        assert_eq!(workflow_id, WORKFLOW_ID);
        assert!(run_id.is_none());
        assert!(from.is_none());
        assert!(mock.is_none());

        let cli = Cli::try_parse_from(["aion", "inspect", WORKFLOW_ID, "--run-id", RUN_ID])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Inspect { run_id, .. }))) =
            cli.command
        else {
            anyhow::bail!("expected inspect command");
        };
        assert_eq!(run_id.as_deref(), Some(RUN_ID));

        let cli = Cli::try_parse_from([
            "aion",
            "inspect",
            WORKFLOW_ID,
            "--from",
            "3",
            "--mock",
            r#"{"kind":"timer_fired"}"#,
        ])?;
        let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Inspect {
            from, mock, ..
        }))) = cli.command
        else {
            anyhow::bail!("expected inspect command");
        };
        assert_eq!(from, Some(3));
        assert_eq!(mock.as_deref(), Some(r#"{"kind":"timer_fired"}"#));
        Ok(())
    }

    #[test]
    fn inspect_help_documents_oplog_read_and_what_if() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(inspect) = command.find_subcommand_mut("inspect") else {
            anyhow::bail!("inspect subcommand should be registered");
        };
        let help = inspect.render_long_help().to_string();

        assert!(help.contains("no debug-only log and no second store"));
        assert!(help.contains("random()"));
        assert!(help.contains("divergent command"));
        assert!(help.contains("--from"));
        assert!(help.contains("--mock"));
        assert!(help.contains("what-if"));
        Ok(())
    }

    #[test]
    fn describe_help_documents_raw_payload_flag() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(describe) = command.find_subcommand_mut("describe") else {
            anyhow::bail!("describe subcommand should be registered");
        };
        let help = describe.render_long_help().to_string();

        assert!(help.contains("--raw"));
        assert!(help.contains("Show raw payload byte arrays instead of decoded payloads"));
        Ok(())
    }

    #[test]
    fn query_arguments_default_to_the_null_document_and_accept_a_json_flag() -> anyhow::Result<()> {
        let omitted = Cli::try_parse_from(vec!["aion", "query", WORKFLOW_ID, "state"])?;
        let supplied = Cli::try_parse_from(vec![
            "aion",
            "query",
            WORKFLOW_ID,
            "state",
            "--arguments",
            "{\"n\":7}",
        ])?;

        for (cli, expected) in [(omitted, "null"), (supplied, "{\"n\":7}")] {
            let Some(Command::Client(ClientCommand::Remote(RemoteCommand::Query {
                arguments,
                ..
            }))) = cli.command
            else {
                anyhow::bail!("expected the query command");
            };
            assert_eq!(arguments, expected);
            // Whatever the flag carried reaches the client as a JSON payload;
            // the omitted case is the canonical "no arguments" document, not
            // an empty byte string.
            let expected_value: Value = serde_json::from_str(expected)?;
            assert_eq!(
                json_payload(&arguments)?,
                Payload::from_json(&expected_value)?
            );
        }
        Ok(())
    }

    #[test]
    fn workflow_operations_allow_omitted_run_id() -> anyhow::Result<()> {
        let commands = [
            vec!["aion", "describe", WORKFLOW_ID],
            vec!["aion", "tail", WORKFLOW_ID, "--no-follow"],
            vec!["aion", "signal", WORKFLOW_ID, "poke", "--payload", "{}"],
            vec!["aion", "cancel", WORKFLOW_ID],
            vec!["aion", "reopen", WORKFLOW_ID],
            vec!["aion", "query", WORKFLOW_ID, "state"],
        ];

        for args in commands {
            let cli = Cli::try_parse_from(args)?;
            let Some(command) = cli.command else {
                anyhow::bail!("expected a subcommand");
            };
            match command {
                Command::Client(ClientCommand::Remote(
                    RemoteCommand::Describe { run_id, .. }
                    | RemoteCommand::Signal { run_id, .. }
                    | RemoteCommand::Cancel { run_id, .. }
                    | RemoteCommand::Reopen { run_id, .. }
                    | RemoteCommand::Pause { run_id, .. }
                    | RemoteCommand::Resume { run_id, .. }
                    | RemoteCommand::Query { run_id, .. },
                )) => assert!(run_id.is_none()),
                Command::Client(ClientCommand::Remote(RemoteCommand::Tail(args))) => {
                    assert!(args.run_id.is_none());
                }
                Command::Server(_)
                | Command::Stop(_)
                | Command::Update(_)
                | Command::Store(_)
                | Command::Dev(_)
                | Command::Awl { .. }
                | Command::Worker { .. }
                | Command::Client(
                    ClientCommand::Remote(
                        RemoteCommand::Start { .. }
                        | RemoteCommand::List { .. }
                        | RemoteCommand::Inspect { .. }
                        // `retire` targets the LOOP, not a run, so it carries
                        // no `run_id` for this sweep to assert on.
                        | RemoteCommand::Retire { .. },
                    )
                    | ClientCommand::New(_)
                    | ClientCommand::Generate { .. }
                    | ClientCommand::Check { .. }
                    | ClientCommand::Input { .. }
                    | ClientCommand::Package { .. }
                    | ClientCommand::Deploy { .. }
                    | ClientCommand::Run { .. }
                    | ClientCommand::Versions { .. }
                    | ClientCommand::Route { .. }
                    | ClientCommand::Unload { .. },
                ) => {
                    anyhow::bail!("expected workflow operation command")
                }
            }
        }
        Ok(())
    }

    #[test]
    fn signal_cancel_and_query_accept_optional_run_id() -> anyhow::Result<()> {
        let commands = [
            vec![
                "aion",
                "signal",
                WORKFLOW_ID,
                "poke",
                "--run-id",
                RUN_ID,
                "--payload",
                "{}",
            ],
            vec!["aion", "cancel", WORKFLOW_ID, "--run-id", RUN_ID],
            vec!["aion", "reopen", WORKFLOW_ID, "--run-id", RUN_ID],
            vec!["aion", "query", WORKFLOW_ID, "state", "--run-id", RUN_ID],
        ];

        for args in commands {
            let cli = Cli::try_parse_from(args)?;
            let Some(command) = cli.command else {
                anyhow::bail!("expected a subcommand");
            };
            match command {
                Command::Client(ClientCommand::Remote(
                    RemoteCommand::Signal { run_id, .. }
                    | RemoteCommand::Cancel { run_id, .. }
                    | RemoteCommand::Reopen { run_id, .. }
                    | RemoteCommand::Pause { run_id, .. }
                    | RemoteCommand::Resume { run_id, .. }
                    | RemoteCommand::Query { run_id, .. },
                )) => assert_eq!(run_id.as_deref(), Some(RUN_ID)),
                Command::Server(_)
                | Command::Stop(_)
                | Command::Update(_)
                | Command::Store(_)
                | Command::Dev(_)
                | Command::Awl { .. }
                | Command::Worker { .. }
                | Command::Client(
                    ClientCommand::Remote(
                        RemoteCommand::Start { .. }
                        | RemoteCommand::List { .. }
                        | RemoteCommand::Describe { .. }
                        | RemoteCommand::Tail(_)
                        | RemoteCommand::Inspect { .. }
                        // `retire` targets the LOOP, not a run.
                        | RemoteCommand::Retire { .. },
                    )
                    | ClientCommand::New(_)
                    | ClientCommand::Generate { .. }
                    | ClientCommand::Check { .. }
                    | ClientCommand::Input { .. }
                    | ClientCommand::Package { .. }
                    | ClientCommand::Deploy { .. }
                    | ClientCommand::Run { .. }
                    | ClientCommand::Versions { .. }
                    | ClientCommand::Route { .. }
                    | ClientCommand::Unload { .. },
                ) => {
                    anyhow::bail!("expected run-targeted workflow operation command")
                }
            }
        }
        Ok(())
    }

    #[test]
    fn deploy_subcommands_parse_with_global_token() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "--token",
            "operator-token",
            "deploy",
            "dist/order.aion",
        ])?;
        assert_eq!(cli.token.as_deref(), Some("operator-token"));
        let Some(Command::Client(ClientCommand::Deploy { archive })) = cli.command else {
            anyhow::bail!("expected deploy command");
        };
        assert_eq!(archive, Path::new("dist/order.aion"));

        let cli = Cli::try_parse_from(["aion", "versions", "--workflow-type", "order"])?;
        assert!(cli.token.is_none());
        let Some(Command::Client(ClientCommand::Versions { workflow_type })) = cli.command else {
            anyhow::bail!("expected versions command");
        };
        assert_eq!(workflow_type.as_deref(), Some("order"));

        let hash = "a".repeat(64);
        let cli = Cli::try_parse_from(["aion", "route", "order", &hash])?;
        let Some(Command::Client(ClientCommand::Route {
            workflow_type,
            content_hash,
        })) = cli.command
        else {
            anyhow::bail!("expected route command");
        };
        assert_eq!(workflow_type, "order");
        assert_eq!(content_hash, hash);

        let cli = Cli::try_parse_from(["aion", "unload", "order", &hash])?;
        let Some(Command::Client(ClientCommand::Unload {
            workflow_type,
            content_hash,
        })) = cli.command
        else {
            anyhow::bail!("expected unload command");
        };
        assert_eq!(workflow_type, "order");
        assert_eq!(content_hash, hash);
        Ok(())
    }

    #[test]
    fn deploy_help_documents_the_operator_surface() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(deploy) = command.find_subcommand_mut("deploy") else {
            anyhow::bail!("deploy subcommand should be registered");
        };
        let help = deploy.render_long_help().to_string();
        assert!(help.contains("operator API"));
        assert!(help.contains("deploy"));
        Ok(())
    }

    #[test]
    fn server_subcommand_converts_all_overrides() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "server",
            "--config",
            "dev-config.toml",
            "--listen-address",
            "127.0.0.1:18080",
            "--grpc-address",
            "127.0.0.1:18051",
            "--store-url",
            "aion.db",
            "--scheduler-threads",
            "2",
            "--drain-timeout",
            "45",
        ])?;
        let Some(Command::Server(args)) = cli.command else {
            anyhow::bail!("expected server command");
        };
        let overrides = CliOverrides::from(args.run);

        assert_eq!(
            overrides.config_path,
            Some(PathBuf::from("dev-config.toml"))
        );
        assert_eq!(
            overrides.listen_address.map(|address| address.port()),
            Some(18080)
        );
        assert_eq!(
            overrides.grpc_address.map(|address| address.port()),
            Some(18051),
            "the gRPC bind override never reached the server config"
        );
        assert_eq!(overrides.store_url.as_deref(), Some("aion.db"));
        assert_eq!(overrides.scheduler_threads, Some(2));
        assert_eq!(overrides.drain_timeout_seconds, Some(45));
        Ok(())
    }

    /// #114: `--endpoint` was accepted by `aion server`, advertised in its
    /// `--help`, and never read. An operator moving the bind off a port
    /// already in use got a server on the default port instead. It now
    /// refuses, and the refusal names the flag that does work — a refusal
    /// pointing at the wrong flag is barely better than the silence.
    #[test]
    fn the_server_refuses_a_client_endpoint_instead_of_binding_the_default() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "--endpoint", "127.0.0.1:51099", "server"])?;
        assert!(matches!(cli.command, Some(Command::Server(_))));

        let refusal = server_endpoint_refusal(cli.endpoint.as_deref())
            .ok_or_else(|| anyhow::anyhow!("an explicit --endpoint was accepted by the server"))?;
        assert!(refusal.contains("127.0.0.1:51099"));
        assert!(refusal.contains("--grpc-address"));
        assert!(refusal.contains("AION_SERVER_GRPC_ADDRESS"));
        Ok(())
    }

    /// The control verbs are local — they signal a pid under this machine's
    /// `AION_HOME` — so a named `--endpoint` is a flag they can never honour.
    /// Before this refusal, `aion --endpoint <remote> server stop` parsed,
    /// ignored the endpoint, and `SIGTERM`ed the LOCAL server: a destructive
    /// action on the wrong machine, delivered silently. The refusal must fire
    /// for both spellings (flag before and after the subcommand — it is
    /// `global = true`) — the rows below are the full verbs × spellings
    /// grid — must say the verb is local, not that it cannot bind, and must
    /// open with a rendered verb PHRASE ("refusing to report status", never
    /// the bare subcommand token).
    #[test]
    fn the_control_verbs_refuse_a_client_endpoint_instead_of_acting_locally() -> anyhow::Result<()>
    {
        for (argv, verb, action) in [
            (
                vec!["aion", "--endpoint", "10.0.0.9:50051", "server", "stop"],
                "stop",
                "stop",
            ),
            (
                vec!["aion", "server", "stop", "--endpoint", "10.0.0.9:50051"],
                "stop",
                "stop",
            ),
            (
                vec!["aion", "--endpoint", "10.0.0.9:50051", "server", "status"],
                "status",
                "report status",
            ),
            (
                vec!["aion", "server", "status", "--endpoint", "10.0.0.9:50051"],
                "status",
                "report status",
            ),
        ] {
            let cli = Cli::try_parse_from(&argv)?;
            let refusal = control_verb_endpoint_refusal(verb, action, cli.endpoint.as_deref())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "an explicit --endpoint was accepted by a control verb: {argv:?}"
                    )
                })?;
            assert!(refusal.contains("10.0.0.9:50051"));
            assert!(
                refusal.contains(&format!("refusing to {action}")),
                "the refusal must open with its rendered verb phrase: {refusal}"
            );
            assert!(
                refusal.contains(&format!("aion server {verb}")),
                "the refusal must name ITS OWN verb, not a sibling's: {refusal}"
            );
            assert!(
                refusal.contains("local"),
                "the refusal must say the verb is local: {refusal}"
            );
            assert!(
                refusal.contains("run the verb on that machine"),
                "the refusal must name the remedy: {refusal}"
            );
        }
        // The control: a bare verb has nothing to refuse.
        let bare = Cli::try_parse_from(["aion", "server", "stop"])?;
        assert!(control_verb_endpoint_refusal("stop", "stop", bare.endpoint.as_deref()).is_none());
        Ok(())
    }

    /// The control: without the flag there is nothing to refuse, and the
    /// client default still applies. A refusal that fired unconditionally
    /// would pass every assertion above and break every other verb.
    #[test]
    fn a_server_run_without_endpoint_is_not_refused_and_clients_keep_the_default()
    -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "server"])?;
        assert!(server_endpoint_refusal(cli.endpoint.as_deref()).is_none());
        assert_eq!(cli.endpoint(), format!("http://{DEFAULT_ENDPOINT}"));

        let dialled = Cli::try_parse_from(["aion", "--endpoint", "example:1", "list"])?;
        assert_eq!(dialled.endpoint(), "http://example:1");
        Ok(())
    }

    #[test]
    fn server_subcommand_maps_authoring_gleam_path() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "server",
            "--gleam-path",
            "/usr/local/bin/gleam",
            "--authoring-project-root",
            "/srv/aion/authoring",
        ])?;
        let Some(Command::Server(args)) = cli.command else {
            anyhow::bail!("expected server command");
        };
        let overrides = CliOverrides::from(args.run);

        assert_eq!(
            overrides.gleam_path,
            Some(PathBuf::from("/usr/local/bin/gleam"))
        );
        assert_eq!(
            overrides.authoring_project_root,
            Some(PathBuf::from("/srv/aion/authoring"))
        );
        Ok(())
    }

    #[test]
    fn server_workflow_package_flag_is_repeatable() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "server",
            "--workflow-package",
            "examples/hello-world/hello-world.aion",
            "--workflow-package",
            "local.aion",
        ])?;
        let Some(Command::Server(args)) = cli.command else {
            anyhow::bail!("expected server command");
        };
        let overrides = CliOverrides::from(args.run);

        assert_eq!(
            overrides.workflow_packages,
            vec![
                PathBuf::from("examples/hello-world/hello-world.aion"),
                PathBuf::from("local.aion"),
            ]
        );
        Ok(())
    }

    #[test]
    fn server_help_documents_the_server_surface() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(server) = command.find_subcommand_mut("server") else {
            anyhow::bail!("server subcommand should be registered");
        };
        let help = server.render_long_help().to_string();

        assert!(help.contains("Run the Aion workflow server"));
        assert!(help.contains("--config"));
        assert!(help.contains("--workflow-package"));
        Ok(())
    }

    #[test]
    fn top_level_help_lists_the_server_subcommand() -> anyhow::Result<()> {
        let error = Cli::try_parse_from(["aion", "--help"])
            .err()
            .ok_or_else(|| anyhow::anyhow!("help should exit early"))?;

        assert_eq!(error.kind(), clap::error::ErrorKind::DisplayHelp);
        let help = error.to_string();
        assert!(help.contains("server"));
        assert!(help.contains("Run the Aion workflow server"));
        Ok(())
    }

    #[test]
    fn bare_invocation_parses_as_the_launcher() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion"])?;
        assert!(cli.command.is_none());
        assert!(!cli.foreground);
        assert!(!cli.no_open);
        assert!(super::launcher_flag_refusal(&cli).is_none());

        let cli = Cli::try_parse_from(["aion", "--foreground", "--no-open"])?;
        assert!(cli.command.is_none());
        assert!(cli.foreground);
        assert!(cli.no_open);
        assert!(super::launcher_flag_refusal(&cli).is_none());
        Ok(())
    }

    #[test]
    fn launcher_flags_with_a_subcommand_are_refused_by_name() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "--foreground", "server"])?;
        let refusal = super::launcher_flag_refusal(&cli)
            .ok_or_else(|| anyhow::anyhow!("expected a refusal"))?;
        assert!(refusal.contains("`--foreground` configures"));

        let cli = Cli::try_parse_from(["aion", "--no-open", "versions"])?;
        let refusal = super::launcher_flag_refusal(&cli)
            .ok_or_else(|| anyhow::anyhow!("expected a refusal"))?;
        assert!(refusal.contains("`--no-open` configures"));

        let cli = Cli::try_parse_from(["aion", "--foreground", "--no-open", "server"])?;
        let refusal = super::launcher_flag_refusal(&cli)
            .ok_or_else(|| anyhow::anyhow!("expected a refusal"))?;
        assert!(refusal.contains("`--foreground` and `--no-open` configure"));
        Ok(())
    }

    #[test]
    fn dev_parses_path_gleam_path_and_optional_debounce() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "dev",
            "examples/hello-world",
            "--gleam-path",
            "/usr/local/bin/gleam",
            "--debounce-ms",
            "200",
        ])?;
        let Some(Command::Dev(args)) = cli.command else {
            anyhow::bail!("expected dev command");
        };
        assert_eq!(args.path, Path::new("examples/hello-world"));
        assert_eq!(args.gleam_path, Path::new("/usr/local/bin/gleam"));
        assert_eq!(
            args.debounce_ms,
            Some(std::time::Duration::from_millis(200))
        );

        // path defaults to the current directory; debounce is optional and
        // omitted by default (no invented interval).
        let cli = Cli::try_parse_from(["aion", "dev", "--gleam-path", "gleam"])?;
        let Some(Command::Dev(args)) = cli.command else {
            anyhow::bail!("expected dev command");
        };
        assert_eq!(args.path, Path::new("."));
        assert!(args.debounce_ms.is_none());
        Ok(())
    }

    #[test]
    fn dev_requires_gleam_path_and_rejects_zero_debounce() {
        // --gleam-path is required: there is no default gleam binary (ADR-001).
        assert!(Cli::try_parse_from(["aion", "dev"]).is_err());
        // A zero debounce window is a no-op the author did not mean to request.
        assert!(
            Cli::try_parse_from(["aion", "dev", "--gleam-path", "gleam", "--debounce-ms", "0"])
                .is_err()
        );
    }

    #[test]
    fn dev_help_documents_the_instant_loop_and_pinning() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(dev) = command.find_subcommand_mut("dev") else {
            anyhow::bail!("dev subcommand should be registered");
        };
        let help = dev.render_long_help().to_string();

        assert!(help.contains("--gleam-path"));
        assert!(help.contains("--debounce-ms"));
        assert!(help.contains("no engine restart"));
        assert!(help.contains("in-flight"));
        Ok(())
    }

    #[test]
    fn worker_agent_parses_the_document_and_repeatable_liminal_addresses() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from([
            "aion",
            "worker",
            "agent",
            "flows/review.awl",
            "--liminal-address",
            "127.0.0.1:50061",
            "--liminal-address",
            "127.0.0.1:50062",
            "--identity",
            "reviewer-worker",
            "--task-queue",
            "reviewers",
        ])?;
        let Some(Command::Worker {
            command: super::WorkerCommand::Agent(args),
        }) = cli.command
        else {
            anyhow::bail!("expected worker agent command");
        };
        assert_eq!(args.awl.as_deref(), Some(Path::new("flows/review.awl")));
        // Repeating the flag builds an ordered candidate list, which is what lets a
        // worker migrate to a survivor after its server dies.
        assert_eq!(
            args.liminal_addresses,
            vec![
                String::from("127.0.0.1:50061"),
                String::from("127.0.0.1:50062")
            ]
        );
        assert_eq!(args.identity.as_deref(), Some("reviewer-worker"));
        assert_eq!(args.task_queue.as_deref(), Some("reviewers"));
        assert!(!args.check);

        // Every connection setting is optional at PARSE time — the environment may
        // supply it — and `--check` needs none of them.
        let cli = Cli::try_parse_from(["aion", "worker", "agent", "flows/review.awl", "--check"])?;
        let Some(Command::Worker {
            command: super::WorkerCommand::Agent(args),
        }) = cli.command
        else {
            anyhow::bail!("expected worker agent command");
        };
        assert!(args.check);
        assert!(args.liminal_addresses.is_empty());
        assert!(args.identity.is_none());
        Ok(())
    }

    /// MIGRATION RIDER (a): the VERB SURVIVES and takes a document reference.
    ///
    /// Deleting the launch flags must not delete the door a human starts a worker
    /// through. The assertion is on the parse, not on the help text: a verb that
    /// documents itself and no longer parses is the same outage.
    #[test]
    fn the_agent_verb_survives_the_flag_deletion_and_takes_a_document() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "worker", "agent", "flows/review.awl"])?;
        let Some(Command::Worker {
            command: super::WorkerCommand::Agent(args),
        }) = cli.command
        else {
            anyhow::bail!("`aion worker agent <FILE.awl>` must still be the door in");
        };
        assert_eq!(args.awl.as_deref(), Some(Path::new("flows/review.awl")));
        Ok(())
    }

    /// MIGRATION RIDER: the per-setting launch flags are DELETED, not deprecated and
    /// not kept alongside.
    ///
    /// Asserted as a PARSE REFUSAL of each one by name, because that is the only
    /// difference an operator can observe between a deleted flag and a silently ignored
    /// one — and a silently ignored `--acp-permission` reads exactly like a permission
    /// policy being enforced.
    #[test]
    fn every_migrated_launch_flag_is_refused_by_the_parser() {
        for (flag, value) in [
            ("--harness", Some("acp")),
            ("--concurrency", Some("4")),
            ("--reconnect-initial-backoff", Some("5")),
            ("--reconnect-max-backoff", Some("60")),
            ("--reconnect-max-attempts", Some("100")),
            ("--acp-command", Some("/usr/local/bin/claude-code-acp")),
            ("--acp-arg", Some("--acp")),
            ("--acp-cwd", Some("/srv/workspace")),
            ("--acp-permission", Some("deny")),
            ("--acp-env", Some("KEY=value")),
            ("--acp-exit-grace", Some("2")),
            ("--norn-binary", Some("/usr/local/bin/norn")),
            ("--norn-arg", Some("--account")),
            ("--norn-unset-env", Some("OPENAI_API_KEY")),
        ] {
            let mut argv = vec!["aion", "worker", "agent", "flows/review.awl", flag];
            argv.extend(value);
            assert!(
                Cli::try_parse_from(&argv).is_err(),
                "`{flag}` moved into the document's `harness` section and must no longer \
                 parse; accepting it would leave an operator believing a setting is in force \
                 when the document decides it"
            );
        }
    }

    /// A managed-worker record minted BEFORE the launch settings moved replays its stored
    /// argv on every supervised restart, and nobody is at a terminal to read a generic
    /// "unexpected argument". This is the exact argv such a record carries — the tail a
    /// `WorkerArtifactRef::Builtin` stores — and what it must earn is the NAMED refusal,
    /// not clap's bare exit-2.
    ///
    /// The supervisor forwards a managed worker's stderr into the server's tracing output
    /// tagged with the worker name, so this sentence is what an operator finds against the
    /// record that could not start.
    #[test]
    fn the_argv_a_pre_change_managed_worker_record_replays_earns_a_named_refusal() {
        // Verbatim from market-mirror's OPERATIONS.md, the surface the CLAUDECODE
        // incident lived on — a record minted from that runbook carries exactly this.
        let stored_verb = [
            "worker",
            "agent",
            "workflows/mm_fleet_loop.awl",
            "--harness",
            "norn",
            "--task-queue",
            "mm_loop_build",
            "--identity",
            "mm-loop-builder",
            "--liminal-address",
            "127.0.0.1:50061",
            "--concurrency",
            "4",
            "--reconnect-initial-backoff",
            "1",
            "--reconnect-max-backoff",
            "30",
            "--reconnect-max-attempts",
            "100",
            "--norn-arg=--account",
            "--norn-arg=default",
        ];

        // It must still be REFUSED — a tolerant parse would be a compat shim.
        let mut argv = vec!["aion"];
        argv.extend(stored_verb);
        assert!(
            Cli::try_parse_from(&argv).is_err(),
            "a stored record does not buy a deleted flag its way back in"
        );

        // ...and the refusal must NAME what happened, on the record-driven path as much
        // as on a hand-typed one.
        let owned: Vec<String> = argv.into_iter().map(str::to_owned).collect();
        let Some(advice) = super::migrated_flag_advice(&owned) else {
            unreachable!("an argv carrying deleted flags must earn the advice");
        };
        for expected in [
            "--harness",
            "--concurrency",
            "--reconnect-initial-backoff",
            // The `=`-joined form is what the runbook actually used; matching only the
            // separated spelling would miss every record minted from it.
            "--norn-arg",
            "harness",
            "managed-worker",
            "re-mint",
            "worker-deployments",
        ] {
            assert!(
                advice.contains(expected),
                "the refusal must name `{expected}`, got:\n{advice}"
            );
        }
        assert!(
            !advice.contains("--task-queue") && !advice.contains("--identity"),
            "a flag that SURVIVED must not be named as deleted, or the operator strips a \
             setting the verb still needs:\n{advice}"
        );
    }

    /// THE VERB GATE. `--concurrency` and the three `--reconnect-*` are DELETED on
    /// `worker agent` and LIVE, REQUIRED flags on `worker awl` — same spellings, two
    /// verbs, opposite truths.
    ///
    /// This argv is the start command `docs/workers/AWL-WORKER-DOCUMENTS.md` publishes,
    /// with one typo'd flag so the parse fails and the advice path runs. It must earn
    /// NOTHING: telling this operator those flags are deleted would send them to write a
    /// `harness` section into a worker-family document, which the checker forbids
    /// outright — a refusal reached from a refusal, over a flag that was never wrong.
    #[test]
    fn a_worker_awl_invocation_earns_no_migration_advice() {
        let published: Vec<String> = [
            "aion",
            "worker",
            "awl",
            "greeter.awl",
            "--concurrency",
            "4",
            "--reconnect-initial-backoff",
            "0.1",
            "--reconnect-max-backoff",
            "5",
            "--reconnect-max-attempts",
            "1000",
            // The typo that makes clap fail and puts the advice path in play.
            "--identty",
            "greeter-worker",
        ]
        .into_iter()
        .map(str::to_owned)
        .collect();

        assert!(
            Cli::try_parse_from(&published).is_err(),
            "the control needs a FAILING parse, or the advice path never runs and this \
             test proves nothing"
        );
        assert_eq!(
            super::migrated_flag_advice(&published),
            None,
            "`--concurrency` and the reconnect trio are live REQUIRED flags on `worker \
             awl`; advice claiming they are deleted would send an operator to write a \
             `harness` section into a worker-family document, which cannot carry one"
        );
    }

    /// The presence twin of the gate: the SAME four flags, on the verb where they really
    /// were deleted, still earn the advice. Without this arm the gate above could pass on
    /// an implementation that never gives advice at all.
    #[test]
    fn the_same_flags_on_worker_agent_still_earn_the_advice() {
        let agent: Vec<String> = [
            "aion",
            "worker",
            "agent",
            "flows/review.awl",
            "--concurrency",
            "4",
            "--reconnect-initial-backoff",
            "1",
        ]
        .into_iter()
        .map(str::to_owned)
        .collect();
        let Some(advice) = super::migrated_flag_advice(&agent) else {
            unreachable!("the deleted flags on their own verb must earn the advice");
        };
        assert!(advice.contains("--concurrency") && advice.contains("harness"));
    }

    /// A deleted flag on a THIRD verb earns nothing either — the gate is an allow-list of
    /// one path, not a deny-list of one other.
    #[test]
    fn a_deleted_flag_on_an_unrelated_verb_earns_no_advice() {
        let unrelated: Vec<String> = ["aion", "worker", "shell", "--concurrency", "4"]
            .into_iter()
            .map(str::to_owned)
            .collect();
        assert_eq!(super::migrated_flag_advice(&unrelated), None);
    }

    /// An invocation carrying no deleted flag earns no advice at all — otherwise every
    /// ordinary typo would be answered with a lecture about the harness section.
    #[test]
    fn an_argv_with_no_deleted_flag_earns_no_migration_advice() {
        let argv: Vec<String> = ["aion", "worker", "agent", "flows/review.awl", "--nonsense"]
            .into_iter()
            .map(str::to_owned)
            .collect();
        assert!(
            super::migrated_flag_advice(&argv).is_none(),
            "the advice is for DELETED flags, not for every unknown argument"
        );
    }

    /// MIGRATION RIDER (b): a bare invocation refuses LOUDLY with the new shape named.
    ///
    /// The document is OPTIONAL to clap on purpose, so the refusal is this verb's own
    /// sentence rather than "the following required arguments were not provided" — which
    /// tells an operator arriving with the old flags nothing about where those settings
    /// went.
    #[test]
    fn a_bare_invocation_refuses_loudly_naming_the_new_shape() -> anyhow::Result<()> {
        let cli = Cli::try_parse_from(["aion", "worker", "agent"])?;
        let Some(Command::Worker {
            command: super::WorkerCommand::Agent(args),
        }) = cli.command
        else {
            anyhow::bail!("a bare invocation must PARSE, so the verb can refuse in its own words");
        };
        let Err(error) = args.document() else {
            anyhow::bail!("an invocation naming no document must be refused");
        };
        let message = error.to_string();
        for expected in [
            "aion worker agent <FILE.awl>",
            "harness",
            "aion awl check",
            "no longer a flag",
        ] {
            assert!(
                message.contains(expected),
                "the refusal must name `{expected}`, got:\n{message}"
            );
        }
        Ok(())
    }

    #[test]
    fn worker_agent_help_documents_the_model_it_serves() -> anyhow::Result<()> {
        let mut command = Cli::command();
        let Some(worker) = command.find_subcommand_mut("worker") else {
            anyhow::bail!("worker subcommand should be registered");
        };
        let Some(agent) = worker.find_subcommand_mut("agent") else {
            anyhow::bail!("worker agent subcommand should be registered");
        };
        let help = agent.render_long_help().to_string();

        assert!(help.contains("--liminal-address"));
        assert!(help.contains("--task-queue"));
        assert!(help.contains("--check"));
        // The model, not just the flags: where the queue and the schemas come from,
        // which actions are the server's, and that a queue is served whole.
        assert!(help.contains("worker` block's name"));
        assert!(help.contains("all-or-nothing"));
        assert!(help.contains("liminal worker transport"));
        assert!(help.contains("nothing is invented"));
        // Where the LAUNCH is declared, and every setting it carries: an operator must
        // be able to read off the help alone that these are document fields and not
        // flags, and which ones the kind they are declaring needs.
        for expected in [
            "`harness` section",
            "concurrency",
            "env_pass",
            "permission",
            "exit_grace",
            "no default",
        ] {
            assert!(help.contains(expected), "help must document `{expected}`");
        }
        // The two rulings an operator cannot afford to learn by surprise: the security
        // decision has no default, and the agent's environment is not inherited.
        assert!(help.contains("Nothing is inherited"));
        Ok(())
    }

    #[test]
    fn normalize_endpoint_preserves_explicit_scheme() {
        assert_eq!(
            normalize_endpoint("https://aion.example:50051"),
            "https://aion.example:50051"
        );
    }

    #[test]
    fn normalize_endpoint_defaults_to_http_scheme() {
        assert_eq!(
            normalize_endpoint("127.0.0.1:50051"),
            "http://127.0.0.1:50051"
        );
    }

    /// Whatever the operator typed or did not type, the address the CLI hands a
    /// verb to dial carries a scheme.
    ///
    /// `aion worker shell` used to take [`Cli::endpoint`] raw while every client
    /// verb wrapped it in [`normalize_endpoint`], and since [`DEFAULT_ENDPOINT`]
    /// is scheme-less, a shell worker started without an explicit `http://`
    /// never connected — silently, because that verb also installed no tracing
    /// subscriber, so the redial driver's warnings went nowhere. A worker that
    /// is up and connected to nothing is indistinguishable from an idle one.
    ///
    /// The pin is on the accessor, not on a list of dial sites, deliberately: a
    /// list of sites is an enumeration a new verb joins without anyone noticing,
    /// which is exactly the shape that produced the defect. What actually
    /// forbids the recurrence is the accessor's TYPE — it no longer yields an
    /// un-normalised address for a caller to pass along — and this pin holds the
    /// accessor to its side of that bargain. The one reader of the raw flag,
    /// [`server_endpoint_refusal`], needs the operator's literal text and never
    /// dials.
    #[test]
    fn every_dial_address_the_cli_hands_out_carries_a_scheme() -> anyhow::Result<()> {
        let shell = [
            "aion",
            "--endpoint",
            "example.internal:50051",
            "worker",
            "shell",
            "--manifest",
            "worker.toml",
            "--awl",
            "flow.awl",
        ];
        for argv in [
            vec!["aion", "list"],
            vec![
                "aion",
                "worker",
                "shell",
                "--manifest",
                "w.toml",
                "--awl",
                "f.awl",
            ],
            vec!["aion", "--endpoint", "127.0.0.1:50051", "list"],
            shell.to_vec(),
            vec!["aion", "--endpoint", "https://aion.example:50051", "list"],
        ] {
            let dialled = Cli::try_parse_from(&argv)?.endpoint();
            assert!(
                dialled.contains("://"),
                "the CLI handed `{argv:?}` the scheme-less address `{dialled}`; a verb that \
                 dials it as given connects to nothing, which is exactly how `aion worker \
                 shell` failed"
            );
        }

        // And normalising never doubles a scheme the operator supplied: an
        // address mangled into `http://https://…` fails just as silently as one
        // with no scheme at all.
        let explicit =
            Cli::try_parse_from(["aion", "--endpoint", "https://aion.example:50051", "list"])?;
        assert_eq!(explicit.endpoint(), "https://aion.example:50051");
        Ok(())
    }
}