agentplane 0.39.0

//! Run an agent that is only a file.
//!
//! ```sh
//! agentplane run agent.yaml --input '{"ticket": "printer on fire"}'
//! agentplane run room.yaml  --input '{"topic": "durable execution"}'
//! echo '{"ticket": "…"}' | agentplane run agent.yaml --input -
//! agentplane replay 01J… --store runs.redb --manifest agent.yaml --strict
//! agentplane card agent.yaml --url https://agents.example.com
//! agentplane validate room.yaml
//! agentplane digest room.yaml
//! ```
//!
//! A file may hold **several** manifests separated by `---`, the Kubernetes
//! packaging convention — so a multi-agent room deploys as one file with no
//! Rust anywhere. The file is packaging: each agent keeps its own digest.
//!
//! This binary is the last step of the declarative tier. A manifest with
//! `spec.execution` already needs no skill, but it still needed a `main` to
//! build a runtime and hand it a driver — and Rust is the thing the tier exists
//! to remove. With this, a YAML file and an API key are the whole agent.
//!
//! That is also what makes the digest claim exact rather than nearly true:
//! everything the agent does is in the file, so there is no accompanying program
//! that could diverge from it.
//!
//! # Why the arguments are parsed by a derive
//!
//! A hand-rolled parser reads one flag table for every verb, so a flag
//! belonging to one is **silently accepted** by another:
//!
//! ```sh
//! agentplane run agent.yaml --push-host evil.example.com --tokens /nonexistent
//! # both flags do nothing, and one of them is a security control
//! ```
//!
//! That is a declaration that does nothing, at the command line, and a declared
//! control must be enforced or rejected by the parser rather than accepted and
//! ignored. A derive makes the bad state unrepresentable: a flag lives on its
//! subcommand's struct, so `run --push-host` fails to parse by construction, and
//! `--strict` belongs to `replay` and fails on `run` the same way. `--help` is
//! generated from the structs that enforce the flags rather than being prose
//! that can describe an option nobody implemented.
//!
//! It costs crates on `cli` and **nothing on the library**, which is what
//! settles the trade: `cli` produces a binary and already carries hundreds.
//! Count them with `cargo tree --no-default-features --features cli -e normal
//! --prefix none | sort -u | wc -l` rather than reading a figure here — a number
//! in a comment is one nobody re-derives.

use std::process::ExitCode;
use std::sync::Arc;

use agentplane::core::Tainted;
use agentplane::journal::JournalStore;
use agentplane::manifest::Manifest;
use agentplane::model::ModelProvider;
use agentplane::runtime::{Mode, RunStatus, Runtime, RuntimeBuilder};
use agentplane::store::RedbStore;

fn install_tracing() {
    use tracing_subscriber::{EnvFilter, fmt};
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("warn,agentplane=info"));
    // `try_init` rather than `init`: failing to install a subscriber must not
    // take down a run that would otherwise have worked.
    let _ = fmt()
        .with_env_filter(filter)
        .with_writer(std::io::stderr)
        .try_init();
}

/// The command line.
///
/// One struct per verb, which is the whole point: a flag is reachable only from
/// the subcommand that uses it, so the parser refuses what the old hand-rolled
/// table silently accepted.
#[derive(clap::Parser, Debug)]
#[command(
    name = "agentplane",
    version,
    about = "Run an agent that is only a file",
    long_about = "Run, host and pin agents declared entirely in YAML.\n\n\
                  A file may hold several manifests separated by `---` (the \
                  Kubernetes convention), so a whole multi-agent room deploys as \
                  one file. Each document keeps its own digest — the file is \
                  packaging, not identity.",
    disable_help_subcommand = true
)]
struct Cli {
    #[command(subcommand)]
    verb: Verb,
}

#[derive(clap::Subcommand, Debug)]
enum Verb {
    /// Execute an agent once and print its answer.
    Run(RunArgs),
    /// Re-execute a recorded run: resume it, or verify it with --strict.
    Replay(ReplayArgs),
    /// Print the Agent Card a served manifest would advertise.
    Card(CardArgs),
    /// Host an agent as an A2A 1.0 peer.
    Serve(Box<ServeArgs>),
    /// Check every document in a file, and say what is in it.
    Validate(ValidateArgs),
    /// Print the manifest format as a JSON Schema, for editors and CI linters.
    Schema,
    /// Print the identity a registry pins.
    Digest(FileArgs),
    /// Check a journal's history and print what could not be checked.
    Audit(AuditArgs),
    /// Write a journal's records out as JSON Lines.
    Export(StoreArgs),
    /// Recompute an export and check it against its own checkpoint.
    Verify(VerifyArgs),
    /// Rebuild a store from an export, and prove it by its own checkpoint.
    Restore(RestoreArgs),
    /// Walk the case layer and tell erasure from loss, against the live stores.
    Drill(DrillArgs),
    /// Retire admission keys older than a window you choose.
    ForgetAdmissions(ForgetArgs),
    /// List the closed cases a retention pass would erase.
    Retain(RetainArgs),
    /// Throw or lift the emergency stop, and say what it covers.
    Halt(HaltArgs),
    /// List every emergency stop standing on a tenant.
    Halts(HaltsArgs),
    /// Place, lift, or list the legal holds that stop a retention pass.
    Hold(HoldArgs),
}

/// Retention, as a verb, for the tier that is a manifest and this binary.
///
/// `--older-than-days` is **required and has no default**, for the reason
/// [`ForgetArgs`]'s window is: a retention period is a legal and business
/// decision, and a crate that picked one would be choosing somebody else's.
///
/// `--reason` is required too. It lands on every tombstone and on each key
/// destruction, so a later read says *expired, on this date, for this reason*
/// rather than *missing* — which is the distinction the recovery drill's
/// three-way verdict is built on.
#[derive(clap::Args, Debug)]
struct RetainArgs {
    /// The store holding the case layer, and whose. Blob addresses and key
    /// scopes derive from the tenant, so a pass under the wrong one erases
    /// nothing and says it erased nothing.
    #[command(flatten)]
    at: StoreRef,

    /// Erase closed cases opened longer ago than this, in days. Required.
    #[arg(long)]
    older_than_days: u32,

    /// Why, recorded on every tombstone. Required.
    #[arg(long)]
    reason: String,

    /// List what a pass would erase. Required: this binary wires no blob
    /// store and no key ring, so listing is the only half it can perform.
    #[arg(long)]
    dry_run: bool,
}

/// Preservation, as a verb.
///
/// The counterpart of `retain`: that verb says what a sweep would destroy, this
/// one says what it may not. With no `--case`, it lists — because a hold that
/// can only be read by somebody who already knows which matter to ask about
/// delivers nothing to the person whose job is to find out what is still being
/// preserved and why.
#[derive(clap::Args, Debug)]
struct HoldArgs {
    /// The store holding the case layer, and whose.
    #[command(flatten)]
    at: StoreRef,

    /// The matter to place or lift a hold on. Omit to list every hold standing.
    #[arg(long)]
    case: Option<String>,

    /// Why this matter may not be destroyed. Required to place one: the person
    /// reviewing this listing in two years has only this sentence to act on.
    #[arg(long)]
    reason: Option<String>,

    /// Who is placing it. Required to place one.
    ///
    /// Recorded as **asserted**: nothing here verified it, and what it proves
    /// is that whoever ran this command could open the store. The operator API
    /// records the same field from the credential its authenticator checked,
    /// and the row says which of the two it was.
    #[arg(long)]
    actor: Option<String>,

    /// Lift the hold instead of placing it.
    #[arg(long, conflicts_with_all = ["reason", "actor"])]
    lift: bool,
}

/// The emergency stop, as a verb: an incident is the worst time to discover
/// that the brake needs a compiler.
///
/// `--reason` and `--actor` are required to halt and refused to lift: the next
/// person to look will be somebody else, possibly at three in the morning, and
/// *why* and *who* are the whole question. A lift needs neither, because it
/// restores the default and the row it clears is gone.
#[derive(clap::Args, Debug)]
struct HaltArgs {
    /// The store holding the halt, and which tenant to stop.
    #[command(flatten)]
    at: StoreRef,

    /// What to stop: `tenant`, `agent:<metadata.name>`,
    /// `revision:<manifest digest>`, or `subject:<delegation subject>`.
    ///
    /// `revision:` is the one to reach for when a bad deploy is the incident:
    /// it names the exact reviewed bytes, so a fix published as a new version
    /// runs while the broken revision stays stopped.
    #[arg(long, default_value = "tenant")]
    scope: String,

    /// Why. Required unless `--lift`.
    #[arg(long)]
    reason: Option<String>,

    /// Who is throwing it. Required unless `--lift`.
    ///
    /// It goes on the row, and it is recorded as **asserted** rather than
    /// authenticated: nothing here verified it, and what it proves is that
    /// whoever ran this command could open the store. The operator API records
    /// the same field from the credential its authenticator checked, and the
    /// two are told apart on the row rather than guessed at from the surface.
    ///
    /// Required rather than defaulted from the shell, because a name taken from
    /// `$USER` reads on the record exactly like one somebody chose to put
    /// there, and only one of those is true.
    #[arg(long)]
    actor: Option<String>,

    /// Lift this halt instead of setting it.
    #[arg(long, conflicts_with_all = ["reason", "actor"])]
    lift: bool,
}

/// What is stopped right now — the question a per-scope lookup cannot answer.
#[derive(clap::Args, Debug)]
struct HaltsArgs {
    /// The store holding the halts, and whose to read.
    #[command(flatten)]
    at: StoreRef,
}

/// Retention for the admission index, as a verb.
///
/// The same reasoning [`DrillArgs`] carries: `JournalStore::forget_admissions`
/// could only be reached by writing Rust, and a deployment that is only a YAML
/// file has an index that grows and no way to trim it.
///
/// `--older-than` is **required and has no default**. Retiring a key reopens
/// the door it closed, so a window shorter than the emitter's retry horizon
/// admits a second run on a timer — which is the failure the key exists to
/// prevent. A default here would be this crate choosing somebody else's retry
/// horizon for them.
#[derive(clap::Args, Debug)]
struct ForgetArgs {
    /// The store holding the admission index, and whose.
    #[command(flatten)]
    at: StoreRef,

    /// Retire keys claimed longer ago than this, as days. Required.
    ///
    /// It must exceed how long your emitter keeps retrying a delivery it has
    /// not seen a 2xx for.
    #[arg(long)]
    older_than_days: u32,
}

/// The live half of the case-layer drill, as a verb.
///
/// `Runtime::drill` reachable without writing Rust, which is the dependency the
/// declarative tier exists to remove: a deployment that is only a YAML file
/// still has to be able to rehearse its own recovery. The verb opens the same
/// store file the other journal verbs do; the case layer lives in it, so
/// `--store` is the whole wiring.
///
/// What this verb does NOT check: blob bytes and sealed-state keys. A redb
/// file holds no blob store, and this binary has no key-ring wiring — both
/// are named as unchecked in the report rather than silently passed, which is
/// the same honesty the library's own report shape enforces. An embedder
/// whose plane has those stores runs `Runtime::drill` with them wired.
#[derive(clap::Args, Debug)]
struct DrillArgs {
    /// The store holding the case layer to drill. Required — the drill walks
    /// cases, and a memory store this process did not write holds none.
    #[command(flatten)]
    at: StoreRef,
}

/// Rebuild a journal from an export.
#[derive(clap::Args, Debug)]
struct RestoreArgs {
    /// The export to read.
    file: String,

    /// Where to write the rebuilt journal, and under which tenant. Must not
    /// already hold these runs — this rebuilds a history rather than merging
    /// one. The tenant matters as much as the path: a restore into the unnamed
    /// default of a store whose plane serves `acme` rebuilds a history nobody
    /// serves.
    #[command(flatten)]
    at: StoreRef,
}

/// The restore drill: an export, and nothing else.
///
/// Takes a *file* rather than a store on purpose. This is the one verb that
/// needs neither the runtime that wrote the data nor the store it came from —
/// which is what makes it usable by somebody who was handed a copy and asked
/// whether it is the whole of it.
#[derive(clap::Args, Debug)]
struct VerifyArgs {
    /// The export to check. `-` reads standard input.
    file: String,

    /// Trust records signed by this key, as `<key-id>=<64 hex chars>`.
    /// Repeatable. With a key supplied, an unsigned record is a finding —
    /// that is the auditor's posture, since an unsigned record inside a
    /// signed history is the one an attacker who cannot sign would add.
    #[arg(long)]
    key: Vec<String>,

    /// The checkpoint this export is supposed to be a copy of, as a
    /// `tlog-checkpoint` note or as the JSON an audit report prints.
    ///
    /// This is the deletion check, and without it there is none. The Merkle
    /// root rebuilt from the file can otherwise only be compared with the
    /// file's own header — which an editor who dropped a run rewrites too —
    /// so the report says deletion went unchecked. Supply the checkpoint an
    /// earlier audit printed, or fetch one with `--witness`: the point is that
    /// it comes from somewhere other than the file being checked.
    #[arg(long)]
    checkpoint: Option<String>,

    /// Fetch the anchoring checkpoint from a witness, at its monitoring
    /// prefix. Repeatable.
    ///
    /// The deletion check needs a checkpoint from **outside** the store, and
    /// this is the only way to get one that the operator did not hand over.
    /// A witness keeps the last checkpoint it cosigned for a log and will only
    /// cosign one that provably extends it, so a run removed from the store is
    /// a size the witness still remembers.
    ///
    /// Two or more witnesses are worth naming: a split view is precisely two
    /// witnesses holding one size with different roots, and that is a finding
    /// no single anchor produces.
    #[arg(long = "witness")]
    witness: Vec<String>,

    /// A witness key to trust, as `<name>=<base64 Ed25519 public key>` —
    /// repeatable, and required by `--witness`.
    ///
    /// Without one a fetch could only report what a URL served. The whole
    /// argument for an outside anchor is that an independent party signed it,
    /// and a signature nobody checks makes that an argument about a status
    /// code.
    #[arg(long = "witness-key")]
    witness_key: Vec<String>,

    /// The log's origin line, when it is not this store's own.
    ///
    /// Defaults to what the store reports, which is right for an auditor
    /// holding the database. Naming it explicitly is for the case where the
    /// store's own answer is the thing under suspicion.
    #[arg(long)]
    origin: Option<String>,
}

/// What `audit` takes beyond the shared store arguments: the evidence.
///
/// These flags exist because the library call has taken this evidence all
/// along, while the verb hardcoded none of it — so the signature check and the
/// deletion check were real and unreachable without writing Rust, which is the
/// dependency these verbs exist to remove. A control that must be linked
/// against is not one an independent party holds.
#[derive(clap::Args, Debug)]
struct AuditArgs {
    #[command(flatten)]
    store: StoreArgs,

    /// Trust records signed by this key, as `<key-id>=<64 hex chars>`.
    /// Repeatable. Without one, the report says signatures went unchecked.
    #[arg(long)]
    key: Vec<String>,

    /// A checkpoint saved earlier, as JSON — the `current` field of a previous
    /// audit report. This is the deletion check: a log that shrank or forked
    /// since that checkpoint is a finding, and without one the report says
    /// deletion went unchecked.
    #[arg(long)]
    prior: Option<String>,

    /// Treat an unsigned record as a failure.
    ///
    /// Off by default because history written before signing was configured is
    /// legitimately unsigned, and a wall of failures over a healthy plane
    /// teaches the reader to ignore the report.
    #[arg(long)]
    require_signatures: bool,

    /// Fetch the anchoring checkpoint from a witness, at its monitoring
    /// prefix. Repeatable.
    ///
    /// The deletion check needs a checkpoint from **outside** the store, and
    /// this is the only way to get one that the operator did not hand over.
    /// A witness keeps the last checkpoint it cosigned for a log and will only
    /// cosign one that provably extends it, so a run removed from the store is
    /// a size the witness still remembers.
    ///
    /// Two or more witnesses are worth naming: a split view is precisely two
    /// witnesses holding one size with different roots, and that is a finding
    /// no single anchor produces.
    #[arg(long = "witness")]
    witness: Vec<String>,

    /// A witness key to trust, as `<name>=<base64 Ed25519 public key>` —
    /// repeatable, and required by `--witness`.
    ///
    /// Without one a fetch could only report what a URL served. The whole
    /// argument for an outside anchor is that an independent party signed it,
    /// and a signature nobody checks makes that an argument about a status
    /// code.
    #[arg(long = "witness-key")]
    witness_key: Vec<String>,

    /// The log's origin line, when it is not this store's own.
    ///
    /// Defaults to what the store reports, which is right for an auditor
    /// holding the database. Naming it explicitly is for the case where the
    /// store's own answer is the thing under suspicion.
    #[arg(long)]
    origin: Option<String>,
}

/// The arguments the two journal verbs share.
///
/// Both read a store and neither reads a manifest, which is the point: an
/// auditor holds a database file and a checkpoint somebody gave them, not the
/// deployment's source tree.
#[derive(clap::Args, Debug)]
struct StoreArgs {
    /// The journal to read, and whose. Required — there is nothing to audit or
    /// export in a memory store that this process did not itself write.
    #[command(flatten)]
    at: StoreRef,

    /// Which runs, by outcome. Repeatable. Defaults to every sealed outcome.
    #[arg(long)]
    outcome: Vec<String>,

    /// How many runs to consider per outcome.
    #[arg(long, default_value_t = 1000)]
    limit: usize,
}

/// Where a plane's state is, and whose.
///
/// One type rather than a `--store`/`--tenant` pair written out per verb. Four
/// verbs had both, five had only the store, and the binary could serve only the
/// unnamed tenant — so an operator who learnt `--tenant` from `halt` and reached
/// for `export` got an artifact about a different plane, empty, well-formed and
/// exit zero. A verb cannot now name a store without saying whose it is,
/// because there is no other way to name one.
#[derive(clap::Args, Debug, Clone)]
struct StoreRef {
    /// The plane's store: a redb file, or a `postgres://` connection string.
    #[arg(long, env = "AGENTPLANE_STORE")]
    store: String,

    /// Which tenant's plane. Defaults to the unnamed single-tenant plane.
    ///
    /// Every key in both backends leads with the tenant, so naming the wrong
    /// one is a *miss* rather than an error: the verb answers about a plane
    /// nobody runs and reports success.
    #[arg(long, env = "AGENTPLANE_TENANT")]
    tenant: Option<String>,
}

impl StoreRef {
    async fn open(&self) -> Result<Backend, String> {
        Backend::open(&self.store, self.tenant.as_deref()).await
    }
}

/// [`StoreRef`] for the two verbs a store is genuinely optional for.
///
/// `run` may journal to memory because it exits with its answer; `serve` refuses
/// without one and says why in its own words rather than clap's.
#[derive(clap::Args, Debug, Clone)]
struct MaybeStoreRef {
    /// The plane's store: a redb file, or a `postgres://` connection string.
    #[arg(long, env = "AGENTPLANE_STORE")]
    store: Option<String>,

    /// Which tenant's plane. Defaults to the unnamed single-tenant plane.
    #[arg(long, env = "AGENTPLANE_TENANT")]
    tenant: Option<String>,
}

impl MaybeStoreRef {
    /// Open the named store, or say nothing was named.
    ///
    /// `Ok(None)` rather than a default, so each caller decides what an absent
    /// store means: `run` journals to memory and says so, `serve` refuses.
    async fn open(&self) -> Result<Option<Backend>, String> {
        match &self.store {
            Some(spec) => Backend::open(spec, self.tenant.as_deref()).await.map(Some),
            None => Ok(None),
        }
    }
}

/// Build a verifier from repeated `--key <key-id>=<hex>` flags.
///
/// `None` when no key was given, so the report's `not_checked` half can say
/// signatures went unchecked — which is a different statement from checked and
/// clean, and the difference is the whole reason the field exists.
fn verifier_from(keys: &[String]) -> Result<Option<agentplane::policy::Ed25519Verifier>, String> {
    if keys.is_empty() {
        return Ok(None);
    }
    let mut verifier = agentplane::policy::Ed25519Verifier::new();
    for entry in keys {
        let Some((id, hex_key)) = entry.split_once('=') else {
            return Err(format!(
                "--key takes <key-id>=<64 hex chars>, got '{entry}' — the id is what records \
                 name as their signer, and the hex is the Ed25519 public key"
            ));
        };
        let mut bytes = [0u8; 32];
        hex::decode_to_slice(hex_key, &mut bytes)
            .map_err(|e| format!("--key {id}: not 64 hex characters: {e}"))?;
        verifier = verifier
            .trust(id, &bytes)
            .map_err(|e| format!("--key {id}: not a valid Ed25519 public key: {e}"))?;
    }
    Ok(Some(verifier))
}

#[derive(clap::Args, Debug)]
struct FileArgs {
    /// The manifest, or a `---`-separated file of them.
    manifest: String,
}

#[derive(clap::Args, Debug)]
struct ValidateArgs {
    /// The manifest, or a `---`-separated file of them.
    manifest: String,

    /// Require this annotation key to be present and non-empty. Repeatable.
    ///
    /// The runtime never reads `metadata.annotations` — that is what makes them
    /// safe to carry, and it is why nothing can notice a production agent that
    /// shipped without an owner. A control nobody checks is a convention.
    ///
    /// This does not change that: the check lives in review, the keys stay the
    /// deployment's own vocabulary, and no interpretation crosses the trust
    /// boundary. It is the division `--policy` already draws — the rule is
    /// yours, the enforcement is a job you run.
    #[arg(long = "require-annotation", value_name = "KEY")]
    require_annotation: Vec<String>,
}

#[derive(clap::Args, Debug)]
struct RunArgs {
    /// The manifest, or a `---`-separated file of them.
    manifest: String,

    /// The run's input, as JSON. `-` reads standard input. Defaults to `{}`.
    #[arg(long, conflicts_with = "input_file")]
    input: Option<String>,

    /// Read the run's input from a file instead.
    #[arg(long)]
    input_file: Option<String>,

    /// Which capability to run. Optional when the file leaves no doubt.
    #[arg(long)]
    capability: Option<String>,

    /// Journal on disk, and whose. Defaults to memory, which keeps nothing.
    #[command(flatten)]
    at: MaybeStoreRef,

    /// Run an MCP server as a child process and reach it as `tool://NAME/...`.
    ///
    /// Repeatable, one per server. The manifest grants the tools; this says only
    /// which transport reaches the server offering them, because an agent's
    /// digest must not change when it moves between a laptop and a cluster.
    /// Needs the `mcp-stdio` feature.
    #[arg(long, value_name = "NAME=COMMAND")]
    mcp: Vec<String>,

    /// Reach an A2A peer at URL as `tool://NAME/...`.
    ///
    /// Repeatable, one per peer. The manifest grants the capabilities; this
    /// says where the peer is. Its bearer token, when it needs one, comes from
    /// the environment as `AGENTPLANE_PEER_TOKEN_<NAME>` (upper-cased, `.`
    /// and `-` as `_`), never from the command line. Needs the `a2a` feature.
    #[arg(long, value_name = "NAME=URL")]
    peer: Vec<String>,
}

/// Re-execute a recorded run.
///
/// Its own verb rather than a `run --replay` flag, because the two share
/// almost nothing: a replay has no input, no capability choice and no default
/// store — the journal *is* the subject — and a flag table where half the
/// flags are meaningless under another flag is the silently-accepted-option
/// defect this parser was adopted to remove.
#[derive(clap::Args, Debug)]
struct ReplayArgs {
    /// The run to re-execute.
    run_id: String,

    /// The journal holding it, and whose. Required: there is nothing to replay
    /// in a memory store this process did not itself write.
    #[command(flatten)]
    at: StoreRef,

    /// The manifest (or `---`-separated room) the run executed under.
    ///
    /// Required, because a replay re-executes the deterministic zone and a
    /// declarative agent's manifest *is* that code. Hand it the same document;
    /// a journal written by a different declaration is divergence, and the
    /// run is quarantined rather than continued — which is the desired
    /// outcome, not a limitation.
    #[arg(long)]
    manifest: String,

    /// Verify rather than resume: read every effect back and fail if this
    /// build would do more, less, or different work than the record.
    #[arg(long)]
    strict: bool,

    /// Run an MCP server as a child process, as `run` takes it. A resume that
    /// continues past its recorded history dispatches live and may need one.
    #[arg(long, value_name = "NAME=COMMAND")]
    mcp: Vec<String>,

    /// Reach an A2A peer, as `run` takes it.
    #[arg(long, value_name = "NAME=URL")]
    peer: Vec<String>,
}

/// Print the Agent Card a served manifest would advertise.
#[derive(clap::Args, Debug)]
struct CardArgs {
    /// The manifest. A card names one agent, exactly as `serve` hosts one.
    manifest: String,

    /// The public base URL the card advertises — what `serve --url` would be
    /// handed, without serving anything.
    #[arg(long, env = "AGENTPLANE_URL")]
    url: String,
}

#[derive(clap::Args, Debug)]
struct ServeArgs {
    /// The manifest. `serve` hosts exactly one agent.
    manifest: String,

    /// Where callers reach this plane. Goes on the Agent Card, so it is the
    /// public URL rather than what you bind.
    #[arg(long, env = "AGENTPLANE_URL")]
    url: Option<String>,

    /// What to bind the peer surface to.
    #[arg(long, env = "AGENTPLANE_ADDR", default_value = "127.0.0.1:8080")]
    addr: String,

    /// A Cedar policy set. No default: a permissive engine and no engine are the
    /// same behaviour, and only one of them looks governed.
    #[arg(long, env = "AGENTPLANE_POLICY")]
    policy: Option<String>,

    /// Bearer tokens naming the callers this plane accepts.
    #[arg(long, env = "AGENTPLANE_TOKENS")]
    tokens: Option<String>,

    /// Journal on disk, and which tenant this plane serves. Required: a served
    /// task's id is a promise it can be fetched again.
    #[command(flatten)]
    at: MaybeStoreRef,

    /// Also serve the operator surface — the worklist, task decisions and
    /// `GET /runs?outcome=quarantined` — on its own listener.
    #[arg(long, env = "AGENTPLANE_OPERATOR_ADDR")]
    operator_addr: Option<String>,

    /// How often deadlines, task expiry, dead letters and due timers are swept.
    /// `0` runs the sweep from your own scheduler instead.
    #[arg(long, value_name = "SECS", env = "AGENTPLANE_SWEEP_EVERY")]
    sweep_every: Option<u32>,

    /// How often the recovery drill runs, in seconds. Off unless you say.
    ///
    /// `Runtime::drill` walks every case and holds its blob and sealed-state
    /// references against the live stores, telling *intact* from *erased by
    /// design* from *lost*. Nothing invoked it on a schedule, and a control
    /// that exists and is never exercised is one an audit cannot count — the
    /// rehearsal was left as something an operator would arrange, and the
    /// deployments that most need it are the ones that write no Rust.
    ///
    /// **Off by default, deliberately.** A drill is a full walk of the case
    /// layer, so its cost grows with the case layer and this crate does not
    /// know how large yours is or when your quiet hour is. Daily (`86400`) is
    /// the shape most deployments want; a finding is logged at `error` with
    /// the report attached, which is what an alert rule keys on.
    #[arg(long, value_name = "SECS", env = "AGENTPLANE_DRILL_EVERY")]
    drill_every: Option<u32>,

    /// Permit A2A push notifications to this exact host. Repeatable.
    ///
    /// Without one, push is not wired and the Agent Card advertises it as
    /// absent rather than claiming a capability nothing serves.
    #[arg(long, value_name = "HOST")]
    push_host: Vec<String>,

    /// Run an MCP server as a child process and reach it as `tool://NAME/...`.
    #[arg(long, value_name = "NAME=COMMAND")]
    mcp: Vec<String>,

    /// Reach an A2A peer at URL as `tool://NAME/...`, as `run` takes it.
    #[arg(long, value_name = "NAME=URL")]
    peer: Vec<String>,

    /// How long to keep working after a stop signal, in seconds.
    ///
    /// On `SIGTERM` or `SIGINT` this process stops accepting connections,
    /// finishes the requests already in hand, lets the periodic passes complete
    /// the tick they are in, and waits this long for the runs it started in the
    /// background to reach a journaled resting point.
    ///
    /// **It has to fit inside the supervisor's own grace period**, which is what
    /// sends `SIGKILL` afterwards — 30 seconds on Kubernetes and Docker unless
    /// raised. The default leaves margin under that. `0` exits as soon as the
    /// listeners are closed.
    ///
    /// What it buys: a run killed inside a tool call leaves an announced effect
    /// with no outcome, and no later reader can tell whether that call reached
    /// the world — so the effect's declared recovery decides, which for anything
    /// not safe to repeat means waiting for a person. Draining turns the
    /// ordinary case of a deploy back into an ordinary conclusion.
    #[arg(
        long,
        value_name = "SECS",
        env = "AGENTPLANE_DRAIN_SECS",
        default_value_t = 25
    )]
    drain_secs: u64,
}

/// The anchoring checkpoint an audit was given, and **how it was obtained**.
///
/// The basis of a fact is part of the fact. A checkpoint fetched from two
/// independent witnesses and verified against keys the reader supplied, and one
/// typed out of a ticket, are different grounds for the same verdict — and an
/// artifact that records only the checkpoint lets the second be read as the
/// first. That is trust laundering by omission, and it happens at a shell
/// redirect: the report goes to stdout, and a basis printed only to stderr is
/// gone the moment somebody writes `> report.json`.
///
/// So the basis travels with the report. What it records is what **this
/// command** established, never what the library verified — the audit checks
/// that a checkpoint *extends*, and cannot check who vouched for it.
#[derive(Debug, Default, serde::Serialize)]
struct Anchor {
    /// The checkpoint itself, held for the caller and **not serialized**.
    ///
    /// The report already names it — `held_to` on an audit, the header
    /// comparison on a verify — and one document carrying one checkpoint in
    /// two fields is two answers waiting to disagree. What this object adds is
    /// the half the report cannot have: how the checkpoint was obtained.
    #[serde(skip)]
    checkpoint: Option<agentplane::journal::Checkpoint>,
    /// Where it came from: a witness's monitoring prefix, or a file.
    #[serde(skip_serializing_if = "Option::is_none")]
    obtained_from: Option<String>,
    /// The witness keys whose cosignature over **this** checkpoint verified.
    ///
    /// Empty for a checkpoint read from a file: a file carries no signature
    /// this command can check, so it is an asserted fact and saying nothing is
    /// how that is said.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    cosigned_by: Vec<String>,
    /// Witnesses that were asked and gave no anchor, with why.
    ///
    /// Kept because a clean report over one witness's anchor and a clean report
    /// over three are different statements, and the difference is only visible
    /// here.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    unreached: Vec<String>,
    /// Two witnesses holding one tree size with two different roots.
    ///
    /// The event witnessing exists to detect. It fails the command: a finding
    /// that only prints is one nobody files.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    split_view: Vec<String>,
}

/// What `audit` prints: the library's report, and what this command
/// established about the evidence it fed in.
///
/// One document rather than two streams, so the basis cannot be separated from
/// the verdict by a redirect.
#[derive(serde::Serialize)]
struct AuditDocument<'a> {
    anchor: &'a Anchor,
    #[serde(flatten)]
    report: &'a agentplane::audit::AuditReport,
}

/// What `verify` prints: the file's report, and the anchor it was held to.
#[derive(serde::Serialize)]
struct VerifyDocument<'a> {
    anchor: &'a Anchor,
    #[serde(flatten)]
    report: &'a agentplane::export::VerifyReport,
}

/// The two verbs that read a journal instead of a manifest.
///
/// One function because they differ only in what they do with the run list, and
/// the half that is easy to get wrong — *which* runs, and saying so when the
/// limit truncated — is the half they share. `audit` is `None` for an export,
/// and carries the evidence flags for an audit.
/// The trusted witness keys an auditor named, as the reader takes them.
fn witness_keys(keys: &[String]) -> Result<Vec<agentplane::journal::TrustedWitness>, String> {
    let mut out = Vec::new();
    for entry in keys {
        let Some((name, encoded)) = entry.split_once('=') else {
            return Err(format!(
                "--witness-key takes <name>=<base64 Ed25519 public key>, got '{entry}' — \
                 the name is the one the witness signs its lines with, and the key is \
                 what makes a cosignature checkable rather than a string"
            ));
        };
        // Base64 where the sibling `--key` takes hex, and the difference is
        // whose key it is. A record-signing key id is this deployment's own,
        // so an operator can publish it in whatever form the flag wants; a
        // witness key arrives from a third party, published in base64 by
        // every witness in the existing network. Making an auditor re-encode
        // it by hand adds a step where a typo reads as "the witness is down".
        let bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, encoded)
            .map_err(|e| format!("--witness-key {name}: not base64: {e}"))?;
        let key: [u8; 32] = bytes.try_into().map_err(|b: Vec<u8>| {
            format!(
                "--witness-key {name}: an Ed25519 public key is 32 bytes, this is {}",
                b.len()
            )
        })?;
        out.push(agentplane::journal::TrustedWitness::ed25519(name, key));
    }
    Ok(out)
}

/// What the named witnesses hold for `origin`, and whether they agree.
///
/// The **anchor** an auditor needs is one checkpoint from outside the store.
/// Naming several witnesses buys something a single one cannot: a split view
/// is exactly two witnesses holding one size with different roots, and no
/// single anchor exhibits it. So this returns the highest checkpoint any of
/// them cosigned, and reports a disagreement on stderr as what it is — the
/// event witnessing exists to detect, found by the party it exists to protect.
///
/// `Ok(None)` when no witness was named. A named witness that has never seen
/// this log is said out loud and is not an anchor: an auditor holding a clean
/// report has to know that the check they asked for did not happen.
async fn anchor_from_witnesses(
    prefixes: &[String],
    keys: &[String],
    origin: &str,
) -> Result<Anchor, String> {
    let mut anchor = Anchor::default();
    if prefixes.is_empty() {
        if !keys.is_empty() {
            return Err("--witness-key was given with no --witness to use it against".to_owned());
        }
        return Ok(anchor);
    }
    let trusted = witness_keys(keys)?;
    if trusted.is_empty() {
        return Err(
            "--witness needs at least one --witness-key: a checkpoint fetched from a URL \
             nobody's signature covers is a stranger's claim presented as an independent \
             anchor"
                .to_owned(),
        );
    }

    let mut held: Vec<(String, agentplane::journal::Checkpoint)> = Vec::new();
    for prefix in prefixes {
        let reader = agentplane::journal::WitnessReader::new(prefix, trusted.clone())
            .map_err(|e| format!("--witness {prefix}: {e}"))?;
        match reader.latest(origin).await {
            Ok(Some(cosigned)) => {
                eprintln!(
                    "witness {prefix}: log '{}' at size {} with root {}, {} cosignature(s)",
                    cosigned.checkpoint.origin,
                    cosigned.checkpoint.size,
                    cosigned.checkpoint.root.to_hex(),
                    cosigned.cosignatures.len(),
                );
                // The highest wins: the append-only check asks *since when*,
                // so the later anchor is the stronger claim. The basis is
                // taken from the same answer, never assembled separately —
                // a basis describing a checkpoint other than the one used
                // would be the laundering this records exist to prevent.
                if anchor
                    .checkpoint
                    .as_ref()
                    .is_none_or(|b| b.size < cosigned.checkpoint.size)
                {
                    anchor.checkpoint = Some(cosigned.checkpoint.clone());
                    anchor.obtained_from = Some(format!("witness {prefix}"));
                    anchor.cosigned_by = cosigned
                        .cosignatures
                        .iter()
                        .map(|c| c.key_id.clone())
                        .collect();
                }
                held.push((prefix.clone(), cosigned.checkpoint));
            }
            // An answer, and one an auditor acts on: submission never reached
            // this witness, so the anchor they asked for does not exist.
            Ok(None) => {
                let said = format!(
                    "witness {prefix}: has never cosigned log '{origin}' — no anchor from \
                     this one, and nothing here is evidence about deletion"
                );
                eprintln!("{said}");
                anchor.unreached.push(said);
            }
            // Not fatal: one unreachable witness among several still leaves an
            // anchor, and failing the command over it would make an auditor's
            // check depend on every witness being up at once.
            Err(e) => {
                let said = format!("witness {prefix}: {e}");
                eprintln!("{said}");
                anchor.unreached.push(said);
            }
        }
    }

    // The rule lives in the library, where it can be tested and where an
    // embedder auditing with several witnesses gets it too. Said before the
    // report so it is not lost in the scroll — and **kept**, because a
    // finding delivered only to a terminal is a finding nobody files.
    for split in agentplane::journal::split_views(&held) {
        eprintln!("{split}");
        anchor.split_view.push(split.to_string());
    }
    Ok(anchor)
}

/// The audit half of `journal_verb`, printed and turned into an exit code.
///
/// Its own function because the export half and this one share only the run
/// list, and because assembling the evidence — a key, a saved checkpoint, an
/// anchor fetched from witnesses — is the part with rules in it.
async fn audit_report(
    store: &Arc<dyn JournalStore>,
    runs: &[agentplane::RunId],
    audit: &AuditArgs,
) -> Result<ExitCode, String> {
    // An audit with no prior checkpoint and no key still checks every
    // chain, and reports the two things it could not do. That is the
    // honest default for somebody who has just been handed a database —
    // and the flags are how they narrow it on the second pass, with the
    // key the operator published and the checkpoint the first pass printed.
    let verifier = verifier_from(&audit.key)?;
    // The store's own origin, unless the auditor named one — which they do
    // exactly when the store's answer is the thing under suspicion.
    let origin = match &audit.origin {
        Some(o) => o.clone(),
        None => store.checkpoint().await.map_err(|e| e.to_string())?.origin,
    };
    let fetched = anchor_from_witnesses(&audit.witness, &audit.witness_key, &origin).await?;
    let prior: Option<agentplane::journal::Checkpoint> = match &audit.prior {
        Some(path) => Some(
            std::fs::read_to_string(path)
                .map_err(|e| format!("reading --prior {path}: {e}"))
                .and_then(|text| {
                    serde_json::from_str(&text).map_err(|e| {
                        format!(
                            "--prior {path} is not a checkpoint — expected the `current` \
                             field of an earlier audit report: {e}"
                        )
                    })
                })?,
        ),
        None => None,
    };
    // A checkpoint from a file and one from a witness are the same kind of
    // evidence, and the higher one establishes more: the append-only check
    // is *since when*, so the later anchor is the stronger claim. Both
    // named is not a conflict to resolve — a witness holding a size the
    // saved checkpoint has passed is ordinary, and the reverse is a
    // finding the check below produces on its own.
    // A checkpoint read from a file wins only where no witness answered
    // higher, and the basis moves with it: whichever checkpoint is used, the
    // record says how *that* one was obtained. Reporting a witness's
    // cosignatures beside a file's checkpoint would be the exact laundering
    // this field exists to prevent.
    let mut anchor = fetched;
    match (prior, anchor.checkpoint.as_ref()) {
        (Some(saved), Some(found)) if found.size >= saved.size => {}
        (Some(saved), _) => {
            anchor.obtained_from = Some(match &audit.prior {
                Some(path) => format!("file {path}"),
                None => "file".to_owned(),
            });
            anchor.cosigned_by.clear();
            anchor.checkpoint = Some(saved);
        }
        (None, _) => {}
    }

    let evidence = agentplane::audit::Evidence {
        prior: anchor.checkpoint.as_ref(),
        verifier: verifier
            .as_ref()
            .map(|v| v as &dyn agentplane::core::Verifier),
        require_signatures: audit.require_signatures,
    };
    let report = agentplane::audit::audit(store, runs, &evidence)
        .await
        .map_err(|e| e.to_string())?;
    println!(
        "{}",
        serde_json::to_string_pretty(&AuditDocument {
            anchor: &anchor,
            report: &report,
        })
        .map_err(|e| e.to_string())?
    );
    // Findings are a failure; `not_checked` is not. An auditor who supplied
    // nothing gets a clean exit and a populated `not_checked`, and it is
    // their call whether that is enough.
    //
    // A split view fails too, and it is not the library's to report: two
    // witnesses holding one size with two different roots is the event
    // witnessing exists to detect, and only this command — which asked more
    // than one witness — is in a position to see it.
    Ok(if report.is_sound() && anchor.split_view.is_empty() {
        ExitCode::SUCCESS
    } else {
        ExitCode::FAILURE
    })
}

fn journal_verb(opts: &StoreArgs, audit: Option<&AuditArgs>) -> Result<ExitCode, String> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let backend = opts.at.open().await?;
        let store = backend.journal();
        // The same store holds the case layer, so the export always carries it.
        // An optional flag here would be a way to quietly produce the file the
        // verifier flags — the matters the journal names, missing.
        let cases = backend.cases();

        // The library's own list, not literals restated here: the store indexes
        // runs *by* outcome and has no "all runs" query, so an offline verb has
        // to name every outcome it wants — and a copy of that list in a binary
        // is where a new one would be silently dropped from exactly the
        // artifact an auditor asks for.
        let wanted: Vec<String> = if opts.outcome.is_empty() {
            agentplane::runtime::OUTCOMES_OF_RECORD
                .iter()
                .map(|s| (*s).to_owned())
                .collect()
        } else {
            opts.outcome.clone()
        };

        let mut runs = Vec::new();
        let mut truncated = Vec::new();
        for outcome in &wanted {
            let found = store
                .runs_by_outcome(outcome, opts.limit + 1)
                .await
                .map_err(|e| e.to_string())?;
            // One more than asked for, so a full page and an overflowing one are
            // distinguishable. An export that quietly stopped at the limit is
            // shaped exactly like a complete one.
            if found.len() > opts.limit {
                truncated.push(outcome.clone());
            }
            runs.extend(found.into_iter().take(opts.limit));
        }

        // The runs no outcome names. Skipped when the caller narrowed to
        // specific outcomes, because that is a request for exactly those
        // conclusions — and an in-flight run is not one.
        if opts.outcome.is_empty() {
            let flight = agentplane::export::runs_in_flight(&store, opts.limit)
                .await
                .map_err(|e| e.to_string())?;
            if flight.truncated {
                truncated.push("in-flight runs".to_owned());
            }
            for (run, why) in &flight.unreadable {
                eprintln!("warning: in-flight run {run} could not be read: {why}");
            }
            if !flight.runs.is_empty() {
                eprintln!(
                    "including {} run(s) still in flight — sleeping, awaiting a message, \
                     or waiting on a person. The Merkle log commits to sealed runs only, \
                     so these are carried and the checkpoint does not cover them",
                    flight.runs.len()
                );
            }
            runs.extend(flight.runs);
        }

        // Said on stderr so it survives `> out.jsonl`, and said before the work
        // rather than after: an operator who pipes this somewhere is not going
        // to re-read the tail.
        if !truncated.is_empty() {
            eprintln!(
                "warning: --limit {} was reached for: {}. This is a partial view; \
                 raise --limit or narrow --outcome",
                opts.limit,
                truncated.join(", ")
            );
        }

        let Some(audit) = audit else {
            let stdout = std::io::stdout();
            let trailer = agentplane::export::to_jsonl(
                &store,
                Some(&cases),
                &runs,
                std::io::BufWriter::new(stdout.lock()),
            )
            .await
            .map_err(|e| e.to_string())?;
            eprintln!(
                "exported {} record(s) from {}/{} run(s) and {} case(s)",
                trailer.records, trailer.runs_exported, trailer.runs_requested, trailer.cases
            );
            if !trailer.unreadable.is_empty() {
                for u in &trailer.unreadable {
                    eprintln!("unreadable: {} — {}", u.run, u.reason);
                }
                return Ok(ExitCode::FAILURE);
            }
            return Ok(ExitCode::SUCCESS);
        };

        audit_report(&store, &runs, audit).await
    })
}

/// The recovery rehearsal, from the command line.
///
/// The exit code carries the drill's load-bearing property: **only a loss
/// finding fails the command.** Erasure — tombstoned blobs, destroyed keys —
/// is retention working and lands in the report's counters, never in
/// `findings`, so a plane that erased everything it was asked to exits zero.
/// A drill whose exit code could not tell the two apart would teach whoever
/// scripts it to ignore the nonzero that means bytes are missing with no
/// tombstone to explain them.
fn drill_verb(opts: &DrillArgs) -> Result<ExitCode, String> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let backend = opts.at.open().await?;
        // The same store the other journal verbs open holds the case layer —
        // that is the wiring, whole. Blobs and keys are not in it, and the
        // report says so instead of this verb pretending otherwise.
        let cases = backend.cases();
        // The tenant scopes blob addresses and key scopes. This verb wires
        // neither store, so it reaches neither — but the value is the one the
        // operator named, so a future flag that wires blobs inherits the right
        // scope rather than a default that was never a decision.
        let tenant = opts
            .at
            .tenant
            .as_deref()
            .map(|name| agentplane::core::TenantId::new(name).map_err(|e| format!("--tenant: {e}")))
            .transpose()?
            .unwrap_or_default();
        let stores = agentplane::drill::Stores {
            cases: &cases,
            blobs: None,
            #[cfg(feature = "keyring")]
            keys: None,
            tenant: &tenant,
        };
        let report = agentplane::drill::drill(&stores)
            .await
            .map_err(|e| e.to_string())?;
        println!(
            "{}",
            serde_json::to_string_pretty(&report).map_err(|e| e.to_string())?
        );
        Ok(if report.is_sound() {
            ExitCode::SUCCESS
        } else {
            ExitCode::FAILURE
        })
    })
}

/// The instant a retention window of `days` reaches back to.
///
/// Checked, and the refusal is the point: subtracting a duration from an
/// instant **panics** in `time` when the result leaves the representable range,
/// so a typed-in day count large enough would abort the command rather than
/// explain itself. The two verbs that take such a window share this so they
/// cannot come to disagree about where the edge is.
fn cutoff_before(
    now: agentplane::core::Timestamp,
    days: u32,
) -> Result<agentplane::core::Timestamp, String> {
    now.checked_sub(time::Duration::days(i64::from(days)))
        .ok_or_else(|| {
            format!(
                "--older-than-days {days} reaches back past the first instant this \
                 runtime can name"
            )
        })
}

/// Retire admission keys past a window the operator chose.
///
/// Prints the count, because a retention pass that says nothing is
/// indistinguishable from one that found nothing — and the two call for
/// different responses when the index keeps growing.
fn forget_admissions_verb(opts: &ForgetArgs) -> Result<ExitCode, String> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let store = opts.at.open().await?.journal();
        // Wall clock by design, like the sweeper's: a retention window is a
        // question about how long ago something was claimed, not a journaled
        // observation of a run.
        #[allow(clippy::disallowed_methods)]
        let now = time::OffsetDateTime::now_utc();
        let cutoff = cutoff_before(now, opts.older_than_days)?;
        let retired = store
            .forget_admissions(cutoff)
            .await
            .map_err(|e| e.to_string())?;
        println!(
            "{}",
            serde_json::json!({
                "retired": retired,
                "older_than_days": opts.older_than_days,
                "cutoff": cutoff.unix_timestamp(),
            })
        );
        Ok(ExitCode::SUCCESS)
    })
}

/// Plan a retention pass, and refuse to pretend to run one.
///
/// This binary wires **no blob store and no key ring** — a redb file is a
/// journal and a case layer, not a bucket and not a KMS — so nothing here can
/// make a byte unreadable. A verb that walked the cases, erased nothing, and
/// printed `erased: 0` beside a clean exit code would be the shape this project
/// refuses hardest: a control that reads as having run. So the verb answers the
/// half it can — *what a pass would erase* — through the same selection rule
/// `Runtime::retain` uses, and refuses the other half by name.
fn retain_verb(opts: &RetainArgs) -> Result<ExitCode, String> {
    if !opts.dry_run {
        return Err(concat!(
            "this binary wires no blob store and no key ring, so it cannot erase anything; ",
            "it can only say what a pass would erase. Run again with --dry-run, or call ",
            "`Runtime::retain` from a plane built with `.blobs(..)` and `.keyring(..)`"
        )
        .to_owned());
    }
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let cases = opts.at.open().await?.cases();

        // Wall clock by design, like the sweeper's: a retention window is a
        // question about how long ago a matter opened, not a journaled
        // observation of a run.
        #[allow(clippy::disallowed_methods)]
        let now = time::OffsetDateTime::now_utc();
        let cutoff = cutoff_before(now, opts.older_than_days)?;
        let plan = agentplane::retention::plan(cases.as_ref(), cutoff)
            .await
            .map_err(|e| e.to_string())?;
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({
                "dry_run": true,
                "reason": opts.reason,
                "cutoff": cutoff.unix_timestamp(),
                "scanned": plan.scanned,
                "would_erase": plan.due,
            }))
            .map_err(|e| e.to_string())?
        );
        Ok(ExitCode::SUCCESS)
    })
}

/// Place, lift, or list legal holds.
///
/// Prints what it did for the reason `halt` does: an operator who cannot see
/// the control move has not been told whether they moved it.
fn hold_verb(opts: &HoldArgs) -> Result<ExitCode, String> {
    if opts.case.is_none() && (opts.lift || opts.reason.is_some()) {
        return Err(
            "--lift and --reason act on one matter: name it with --case, or pass \
             neither to list every hold standing"
                .to_owned(),
        );
    }
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let cases = opts.at.open().await?.cases();

        let Some(case) = opts.case.as_deref() else {
            let mut standing = Vec::new();
            let mut after = None;
            loop {
                let page = cases.holds(after, 256).await.map_err(|e| e.to_string())?;
                if page.is_empty() {
                    break;
                }
                after = page.last().map(|(c, _)| *c);
                for (id, hold) in page {
                    standing.push(serde_json::json!({
                        "case": id.to_string(),
                        "placed_at": hold.placed_at.unix_timestamp(),
                        "reason": hold.reason,
                        "by": hold.by.actor(),
                        "basis": hold.by.basis().as_str(),
                    }));
                }
            }
            println!(
                "{}",
                serde_json::to_string_pretty(&serde_json::json!({ "holds": standing }))
                    .map_err(|e| e.to_string())?
            );
            return Ok(ExitCode::SUCCESS);
        };

        let case = agentplane::core::CaseId::parse(case).map_err(|e| format!("--case: {e}"))?;
        if opts.lift {
            let lifted = cases.release_hold(case).await.map_err(|e| e.to_string())?;
            println!(
                "{}",
                serde_json::to_string_pretty(&serde_json::json!({
                    "case": case.to_string(),
                    "lifted": lifted,
                }))
                .map_err(|e| e.to_string())?
            );
            return Ok(ExitCode::SUCCESS);
        }

        let Some(reason) = opts.reason.as_deref() else {
            return Err(concat!(
                "--reason is required to place a hold: a preservation nobody can account for ",
                "is indistinguishable from a sweep that quietly stopped working. ",
                "Use --lift to release one"
            )
            .to_owned());
        };
        let Some(actor) = opts.actor.as_deref() else {
            return Err(concat!(
                "--actor is required to place a hold: a preservation order the runtime ",
                "cannot check is worth the name beside it, and the row records that this ",
                "one was asserted rather than authenticated"
            )
            .to_owned());
        };
        let by = agentplane::core::Operator::asserted(actor).map_err(|e| e.to_string())?;
        // Wall clock by design, like `retain`'s cutoff: when a hold was placed
        // is a fact about the outside world, not a journaled observation.
        #[allow(clippy::disallowed_methods)]
        let now = time::OffsetDateTime::now_utc();
        let placed = cases
            .place_hold(
                case,
                &agentplane::core::LegalHold {
                    placed_at: now,
                    reason: reason.to_owned(),
                    by,
                },
            )
            .await
            .map_err(|e| e.to_string())?;
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({
                "case": case.to_string(),
                "placed": placed,
                // False means one was already standing, and the first placement
                // is the one that counts — so say what is actually in force
                // rather than leaving the operator to assume it is theirs.
                "in_force": cases.hold(case).await.map_err(|e| e.to_string())?
                    .map(|h| serde_json::json!({
                        "placed_at": h.placed_at.unix_timestamp(),
                        "reason": h.reason,
                        "by": h.by.actor(),
                        "basis": h.by.basis().as_str(),
                    })),
            }))
            .map_err(|e| e.to_string())?
        );
        Ok(ExitCode::SUCCESS)
    })
}

/// Throw or lift one emergency stop.
///
/// Prints what it did, because an operator who cannot see the switch move has
/// not been told whether they threw it.
fn halt_verb(opts: &HaltArgs) -> Result<ExitCode, String> {
    let scope = agentplane::quota::HaltScope::parse(&opts.scope).ok_or_else(|| {
        format!(
            "'{}' is not a scope: use {}",
            opts.scope,
            agentplane::quota::HaltScope::forms()
        )
    })?;
    let thrown = if opts.lift {
        None
    } else {
        let reason = opts.reason.as_deref().ok_or(concat!(
            "--reason is required to halt: the next person to look will be somebody else, ",
            "possibly at three in the morning, and why is the whole question. ",
            "Use --lift to clear a halt"
        ))?;
        let actor = opts.actor.as_deref().ok_or(concat!(
            "--actor is required to halt: the runtime cannot check an emergency stop, ",
            "so the name beside it is the whole of its evidence. It is recorded as ",
            "asserted — nothing here verified it"
        ))?;
        let by = agentplane::core::Operator::asserted(actor).map_err(|e| e.to_string())?;
        Some((by, reason))
    };

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let quotas = opts.at.open().await?.quotas();
        let printed = if let Some((by, reason)) = &thrown {
            {
                // Wall clock by design, like `retain`'s cutoff and a hold's
                // instant: when a person threw a stop is a fact about the
                // outside world, not a journaled observation.
                #[allow(clippy::disallowed_methods)]
                let now = time::OffsetDateTime::now_utc();
                quotas
                    .set_halt(&scope, by, now, reason)
                    .await
                    .map_err(|e| e.to_string())?;
                serde_json::json!({
                    "scope": scope.key(),
                    "halted": true,
                    "reason": reason,
                    "by": by.actor(),
                    "basis": by.basis().as_str(),
                })
            }
        } else {
            {
                let was_standing = quotas.lift_halt(&scope).await.map_err(|e| e.to_string())?;
                // Whether one was standing is the answer to *did I clear the
                // right scope*, which is the question during an incident. A lift
                // that found nothing must not read as success.
                serde_json::json!({
                    "scope": scope.key(),
                    "halted": false,
                    "was_standing": was_standing,
                })
            }
        };
        println!("{printed}");
        Ok(ExitCode::SUCCESS)
    })
}

/// Every standing halt, so an operator can see what an incident left behind.
fn halts_verb(opts: &HaltsArgs) -> Result<ExitCode, String> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let quotas = opts.at.open().await?.quotas();
        let halts = quotas.halts().await.map_err(|e| e.to_string())?;
        let rows: Vec<serde_json::Value> = halts
            .iter()
            .map(|h| {
                serde_json::json!({
                    "scope": h.scope.key(),
                    "reason": h.reason,
                    "by": h.by.actor(),
                    "basis": h.by.basis().as_str(),
                    "thrown_at": h.at.unix_timestamp(),
                })
            })
            .collect();
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({ "halts": rows }))
                .map_err(|e| e.to_string())?
        );
        Ok(ExitCode::SUCCESS)
    })
}

/// Whichever backend `--store` names.
///
/// One flag rather than `--store` plus `--database-url`: two flags are mutually
/// exclusive in prose and simultaneously settable in fact, which is a refusal
/// somebody has to remember to write.
///
/// The choice is not cosmetic. redb admits a single writer **process**, so every
/// verb here is one a serving plane locks out; the shared store has no such
/// rule, which is what makes `halt` reachable during the incident it exists for.
enum Backend {
    /// A redb file. One writer process, so these verbs run between serving
    /// sessions rather than beside one.
    Embedded(Arc<RedbStore>, agentplane::core::TenantId),
    /// A shared `PostgreSQL` database. Several processes, so an operator verb
    /// and a serving plane coexist.
    #[cfg(feature = "postgres")]
    Shared(
        Arc<agentplane::store::PostgresStore>,
        agentplane::core::TenantId,
    ),
}

impl Backend {
    /// Open whichever backend `--store` names, scoped to `--tenant`.
    ///
    /// **Both arguments, always.** Every key in both backends leads with the
    /// tenant, so a cross-tenant read is a *miss* rather than a filtered row: a
    /// verb that opened a store without deciding whose would answer about the
    /// unnamed default and exit zero.
    // `async` in every configuration and awaiting nothing in one: connecting is
    // a network call, opening a file is not. One signature so no caller has to
    // know which feature set it was built with.
    #[allow(clippy::unused_async, clippy::unused_async_trait_impl)]
    async fn open(spec: &str, tenant: Option<&str>) -> Result<Self, String> {
        let tenant = tenant
            .map(|name| agentplane::core::TenantId::new(name).map_err(|e| format!("--tenant: {e}")))
            .transpose()?;
        if is_connection_string(spec) {
            // The refusal names the flag rather than saying "not a file": the
            // feature exists, it is one rebuild away, and "no such file or
            // directory" would send somebody to look at their path.
            #[cfg(not(feature = "postgres"))]
            return Err(
                "--store names a PostgreSQL database and this build cannot open one. \
                 Reinstall with `--features cli,postgres`, or use the `:full` container \
                 image, which is built with it"
                    .to_owned(),
            );
            #[cfg(feature = "postgres")]
            return Self::shared(spec, tenant).await;
        }
        let store = RedbStore::open(spec).map_err(|e| held_by_a_plane(&e.to_string()))?;
        let tenant = tenant.unwrap_or_default();
        Ok(Self::Embedded(
            Arc::new(store.for_tenant(tenant.clone())),
            tenant,
        ))
    }

    #[cfg(feature = "postgres")]
    async fn shared(url: &str, tenant: Option<agentplane::core::TenantId>) -> Result<Self, String> {
        let store = agentplane::store::PostgresStore::connect(url)
            .await
            .map_err(|e| e.to_string())?;
        let tenant = tenant.unwrap_or_default();
        Ok(Self::Shared(
            Arc::new(store.for_tenant(tenant.clone())),
            tenant,
        ))
    }

    /// The six stores a plane runs on.
    fn stores(&self) -> agentplane::runtime::Stores {
        match self {
            Self::Embedded(s, _) => agentplane::runtime::Stores::on(Arc::clone(s)),
            #[cfg(feature = "postgres")]
            Self::Shared(s, _) => agentplane::runtime::Stores::on(Arc::clone(s)),
        }
    }

    fn journal(&self) -> Arc<dyn JournalStore> {
        match self {
            Self::Embedded(s, _) => Arc::clone(s) as _,
            #[cfg(feature = "postgres")]
            Self::Shared(s, _) => Arc::clone(s) as _,
        }
    }

    fn cases(&self) -> Arc<dyn agentplane::case::CaseStore> {
        match self {
            Self::Embedded(s, _) => Arc::clone(s) as _,
            #[cfg(feature = "postgres")]
            Self::Shared(s, _) => Arc::clone(s) as _,
        }
    }

    fn quotas(&self) -> Arc<dyn agentplane::quota::QuotaStore> {
        match self {
            Self::Embedded(s, _) => Arc::clone(s) as _,
            #[cfg(feature = "postgres")]
            Self::Shared(s, _) => Arc::clone(s) as _,
        }
    }

    #[cfg(feature = "push")]
    fn push(&self) -> Arc<dyn agentplane::push::PushStore> {
        match self {
            Self::Embedded(s, _) => Arc::clone(s) as _,
            #[cfg(feature = "postgres")]
            Self::Shared(s, _) => Arc::clone(s) as _,
        }
    }

    /// Whose plane this store was opened as.
    ///
    /// The store is scoped by `for_tenant` and the runtime carries a tenant of
    /// its own, and **they are two halves of one decision**: a plane running as
    /// the default against a store scoped to `acme` writes runs into one
    /// keyspace while naming the other in every policy request and every
    /// erasure. `try_build` refuses that, which is how this was found — so the
    /// tenant travels with the backend rather than being passed twice.
    fn tenant(&self) -> agentplane::core::TenantId {
        match self {
            Self::Embedded(_, t) => t.clone(),
            #[cfg(feature = "postgres")]
            Self::Shared(_, t) => t.clone(),
        }
    }

    /// What this is, for a message an operator reads.
    #[cfg(all(feature = "a2a-server", feature = "cedar"))]
    const fn describe(&self) -> &'static str {
        match self {
            Self::Embedded(..) => "embedded redb file",
            #[cfg(feature = "postgres")]
            Self::Shared(..) => "shared PostgreSQL store",
        }
    }
}

/// Say what a locked embedded store means, in the words of the situation.
///
/// redb admits one writer **process**, so the ordinary way to meet this is to
/// run an operator verb against the file a `serve` is holding — which is to say,
/// during an incident, which is when the message matters most. *Database already
/// open* is true and tells nobody what to do about it.
fn held_by_a_plane(detail: &str) -> String {
    if !detail.contains("already open") {
        return detail.to_owned();
    }
    format!(
        "{detail}\n\nAn embedded redb store admits one writer process, and something \
         else is holding this one — most likely `agentplane serve`. Either stop that \
         process and run this again, or put the plane on a shared store \
         (`--store postgres://…`), where an operator verb and a serving plane \
         coexist. To act on a *running* embedded plane meanwhile, the operator API \
         is the surface that reaches it."
    )
}

/// Whether `--store` names a database rather than a file.
///
/// The two schemes `libpq` accepts, and nothing else. A prefix test rather than
/// a URL parse because the alternative treats every path containing `://` as a
/// connection string and every malformed URL as a filename.
fn is_connection_string(spec: &str) -> bool {
    spec.starts_with("postgres://") || spec.starts_with("postgresql://")
}

/// Read and validate the manifests, for every verb.
///
/// A manifest that does not validate is not a thing to run, digest, or reason
/// about — and in a multi-document file every document is held to that, because
/// deploying two thirds of a room is worse than deploying none of it.
fn manifests_at(path: &str) -> Result<Vec<Manifest>, String> {
    let text = std::fs::read_to_string(path).map_err(|e| format!("reading {path}: {e}"))?;
    Manifest::parse_all(&text).map_err(|e| e.to_string())
}

fn dispatch(cli: Cli) -> Result<ExitCode, String> {
    match cli.verb {
        Verb::Validate(a) => validate(&a),
        Verb::Schema => {
            // The parser stays authoritative: the schema is the format's
            // *shape*, and the semantic refusals run only in `validate`. The
            // document says so itself, so a copy pasted into a repo carries
            // the caveat along.
            let schema = Manifest::json_schema();
            println!(
                "{}",
                serde_json::to_string_pretty(&schema).expect("a generated schema serializes")
            );
            Ok(ExitCode::SUCCESS)
        }
        Verb::Digest(a) => {
            let manifests = manifests_at(&a.manifest)?;
            // One document prints the bare digest, so scripts that pin a single
            // agent keep working; a room prints one line per agent, because a
            // bundle digest would make one agent's edit move its neighbours'
            // identities.
            if let [only] = manifests.as_slice() {
                println!("{}", only.digest().map_err(|e| e.to_string())?.to_hex());
            } else {
                for m in &manifests {
                    println!(
                        "{}  {} {}",
                        m.digest().map_err(|e| e.to_string())?.to_hex(),
                        m.metadata.name,
                        m.metadata.version
                    );
                }
            }
            Ok(ExitCode::SUCCESS)
        }
        Verb::Audit(a) => journal_verb(&a.store, Some(&a)),
        Verb::Export(a) => journal_verb(&a, None),
        Verb::Drill(a) => drill_verb(&a),
        Verb::ForgetAdmissions(a) => forget_admissions_verb(&a),
        Verb::Retain(a) => retain_verb(&a),
        Verb::Halt(a) => halt_verb(&a),
        Verb::Hold(a) => hold_verb(&a),
        Verb::Halts(a) => halts_verb(&a),
        Verb::Restore(a) => {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .map_err(|e| format!("could not start the async runtime: {e}"))?;
            rt.block_on(async {
                let backend = a.at.open().await?;
                let store = backend.journal();
                let cases = backend.cases();
                let file =
                    std::fs::File::open(&a.file).map_err(|e| format!("reading {}: {e}", a.file))?;
                let report = agentplane::export::from_jsonl(
                    &store,
                    Some(&cases),
                    std::io::BufReader::new(file),
                )
                .await
                .map_err(|e| e.to_string())?;
                println!(
                    "{}",
                    serde_json::to_string_pretty(&report).map_err(|e| e.to_string())?
                );
                // The result is the comparison, not the loading. Equal roots at
                // equal size means every record, in every run, in the order the
                // log recorded them, rebuilt to the same commitment.
                Ok(if report.is_faithful() {
                    ExitCode::SUCCESS
                } else {
                    ExitCode::FAILURE
                })
            })
        }
        Verb::Verify(a) => verify_verb(&a),
        Verb::Run(a) => {
            let manifests = manifests_at(&a.manifest)?;
            execute(&manifests, &a)
        }
        Verb::Replay(a) => {
            let manifests = manifests_at(&a.manifest)?;
            replay(&manifests, &a)
        }
        Verb::Card(a) => card(&a),
        Verb::Serve(a) => {
            let manifests = manifests_at(&a.manifest)?;
            serve(&manifests, &a)
        }
    }
}

/// Read a checkpoint an auditor was handed, in either form they hold it in.
///
/// Two forms because two things produce one: `audit` prints JSON, and a
/// witness cosigns a `tlog-checkpoint` note. Requiring a conversion between
/// them would put a step between the auditor and the check, and the steps
/// between an auditor and a check are what this crate keeps removing.
fn read_checkpoint(path: &str) -> Result<agentplane::journal::Checkpoint, String> {
    let text =
        std::fs::read_to_string(path).map_err(|e| format!("reading --checkpoint {path}: {e}"))?;
    // The note first: it is the form that travels, and it is unambiguous —
    // JSON never parses as three newline-terminated lines.
    if let Ok(cp) = agentplane::journal::Checkpoint::from_note(&text) {
        return Ok(cp);
    }
    // A signed note carries the checkpoint as its body, so a reader who was
    // handed the cosigned artifact does not have to cut the signatures off.
    if let Ok(note) = agentplane::journal::SignedNote::parse(&text)
        && let Ok(cp) = agentplane::journal::Checkpoint::from_note(&note.text)
    {
        return Ok(cp);
    }
    serde_json::from_str(&text).map_err(|e| {
        format!(
            "--checkpoint {path} is neither a tlog-checkpoint note nor the `current` \
             field of an audit report: {e}"
        )
    })
}

/// `verify`: recompute an export from its own bytes, against a checkpoint from
/// somewhere else.
fn verify_verb(opts: &VerifyArgs) -> Result<ExitCode, String> {
    let verifier = verifier_from(&opts.key)?;
    let saved = match &opts.checkpoint {
        Some(path) => Some(read_checkpoint(path)?),
        None => None,
    };
    // The origin an export is *supposed* to be of. Read from the file would
    // defeat the purpose — the header is written by whoever wrote the file —
    // so `--origin` is required to fetch an anchor here, where `audit` can ask
    // its own store. `verify` runs against a file the auditor was handed, and
    // the name of the log is the one thing they have to know already.
    let fetched = match (&opts.origin, opts.witness.is_empty()) {
        (Some(origin), false) => {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .map_err(|e| format!("could not start the async runtime: {e}"))?;
            rt.block_on(anchor_from_witnesses(
                &opts.witness,
                &opts.witness_key,
                origin,
            ))?
        }
        (None, false) => {
            return Err(
                "--witness needs --origin: the log's name cannot come from the file being \
                 checked, because that header is written by whoever wrote the file"
                    .to_owned(),
            );
        }
        // No witness named. The key check still has to happen, so a
        // `--witness-key` with nothing to use it against is a refusal rather
        // than a flag that did nothing.
        (_, true) => {
            if !opts.witness_key.is_empty() {
                return Err(
                    "--witness-key was given with no --witness to use it against".to_owned(),
                );
            }
            Anchor::default()
        }
    };
    // The same rule `audit` applies, for the same reason: whichever checkpoint
    // is used, the record says how *that* one was obtained.
    let mut anchor = fetched;
    match (saved, anchor.checkpoint.as_ref()) {
        (Some(saved), Some(found)) if found.size >= saved.size => {}
        (Some(saved), _) => {
            anchor.obtained_from = Some(match &opts.checkpoint {
                Some(path) => format!("file {path}"),
                None => "file".to_owned(),
            });
            anchor.cosigned_by.clear();
            anchor.checkpoint = Some(saved);
        }
        (None, _) => {}
    }
    let expected = anchor.checkpoint.clone();
    let verifier = verifier
        .as_ref()
        .map(|v| v as &dyn agentplane::core::Verifier);
    let report = if opts.file == "-" {
        agentplane::export::verify(std::io::stdin().lock(), verifier, expected.as_ref())
            .map_err(|e| e.to_string())
    } else {
        let file =
            std::fs::File::open(&opts.file).map_err(|e| format!("reading {}: {e}", opts.file))?;
        agentplane::export::verify(std::io::BufReader::new(file), verifier, expected.as_ref())
            .map_err(|e| e.to_string())
    }?;
    println!(
        "{}",
        serde_json::to_string_pretty(&VerifyDocument {
            anchor: &anchor,
            report: &report,
        })
        .map_err(|e| e.to_string())?
    );
    // Findings fail; `not_checked` does not. A pass with no key — or with no
    // checkpoint — has established less, and saying so is different from
    // failing.
    // A split view fails here too. Only a caller that asked more than one
    // witness can see it, so neither the library's report nor the file can.
    Ok(if report.is_sound() && anchor.split_view.is_empty() {
        ExitCode::SUCCESS
    } else {
        ExitCode::FAILURE
    })
}

/// Print the Agent Card a served manifest would advertise.
fn card(opts: &CardArgs) -> Result<ExitCode, String> {
    // One card, one agent — the same rule `serve` applies, because this verb
    // prints exactly what `serve` would advertise.
    let manifests = manifests_at(&opts.manifest)?;
    let [manifest] = manifests.as_slice() else {
        return Err(format!(
            "`card` describes one agent and this file holds {}. A2A's card \
             path is well-known and singular — split the file, or point this \
             at the document you would serve",
            manifests.len()
        ));
    };
    let card =
        agentplane::peers::AgentCard::derive(manifest, &opts.url).map_err(|e| e.to_string())?;
    println!(
        "{}",
        serde_json::to_string_pretty(&card).map_err(|e| e.to_string())?
    );
    Ok(ExitCode::SUCCESS)
}

fn main() -> ExitCode {
    install_tracing();
    // `clap` prints its own diagnostics and exits; everything past the parse is
    // this binary's own vocabulary.
    let cli = <Cli as clap::Parser>::parse();
    match dispatch(cli) {
        Ok(code) => code,
        Err(e) => {
            eprintln!("agentplane: {e}");
            ExitCode::FAILURE
        }
    }
}

impl RunArgs {
    fn read_input(&self) -> Result<serde_json::Value, String> {
        let text = match (&self.input, &self.input_file) {
            // `-` is stdin, the convention every pipe-shaped tool honours and
            // `verify` here already does. It belongs to `--input`, not
            // `--input-file`: a file literally named `-` is reachable as
            // `./-`, and a pipe is not reachable any other way.
            (Some(s), _) if s == "-" => {
                use std::io::Read as _;
                let mut text = String::new();
                std::io::stdin()
                    .read_to_string(&mut text)
                    .map_err(|e| format!("reading standard input: {e}"))?;
                text
            }
            (Some(s), _) => s.clone(),
            (_, Some(p)) => std::fs::read_to_string(p).map_err(|e| format!("reading {p}: {e}"))?,
            _ => "{}".into(),
        };
        serde_json::from_str(&text).map_err(|e| format!("the input is not valid JSON: {e}"))
    }
}

/// How many due webhook registrations one push tick delivers.
///
/// Bounded, and the report says when it came back full — a worker still draining
/// a backlog is one not delivering the next notification, and a capped result
/// shaped like a complete one is the silent-truncation shape.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
const PUSH_BATCH: usize = 64;

/// How often a served plane sweeps, when nobody says otherwise.
///
/// Short enough that a breached deadline is noticed in the same minute, long
/// enough that an idle plane is not doing constant store reads.
/// `--sweep-every 0` turns it off, for a deployment running the sweep from its
/// own scheduler.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
const DEFAULT_SWEEP_SECONDS: u32 = 30;

/// Host this agent as an A2A 1.0 peer.
///
/// The A2A server, the Agent Card and the conformance work behind them all
/// existed already and could only be reached by writing Rust — which is the one
/// thing the declarative tier exists to remove. A manifest that can be *run*
/// from a file but not *hosted* from one leaves the interoperability half of
/// this crate behind a language barrier.
///
/// # Everything here fails closed, and each refusal says why
///
/// Both `--policy` and `--tokens` are required with no default. That is the
/// whole design and not an inconvenience to be smoothed away later: a permissive
/// engine and no engine are the same behaviour, and a server that authenticates
/// nobody has no actor to record a decision against. `A2aServer::new` already
/// refuses a runtime with no policy engine and no case layer; this wires both
/// rather than working around either.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn serve(manifests: &[Manifest], opts: &ServeArgs) -> Result<ExitCode, String> {
    use agentplane::api::a2a::A2aServer;
    use agentplane::api::tokens::TokenAuthenticator;

    // One card, one agent. A room is several manifests and A2A's well-known
    // card path is singular, so serving a bundle would have to pick one and
    // silently not serve the others.
    let [manifest] = manifests else {
        return Err(format!(
            "`serve` hosts one agent and this file holds {}. A2A's card path is \
             well-known and singular, so a room would have to advertise one \
             document and quietly not serve the rest — split the file, or run \
             one process per agent",
            manifests.len()
        ));
    };

    let url = opts.url.as_deref().ok_or(
        "`serve` needs --url: the address callers reach this plane on. It goes on the \
         Agent Card, so it is the public URL rather than what you bind — an agent's \
         declaration must not change when its address does",
    )?;
    let policy_path = opts.policy.as_deref().ok_or(
        "`serve` needs --policy: a Cedar policy set. There is deliberately no default — \
         a permissive engine and no engine are the same behaviour, and only one of them \
         looks governed",
    )?;
    let tokens_path = opts.tokens.as_deref().ok_or(
        "`serve` needs --tokens: bearer tokens naming the callers this plane accepts. \
         There is deliberately no default — a server that authenticates nobody has no \
         actor to record a decision against",
    )?;

    let policy_src = std::fs::read_to_string(policy_path)
        .map_err(|e| format!("reading the policy set {policy_path}: {e}"))?;
    let policy = agentplane::policy::CedarEngine::new(&policy_src)
        .map_err(|e| format!("the policy set {policy_path} was refused: {e}"))?;
    let tokens_src = std::fs::read_to_string(tokens_path)
        .map_err(|e| format!("reading the token file {tokens_path}: {e}"))?;
    // One `Arc`, two surfaces: the same accepted credentials govern both, so a
    // token added for a peer is not silently also an operator credential —
    // that separation is policy's job, on `a2a:*` versus `api:*` actions.
    let auth: Arc<dyn agentplane::api::Authenticator> = Arc::new(
        TokenAuthenticator::from_yaml(&tokens_src)
            .map_err(|e| format!("the token file {tokens_path} was refused: {e}"))?,
    );
    let operator_auth = Arc::clone(&auth);

    // Multi-threaded here and current-thread in `execute`, because these are
    // different programs wearing one binary: a run does one agent's work and
    // exits, a server takes concurrent requests for as long as it is up.
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async move {
        // A journal in memory would make every served task disappear on
        // restart, which is the opposite of what a peer promises when it hands
        // back a task id. Refused rather than defaulted.
        let backend = opts.at.open().await?.ok_or(
            "`serve` needs --store: a served task's id is a promise that it can be \
             fetched again, and an in-memory journal breaks that promise at the next \
             restart. `run` may journal to memory because it exits with its answer",
        )?;

        // **The whole plane, not a corner of it.** One store backs every store
        // this runtime has, and a server that wired only the journal and the
        // case layer would accept an agent that waits, sleeps or opens a human
        // task and then never make progress on any of them — a suspended run is
        // a row, and something has to come back for it.
        let mut builder = with_providers(
            Runtime::builder_with(backend.stores()).tenant(backend.tenant()),
            std::slice::from_ref(manifest),
        )
        .await?;
        for (name, client) in connect_mcp_servers(&opts.mcp, std::slice::from_ref(manifest)).await?
        {
            builder = builder.tool_server(name, client);
        }
        if let Some((registry, client)) = connect_peers(&opts.peer, std::slice::from_ref(manifest))?
        {
            builder = builder.peers(registry, client);
        }
        builder = builder
            .policy(Arc::new(policy) as Arc<dyn agentplane::core::PolicyEngine>)
            .agent(agentplane::runtime::Agent::new(manifest));
        // The same handle `wire_push` gives the A2A server below. The plane
        // holds it because the registrations that stop being delivered are a
        // backlog, and the operator surface is where a backlog is answered —
        // the delivery worker that parked one has nothing more to say about it.
        if !opts.push_host.is_empty() {
            builder = builder.push(backend.push());
        }
        let runtime = builder.try_build().map_err(|e| e.to_string())?;

        let security = agentplane::peers::CardSecurity::bearer("bearer", Vec::<String>::new());
        let mut server = A2aServer::new(Arc::clone(&runtime), auth, &security, manifest, url)
            .map_err(|e| e.to_string())?;

        server = wire_push(server, &opts.push_host, &backend)?;
        serve_until_stopped(
            &runtime,
            server,
            operator_auth,
            opts,
            manifest,
            url,
            &backend,
        )
        .await?;
        Ok(ExitCode::SUCCESS)
    })
}

/// Serve until a supervisor says stop, then stop everything this process owns.
///
/// Split from `serve` because wiring a plane and running one are different jobs
/// with different failure modes, and because the shutdown order below is the
/// part worth reading on its own.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn serve_until_stopped(
    runtime: &Arc<Runtime>,
    server: agentplane::api::a2a::A2aServer,
    operator_auth: Arc<dyn agentplane::api::Authenticator>,
    opts: &ServeArgs,
    manifest: &Manifest,
    url: &str,
    backend: &Backend,
) -> Result<(), String> {
    let addr = opts.addr.as_str();
    // One signal, every listener. Raised by the signal watcher below, and
    // by `serve` itself returning for any other reason — a bind that dies
    // under a running plane must not leave the periodic passes sweeping a
    // store nothing is serving from.
    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
    let mut background: Vec<Task> = Vec::new();
    if let Some(worker) = server.push_worker() {
        background.extend(spawn_push_worker(
            worker,
            opts.sweep_every.unwrap_or(DEFAULT_SWEEP_SECONDS),
            stop_rx.clone(),
        ));
    }

    background.extend(spawn_sweeper(
        runtime,
        opts.sweep_every.unwrap_or(DEFAULT_SWEEP_SECONDS),
        stop_rx.clone(),
    ));
    background.extend(spawn_drill(
        runtime,
        opts.drill_every.unwrap_or(0),
        stop_rx.clone(),
    ));

    if let Some(operator_addr) = opts.operator_addr.as_deref() {
        background.push(
            spawn_operator_surface(runtime, operator_auth, operator_addr, stop_rx.clone()).await?,
        );
    }

    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .map_err(|e| format!("could not bind {addr}: {e}"))?;
    // stderr, so the answer stream stays clean for whatever pipes this.
    eprintln!(
        "serving {} {} on {addr} as {url}",
        manifest.metadata.name, manifest.metadata.version
    );
    eprintln!("  card: {url}/.well-known/agent-card.json");
    // Which backend, never the connection string: a `postgres://` URL carries a
    // password, and a startup banner is the most-copied text a deployment has.
    // It matters to an operator because it decides whether the CLI verbs work
    // beside this process or only after it stops.
    eprintln!("  store: {}", backend.describe());
    eprintln!("  stop: SIGTERM drains for up to {}s", opts.drain_secs);

    let mut peer = tokio::spawn(async move {
        axum::serve(listener, server.router())
            .with_graceful_shutdown(stopping(stop_rx))
            .await
    });

    // Whichever comes first. A server that ends on its own — an accept loop that
    // died — must still stop the rest of this process, or the periodic passes go
    // on sweeping a store nothing is serving from.
    let mut ended = tokio::select! {
        () = stop_requested() => None,
        joined = &mut peer => Some(joined),
    };
    let still_serving = ended.is_none();
    let _ = stop_tx.send(true);
    let grace = std::time::Duration::from_secs(opts.drain_secs);
    let deadline = tokio::time::Instant::now() + grace;

    // **Concurrently, and that is the design.** The two waits are for different
    // things — the connections still being served, and the runs this process put
    // on a task of its own — and `Runtime::drain` closes admission on its first
    // poll, which is what lets an open subscription end rather than hold the
    // graceful shutdown open for as long as the run it is watching.
    let (report, closed) = tokio::join!(
        runtime.drain(grace),
        stop_serving(&mut peer, background, still_serving, deadline),
    );
    ended = ended.or(closed);
    report_drain(&report);
    match ended {
        Some(Ok(listening)) => listening.map_err(|e| format!("the server stopped: {e}")),
        Some(Err(e)) => Err(format!("the server task failed: {e}")),
        None => Ok(()),
    }
}

/// Wait out the listeners and the periodic passes, bounded by one deadline.
///
/// Everything here is abandoned rather than waited for once the deadline
/// passes. A periodic pass holds no lease of its own and every write in it is
/// idempotent, so the next instance's first tick repeats whatever it did not
/// finish; a connection still open is a client that reconnects. What must not be
/// abandoned early is a *run*, and that is the other half of the join this is
/// called from.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn stop_serving(
    peer: &mut tokio::task::JoinHandle<std::io::Result<()>>,
    background: Vec<Task>,
    still_serving: bool,
    deadline: tokio::time::Instant,
) -> Option<Result<std::io::Result<()>, tokio::task::JoinError>> {
    let mut ended = None;
    if still_serving {
        match tokio::time::timeout_at(deadline, peer).await {
            Ok(joined) => ended = Some(joined),
            Err(_) => eprintln!("  stop: connections were still open at the grace period"),
        }
    }
    for task in background {
        if tokio::time::timeout_at(deadline, task).await.is_err() {
            break;
        }
    }
    ended
}

/// Say what the stop cost, and to whom.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn report_drain(report: &agentplane::runtime::DrainReport) {
    if report.is_complete() {
        eprintln!(
            "stopped; {} background runs finished first",
            report.settled()
        );
        return;
    }
    // Named, at `warn`, and on stderr: these are runs whose leases this process
    // is about to stop renewing, so each expires unreleased and the next
    // instance's recovery sweep takes it over — the same path a crash takes.
    // Nothing is lost, and the number is what an operator sizing `--drain-secs`
    // against their supervisor's grace period has to see.
    let unfinished: Vec<String> = report.unfinished.iter().map(ToString::to_string).collect();
    tracing::warn!(
        settled = report.settled(),
        unfinished = ?unfinished,
        "the grace period ended with runs still executing; they are left for the recovery sweep"
    );
    eprintln!(
        "stopped; {} background runs finished, {} left to the recovery sweep: {}",
        report.settled(),
        unfinished.len(),
        unfinished.join(" ")
    );
}

/// A stop signal, broadcast to everything this process started.
///
/// `watch` rather than a one-shot because there are several listeners and none
/// of them owns the signal: two HTTP servers and up to three periodic passes all
/// have to hear the same thing, and a channel that only one can take would make
/// the order they were started in decide which ones stop.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
type Stop = tokio::sync::watch::Receiver<bool>;

/// A background pass this process must see the end of before it exits.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
type Task = tokio::task::JoinHandle<()>;

/// Wait for the next tick, or for the stop signal. `false` means stop.
///
/// The check is here and not around the work, so a pass that is *mid-tick* when
/// the signal arrives finishes that tick and stops before the next one. A sweep
/// abandoned halfway is not a fault — every write in it is idempotent and the
/// next instance repeats it — but finishing costs milliseconds and leaves less
/// for somebody else to redo.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn next_tick(tick: &mut tokio::time::Interval, stop: &mut Stop) -> bool {
    tokio::select! {
        _ = tick.tick() => true,
        _ = stop.changed() => false,
    }
}

/// Resolve when the stop signal is raised, or when the last sender is dropped.
///
/// A dropped sender is treated as a stop rather than as a hang: the sender lives
/// in `serve`, so its absence means `serve` has already returned.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn stopping(mut stop: Stop) {
    let _ = stop.changed().await;
}

/// The signals a supervisor stops a process with.
///
/// `SIGTERM` is what Kubernetes, Docker and systemd send; `SIGINT` is a person
/// at a terminal. They mean the same thing here and are handled the same way —
/// a second one is *not* special-cased into an immediate exit, because the whole
/// point of the drain is that the grace period belongs to the supervisor, which
/// already holds a `SIGKILL` for a process that overstays it.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn stop_requested() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{SignalKind, signal};
        let term = match signal(SignalKind::terminate()) {
            Ok(s) => Some(s),
            // A process that cannot install the handler must not silently become
            // one that ignores the signal. Said out loud, and `SIGTERM` keeps
            // its default disposition — which kills this process without a
            // drain, exactly as it would have before the handler was attempted.
            Err(error) => {
                tracing::error!(%error, "could not listen for SIGTERM; this process will not drain");
                None
            }
        };
        let terminated = async move {
            match term {
                Some(mut term) => {
                    term.recv().await;
                }
                None => std::future::pending().await,
            }
        };
        tokio::select! {
            () = terminated => {}
            r = tokio::signal::ctrl_c() => {
                if let Err(error) = r {
                    tracing::error!(%error, "could not listen for SIGINT");
                }
            }
        }
    }
    #[cfg(not(unix))]
    {
        if let Err(error) = tokio::signal::ctrl_c().await {
            tracing::error!(%error, "could not listen for an interrupt");
        }
    }
}

/// Rehearse recovery on a timer, because a control nobody exercises is one an
/// audit cannot count.
///
/// The three-way verdict — *intact*, *erased by design*, *lost* — is the whole
/// value: an erasure that worked and a byte that went missing look identical
/// to a store, and telling them apart is what a rehearsal establishes.
///
/// A finding is an `error` with the report attached, not a panic: a drill
/// reports on the past and stopping a serving plane because yesterday's bytes
/// are gone helps nobody. `not_checked` is logged whenever it is non-empty,
/// because the difference between *sound* and *nothing I looked at was wrong*
/// is exactly that list — the same reason the report carries it at all.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn spawn_drill(runtime: &Arc<Runtime>, every: u32, stop: Stop) -> Option<Task> {
    if every == 0 {
        return None;
    }
    if runtime.cases().is_none() {
        eprintln!(
            "  drill: --drill-every was given but this plane has no case store, \
             so there are no cases to walk"
        );
        return None;
    }
    let plane = Arc::clone(runtime);
    Some(tokio::spawn(async move {
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(u64::from(every)));
        let mut stop = stop;
        loop {
            if !next_tick(&mut tick, &mut stop).await {
                break;
            }
            match plane.drill().await {
                Ok(report) if !report.is_sound() => {
                    tracing::error!(?report, "the recovery drill found unrecoverable references");
                }
                Ok(report) if !report.not_checked.is_empty() => {
                    tracing::warn!(
                        ?report,
                        "the recovery drill passed, but could not check everything"
                    );
                }
                Ok(report) => tracing::info!(cases = report.cases, "the recovery drill passed"),
                Err(error) => tracing::error!(%error, "the recovery drill could not run"),
            }
        }
    }))
}

/// Sweep on a clock, because nothing else will.
///
/// Deadlines warn and breach, tasks expire, dead letters accumulate, and a run
/// suspended on `cx.sleep` or a correlated event is a **row** waiting for a
/// sweep — not a task waiting on a timer. Without this a served plane accepts
/// all of that and silently never progresses any of it, which is worse than
/// refusing it: the agent looks like it is working.
///
/// `sweep` is idempotent by contract, so a tick overlapping the last one, or a
/// second instance sweeping the same store, is safe. `0` turns it off for a
/// deployment driving the sweep from its own scheduler.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn spawn_sweeper(runtime: &Arc<Runtime>, every: u32, stop: Stop) -> Option<Task> {
    if every == 0 {
        return None;
    }
    let sweeper = Arc::clone(runtime);
    Some(tokio::spawn(async move {
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(u64::from(every)));
        let mut stop = stop;
        loop {
            if !next_tick(&mut tick, &mut stop).await {
                break;
            }
            // The sweeper's clock is the wall clock by design: it decides *when*
            // an obligation is late, which is not a journaled observation of a
            // run. Every transition it makes is journaled by the sweep's own
            // sealed run.
            #[allow(clippy::disallowed_methods)]
            let now = time::OffsetDateTime::now_utc();
            match sweeper.fire_timers(now).await {
                Ok(w) if w.failed > 0 => {
                    tracing::warn!(fired = w.fired, failed = w.failed, "timer wakes failed");
                }
                Ok(w) if w.fired > 0 => tracing::info!(fired = w.fired, "timers fired"),
                Ok(_) => {}
                Err(error) => tracing::error!(%error, "firing timers failed"),
            }
            match sweeper
                .sweep(now, std::time::Duration::from_secs(3600))
                .await
            {
                // A sweep that decided something, hit its cap, or lost its own
                // evidence is a finding an operator must clear rather than a
                // line in a log — I13 applies to the sweeper's own report.
                Ok(report) if report.needs_attention() => {
                    tracing::warn!(?report, "the sweep needs attention");
                }
                Ok(report) if !report.is_quiet() => tracing::info!(?report, "swept"),
                Ok(_) => {}
                Err(error) => tracing::error!(%error, "the sweep failed"),
            }
        }
    }))
}

/// The operator surface, on its **own listener**.
///
/// Off unless asked for, and deliberately not the peer's port. Sharing it would
/// put the worklist, task decisions and `GET /runs?outcome=quarantined` behind
/// the public address an A2A peer is handed — one policy mistake away from a
/// peer reading every run on the plane. A separate binding lets an operator keep
/// this on loopback or a private interface while the card stays public.
///
/// The real separation is **policy**, not the port: both surfaces authenticate
/// against the same token file, and a peer token permitted only `a2a:*` is
/// refused `api:run.list` even when it reaches this socket. The port is defence
/// in depth.
///
/// # Errors
///
/// If the plane has no policy engine, or the address cannot be bound.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
async fn spawn_operator_surface(
    runtime: &Arc<Runtime>,
    auth: Arc<dyn agentplane::api::Authenticator>,
    addr: &str,
    stop: Stop,
) -> Result<Task, String> {
    let api = agentplane::api::Api::new(Arc::clone(runtime), auth).map_err(|e| e.to_string())?;
    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .map_err(|e| format!("could not bind the operator surface {addr}: {e}"))?;
    eprintln!("  operator: http://{addr}/runs?outcome=failed");
    Ok(tokio::spawn(async move {
        let served = axum::serve(listener, api.router())
            .with_graceful_shutdown(stopping(stop))
            .await;
        if let Err(error) = served {
            tracing::error!(%error, "the operator surface stopped");
        }
    }))
}

/// A peer registry and the transport that reaches every peer in it.
type WiredPeers = (
    agentplane::peers::PeerRegistry,
    Arc<dyn agentplane::peers::PeerClient>,
);

/// Wire the A2A peers `--peer` names, granted what the manifests grant them.
///
/// The registry scope for a peer is exactly the set of capabilities some
/// manifest grants under `tool://<name>/…`: the reviewed documents are what
/// say what this plane may ask a peer for, and a peer nothing grants is a
/// flag naming nobody. The token rides in the environment, because a command
/// line is visible to every process on the host.
#[cfg(feature = "a2a")]
fn connect_peers(specs: &[String], manifests: &[Manifest]) -> Result<Option<WiredPeers>, String> {
    use agentplane::core::Scope;
    use agentplane::peers::a2a::{A2aClient, Endpoint};
    use agentplane::peers::{PeerCredential, PeerGrant, PeerId, PeerRegistry, PeerRouter};

    if specs.is_empty() {
        return Ok(None);
    }
    let mut registry = PeerRegistry::new();
    let mut router = PeerRouter::new();
    for spec in specs {
        let (name, url) = spec.split_once('=').ok_or_else(|| {
            format!(
                "--peer wants `<name>=<url>`, got `{spec}`. The name is the one your \
                 manifest's grants use: a grant `tool://reviewer/audit.check` needs \
                 `--peer reviewer=https://...`"
            )
        })?;
        if name.trim().is_empty() || url.trim().is_empty() {
            return Err(format!("--peer `{spec}` names no peer or no URL"));
        }
        let granted: Vec<String> = manifests
            .iter()
            .flat_map(|m| &m.spec.tools)
            .filter_map(|g| agentplane::tools::ToolId::parse(&g.reference))
            .filter(|id| id.server == name)
            .map(|id| id.tool)
            .collect();
        if granted.is_empty() {
            return Err(format!(
                "--peer `{name}` is wired but no manifest grants a `tool://{name}/…` \
                 capability, so nothing could ever call it"
            ));
        }
        let peer = PeerId::new(name);
        let mut grant = PeerGrant::new(Scope::of(granted.iter().cloned()));
        let token_var = format!(
            "AGENTPLANE_PEER_TOKEN_{}",
            name.to_ascii_uppercase().replace(['.', '-'], "_")
        );
        if let Ok(token) = std::env::var(&token_var)
            && !token.is_empty()
        {
            grant = grant.with_credential(&peer, PeerCredential::for_audience(peer.clone(), token));
        }
        registry = registry.allow(peer.clone(), grant);
        // Refused here rather than at the first call, because the two failures
        // read nothing alike: a plaintext peer wired at boot fails much later,
        // once, inside whichever run happened to reach it, as a peer refusal
        // that names no cause a person can act on.
        //
        // This build cannot lift it. The exception for `http://` to this machine
        // is `testkit`, which the released binary does not carry — see the `cli`
        // feature — and reaching a local peer from a development build means
        // `--features cli,a2a,testkit`.
        if !url.starts_with("https://") {
            return Err(format!(
                "--peer `{name}` is `{url}`, and a peer is reached over HTTPS. A card or a \
                 peer answer steers the calls that follow it, so a plaintext hop would let \
                 the network choose them. For a peer on this machine, build with \
                 `--features cli,a2a,testkit`."
            ));
        }
        let client = A2aClient::new(Endpoint::new(url))
            .map_err(|e| format!("could not build a client for peer `{name}`: {e}"))?;
        router = router.peer(
            peer,
            Arc::new(client) as Arc<dyn agentplane::peers::PeerClient>,
        );
        eprintln!("  peer: {name} <- {url} ({})", granted.join(", "));
    }
    Ok(Some((
        registry,
        Arc::new(router) as Arc<dyn agentplane::peers::PeerClient>,
    )))
}

/// Parse every manifest in the file, and hold each to the annotations a
/// deployment requires.
///
/// The runtime never reads `metadata.annotations` — that is what makes them safe
/// to carry, and it is why nothing in a running plane can notice an agent that
/// shipped without an owner. This does not change that: the keys stay the
/// deployment's vocabulary, no interpretation crosses the trust boundary, and
/// the enforcement is a job somebody runs in review.
///
/// # Errors
///
/// If a manifest will not parse, or a required annotation is absent.
fn validate(a: &ValidateArgs) -> Result<ExitCode, String> {
    let mut missing = Vec::new();
    for m in &manifests_at(&a.manifest)? {
        // Checked per agent, because a file may hold a room and "one of them
        // has an owner" is not the rule anybody meant.
        //
        // Presence only: an annotation *present and empty* never reaches here,
        // because the parser refuses it — a key that answers nothing reads to a
        // reviewer like a question that was answered.
        let absent: Vec<&String> = a
            .require_annotation
            .iter()
            .filter(|key| !m.metadata.annotations.contains_key(*key))
            .collect();
        if absent.is_empty() {
            println!("ok: {} {}", m.metadata.name, m.metadata.version);
        } else {
            for key in absent {
                println!("MISSING: {} — annotation '{key}'", m.metadata.name);
                missing.push(format!("{}: {key}", m.metadata.name));
            }
        }
    }
    if missing.is_empty() {
        return Ok(ExitCode::SUCCESS);
    }
    Err(format!(
        "{} required annotation(s) absent: {}",
        missing.len(),
        missing.join(", ")
    ))
}

/// Naming the feature rather than ignoring the flag, as `--mcp` does.
#[cfg(not(feature = "a2a"))]
fn connect_peers(specs: &[String], _manifests: &[Manifest]) -> Result<Option<WiredPeers>, String> {
    if specs.is_empty() {
        return Ok(None);
    }
    Err(
        "this build cannot call an A2A peer: `--peer` needs the `a2a` feature. Reinstall \
         with `--features cli,a2a`, or use the `:full` container image"
            .to_owned(),
    )
}

/// Connect the MCP servers named on the command line.
///
/// # Why the command line and not the manifest
///
/// The manifest grants `tool://tickets/read`; **which transport reaches
/// `tickets`** is deployment wiring, exactly as a model's base URL and an API
/// key are. Putting it in the reviewed file would mean an agent's declaration —
/// and therefore its digest — changed when it moved between a laptop and a
/// cluster, and the whole point of the digest is that it does not.
///
/// # The trust boundary, stated
///
/// This **executes a command**. That is not an escalation over what the caller
/// already had: the operator typed it on the same command line as the manifest
/// path, the policy set and the token file, and anyone who can choose this
/// process's arguments can run their own process instead. It is emphatically
/// *not* a capability the manifest, a model, or an A2A peer can reach — nothing
/// in a run's data path chooses a server, only the operator's argv does.
///
/// The command is split on whitespace, which covers `npx -y @scope/server` and
/// `python server.py` and stops short of a shell: no globbing, no pipelines, no
/// `$(...)`. A path containing spaces needs a wrapper script, and that is the
/// right trade for not embedding a shell in a governed runtime.
#[cfg(feature = "mcp-stdio")]
async fn connect_mcp_servers(
    specs: &[String],
    manifests: &[Manifest],
) -> Result<Vec<(String, Arc<dyn agentplane::tools::ToolClient>)>, String> {
    let mut wired = Vec::with_capacity(specs.len());
    for spec in specs {
        let (name, command) = spec.split_once('=').ok_or_else(|| {
            format!(
                "--mcp wants `<server>=<command>`, got `{spec}`. The server name is the \
                 one your manifest's grants use: a grant `tool://tickets/read` needs \
                 `--mcp tickets=...`"
            )
        })?;
        if name.trim().is_empty() {
            return Err(format!("--mcp `{spec}` names no server"));
        }
        let mut parts = command.split_whitespace();
        let program = parts
            .next()
            .ok_or_else(|| format!("--mcp `{spec}` names server `{name}` but no command"))?;
        let mut process = tokio::process::Command::new(program);
        process.args(parts);
        let transport = rmcp::transport::TokioChildProcess::new(process)
            .map_err(|e| format!("could not start the MCP server `{name}` (`{command}`): {e}"))?;
        // A child process over stdio: this plane opens no socket for it, so
        // there is no host for an egress allowlist to judge. That is not a
        // claim about what the server itself reaches — the same residual a
        // compromised allowlisted endpoint carries — and it is the honest
        // answer to the only question an allowlist can decide.
        let client = agentplane::tools::McpClient::connect(
            name,
            transport,
            agentplane::tools::Destination::Local,
        )
        .await
        .map_err(|e| format!("the MCP server `{name}` cannot be used: {e}"))?;
        // The negotiated version, not the offered one. MCP negotiation is a
        // designed downgrade, and a server that answered with an older version
        // still serves `tools/call` — it simply never returns a task, so a
        // long-running tool behaves synchronously and nothing says why. This
        // tier has no Rust in which to ask, so the line says it.
        match client.negotiated_version() {
            Some(version) => eprintln!("  mcp: {name} <- {command} (MCP {version})"),
            None => eprintln!("  mcp: {name} <- {command}"),
        }
        // What the server advertises, put beside what the operator granted.
        // The grant rules either way; what the operator is being told is that
        // the server now *wants* more than they gave it — the first observable
        // move of a server going bad, or having been swapped. Comparison, not
        // configuration: a listing failure costs the warning, never the plane.
        match client.discover().await {
            Ok(advertised) => {
                for manifest in manifests {
                    let mut catalog = agentplane::tools::ToolCatalog::from_manifest(manifest);
                    for (id, adv) in &advertised {
                        catalog = catalog.observed(id, *adv);
                    }
                    for id in catalog.overclaiming() {
                        eprintln!(
                            "  mcp: {name}: warning: `{id}` advertises more safety than \
                             manifest `{}` grants; the grant still rules",
                            manifest.metadata.name
                        );
                    }
                }
            }
            Err(e) => {
                eprintln!("  mcp: {name}: tools/list failed, advertisements not compared: {e}");
            }
        }
        wired.push((
            name.to_owned(),
            Arc::new(client) as Arc<dyn agentplane::tools::ToolClient>,
        ));
    }
    Ok(wired)
}

/// The same, in a build without the transport.
///
/// Naming the feature rather than ignoring the flag: a `--mcp` that silently did
/// nothing would produce a plane whose build then refuses for a *different*
/// reason — no tool catalogue — and send a reader looking at their manifest for
/// a mistake that is in their build.
#[cfg(not(feature = "mcp-stdio"))]
#[allow(clippy::unused_async)]
async fn connect_mcp_servers(
    specs: &[String],
    _manifests: &[Manifest],
) -> Result<Vec<(String, Arc<dyn agentplane::tools::ToolClient>)>, String> {
    if specs.is_empty() {
        return Ok(Vec::new());
    }
    Err(
        "this build cannot run an MCP server: `--mcp` needs the `mcp-stdio` feature. \
         Reinstall with `--features cli,mcp-stdio`, or use the `:full` container \
         image, which is built with it"
            .to_owned(),
    )
}

/// Turn on A2A push, if the operator granted anywhere to send it.
///
/// `--push-host` is the *whole* configuration, and that is the point:
/// [`PushSender`](agentplane::push::PushSender) already owns HTTPS-only, the
/// all-answer public-IP check, DNS pinning, manual per-hop redirects, the
/// timeout and secret redaction. What an operator supplies is **where**, which
/// is the one thing the crate cannot decide for them.
///
/// No host means push is not wired **and the card says so** — advertising a
/// capability nothing serves is worse than not having it, because a peer that
/// registers a webhook and never hears back has a worse day than one told up
/// front.
///
/// # Errors
///
/// If the card has already been signed, since push changes what the signature
/// covers.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn wire_push(
    server: agentplane::api::a2a::A2aServer,
    hosts: &[String],
    backend: &Backend,
) -> Result<agentplane::api::a2a::A2aServer, String> {
    if hosts.is_empty() {
        return Ok(server);
    }
    let policy = hosts
        .iter()
        .fold(agentplane::push::PushPolicy::new(), |policy, host| {
            policy.allow_host(host)
        });
    let server = server
        .with_push(
            backend.push(),
            Arc::new(agentplane::push::PushSender::new(policy))
                as Arc<dyn agentplane::push::PushTransport>,
        )
        .map_err(|e| e.to_string())?;
    for host in hosts {
        eprintln!("  push: https://{host}");
    }
    Ok(server)
}

/// Deliver due webhooks on a clock.
///
/// The task journal is the outbox: each receiver stores its first unacknowledged
/// sequence and the cursor advances only after HTTP 2xx, so a crash after the
/// POST but before persistence **repeats** an event rather than losing it —
/// which is the right way round, and which A2A receivers are required to
/// tolerate.
///
/// Shares the sweeper's cadence because it is the same job: the operator's
/// scheduler running the plane's periodic work. Several instances may race and
/// produce duplicates; cursors advance monotonically, so none can regress.
#[cfg(all(feature = "a2a-server", feature = "cedar"))]
fn spawn_push_worker(
    worker: agentplane::api::a2a::A2aPushWorker,
    every: u32,
    stop: Stop,
) -> Option<Task> {
    if every == 0 {
        return None;
    }
    Some(tokio::spawn(async move {
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(u64::from(every)));
        let mut stop = stop;
        loop {
            if !next_tick(&mut tick, &mut stop).await {
                break;
            }
            #[allow(clippy::disallowed_methods)]
            let at = time::OffsetDateTime::now_utc().unix_timestamp();
            let Ok(at) = u64::try_from(at) else { continue };
            match worker.run_once(at, PUSH_BATCH).await {
                // A batch that came back full is a backlog, and a parked
                // registration is a peer that will hear nothing until an
                // operator re-arms it. Neither may produce the same numbers —
                // or the same log level — as a quiet plane, which is I13
                // applied to this worker's own report.
                Ok(report) if report.needs_attention() => {
                    tracing::warn!(?report, "push delivery needs attention");
                }
                // On deliveries, not registrations: an idle plane re-reads
                // its registrations every tick, and that is not a delivery.
                Ok(report) if report.deliveries > 0 => {
                    tracing::info!(?report, "push delivered");
                }
                Ok(_) => {}
                Err(error) => tracing::error!(%error, "push delivery failed"),
            }
        }
    }))
}

/// The same verb, in a build that cannot answer it.
///
/// A binary that met `serve` with *unknown command* would be telling a reader
/// the feature does not exist, when it does and is one build flag away. Naming
/// the flag is the difference between a dead end and a next step — the same
/// reason the provider list is derived from the build rather than written out.
#[cfg(not(all(feature = "a2a-server", feature = "cedar")))]
#[allow(clippy::unnecessary_wraps)]
fn serve(_manifests: &[Manifest], _opts: &ServeArgs) -> Result<ExitCode, String> {
    Err(
        "this build cannot serve: `serve` needs the `a2a-server` and `cedar` features. \
         Reinstall with `--features cli,a2a-server,cedar`, or use the `:full` \
         container image, which is built with them"
            .to_owned(),
    )
}

/// Refuse a file whose behaviour is not in the file.
fn require_declarative(manifests: &[Manifest]) -> Result<(), String> {
    for manifest in manifests {
        if manifest.spec.execution.is_none() {
            return Err(format!(
                "manifest '{}' declares no `spec.execution`, so its behaviour is a skill somebody \
                 wrote and there is nothing here for this binary to run. Register it in your own \
                 binary with `RuntimeBuilder::agent(Agent::new(&manifest).skill(YourSkill))` instead",
                manifest.metadata.name
            ));
        }
    }
    Ok(())
}

/// The verbs' shared tail: report the run, print the answer, exit honestly.
fn conclude(outcome: &agentplane::runtime::RunOutcome) -> ExitCode {
    eprintln!("run {} — {:?}", outcome.run_id, outcome.status);
    if let Some(output) = &outcome.output {
        // The answer on stdout and everything else on stderr, so this
        // composes with a pipe instead of needing a flag to be quiet.
        println!("{}", output.peek());
    }
    // A refused, exhausted or failed run must not exit zero: whoever scripts
    // this needs the shell's own answer to "did it work".
    if matches!(outcome.status, RunStatus::Succeeded) {
        ExitCode::SUCCESS
    } else {
        ExitCode::FAILURE
    }
}

fn execute(manifests: &[Manifest], opts: &RunArgs) -> Result<ExitCode, String> {
    require_declarative(manifests)?;

    // Current-thread on purpose. A CLI runs one agent and exits, so a work
    // stealing pool buys nothing and would mean pulling `rt-multi-thread` into
    // a crate that has so far needed four tokio features.
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        // **The whole plane, not the journal alone**, as `serve` wires it: a
        // declaration naming memory or a wait is not a different declaration
        // because of which verb reached it.
        let (stores, tenant) = if let Some(backend) = opts.at.open().await? {
            (backend.stores(), backend.tenant())
        } else {
            // Said out loud rather than assumed: a run whose journal disappears
            // is the opposite of what this crate is for.
            eprintln!("note: journaling to memory; this run will not survive the process");
            (
                agentplane::runtime::Stores::on(Arc::new(
                    RedbStore::open_in_memory().map_err(|e| e.to_string())?,
                )),
                agentplane::core::TenantId::default(),
            )
        };
        let mut builder =
            with_providers(Runtime::builder_with(stores).tenant(tenant), manifests).await?;
        for (name, client) in connect_mcp_servers(&opts.mcp, manifests).await? {
            builder = builder.tool_server(name, client);
        }
        if let Some((registry, client)) = connect_peers(&opts.peer, manifests)? {
            builder = builder.peers(registry, client);
        }
        for manifest in manifests {
            builder = builder.agent(agentplane::runtime::Agent::new(manifest));
        }
        // `try_build`, because everything on this plane arrived as input: a
        // wiring mistake in a file somebody handed us is a refusal with a
        // sentence, not a programmer error worth a crash.
        let agent = builder.try_build().map_err(|e| e.to_string())?;

        let capability = entry_capability(manifests, opts.capability.as_deref())?;
        let outcome = agent
            .run(&capability, Tainted::trusted(opts.read_input()?))
            .await
            .map_err(|e| e.to_string())?;

        Ok(conclude(&outcome))
    })
}

/// Re-execute a recorded run against the same declaration.
///
/// The plane is rebuilt exactly as `run` builds it — same providers, same MCP
/// wiring, same agents — plus the whole case layer, because a **resume** may
/// continue past its recorded history and dispatch live: a run that suspended
/// on a task or a timer needs the stores those live in. `--strict` never
/// dispatches; it reads the history back and fails if this build diverges.
fn replay(manifests: &[Manifest], opts: &ReplayArgs) -> Result<ExitCode, String> {
    require_declarative(manifests)?;

    let run = agentplane::core::RunId::parse(&opts.run_id)
        .map_err(|e| format!("`{}` is not a run id: {e}", opts.run_id))?;
    let mode = if opts.strict {
        Mode::Strict
    } else {
        Mode::Resume
    };

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("could not start the async runtime: {e}"))?;

    rt.block_on(async {
        let backend = opts.at.open().await?;
        let mut builder = with_providers(
            Runtime::builder_with(backend.stores()).tenant(backend.tenant()),
            manifests,
        )
        .await?;
        for (name, client) in connect_mcp_servers(&opts.mcp, manifests).await? {
            builder = builder.tool_server(name, client);
        }
        if let Some((registry, client)) = connect_peers(&opts.peer, manifests)? {
            builder = builder.peers(registry, client);
        }
        for manifest in manifests {
            builder = builder.agent(agentplane::runtime::Agent::new(manifest));
        }
        let agent = builder.try_build().map_err(|e| e.to_string())?;

        let outcome = agent.replay(run, mode).await.map_err(|e| e.to_string())?;
        Ok(conclude(&outcome))
    })
}

/// Register a driver for each provider the manifest names — and only those.
///
/// Registering every driver whose key happens to be set would make the agent
/// runnable on a model its declaration does not name, the moment somebody
/// exports the wrong variable.
async fn with_providers(
    builder: RuntimeBuilder,
    manifests: &[Manifest],
) -> Result<RuntimeBuilder, String> {
    let mut builder = builder;
    let mut seen: Vec<String> = Vec::new();

    for manifest in manifests {
        let Some(models) = &manifest.spec.models else {
            continue;
        };
        for m in [models.privileged.as_ref(), models.quarantined.as_ref()]
            .into_iter()
            .flatten()
        {
            if seen.contains(&m.provider) {
                continue;
            }
            seen.push(m.provider.clone());
            builder = builder.provider(m.provider.clone(), driver(&m.provider).await?);
        }
    }
    Ok(builder)
}

/// Which capability a `run` starts, when the file holds a room.
///
/// Explicit beats implicit, and implicit is allowed only where the file leaves
/// no doubt: `--capability` always wins; a file providing exactly one
/// capability runs it; and a room with exactly one agent declaring
/// `topology.role: orchestrator` — whose declaration provides exactly one
/// capability — starts there, because the topology *is* the file saying where
/// the room begins. Anything else is a refusal that lists the candidates,
/// never a guess.
fn entry_capability(manifests: &[Manifest], asked: Option<&str>) -> Result<String, String> {
    let all: Vec<(&str, &str)> = manifests
        .iter()
        .flat_map(|m| {
            m.spec
                .capabilities
                .provides
                .iter()
                .map(move |c| (m.metadata.name.as_str(), c.as_str()))
        })
        .collect();

    if let Some(asked) = asked {
        if all.iter().any(|(_, c)| *c == asked) {
            return Ok(asked.to_owned());
        }
        return Err(format!(
            "no agent in this file provides '{asked}'. It provides: {}",
            all.iter().map(|(_, c)| *c).collect::<Vec<_>>().join(", ")
        ));
    }
    if let [(_, only)] = all.as_slice() {
        return Ok((*only).to_owned());
    }
    let orchestrators: Vec<&Manifest> = manifests
        .iter()
        .filter(|m| {
            m.spec
                .topology
                .as_ref()
                .is_some_and(|t| t.role == agentplane::manifest::Role::Orchestrator)
        })
        .collect();
    if let [desk] = orchestrators.as_slice()
        && let [only] = desk.spec.capabilities.provides.as_slice()
    {
        return Ok(only.clone());
    }
    Err(format!(
        "this file provides several capabilities and no single orchestrator to \
         start at — say which one with --capability. It provides: {}",
        all.iter()
            .map(|(agent, c)| format!("{c} ({agent})"))
            .collect::<Vec<_>>()
            .join(", ")
    ))
}

async fn driver(name: &str) -> Result<Arc<dyn ModelProvider>, String> {
    match name {
        #[cfg(feature = "providers")]
        "anthropic" => Ok(Arc::new(
            agentplane::model::anthropic::Anthropic::new(key("ANTHROPIC_API_KEY")?)
                .map_err(|e| e.to_string())?,
        )),
        #[cfg(feature = "bedrock")]
        "bedrock" => Ok(Arc::new(
            agentplane::model::bedrock::Bedrock::from_env(
                std::env::var("AWS_REGION").map_err(|_| {
                    "AWS_REGION is not set, and the manifest names Bedrock".to_owned()
                })?,
            )
            .await?,
        )),
        // `GEMINI_API_KEY`, falling back to `GOOGLE_API_KEY`: both are in wide
        // use, and a deployment that exported the other one would otherwise
        // meet an authentication failure naming neither.
        #[cfg(feature = "providers")]
        "gemini" => Ok(Arc::new(
            agentplane::model::gemini::Gemini::from_env().map_err(|e| e.to_string())?,
        )),
        #[cfg(feature = "providers")]
        "openai" => Ok(Arc::new(
            agentplane::model::openai::OpenAi::new(key("OPENAI_API_KEY")?)
                .map_err(|e| e.to_string())?,
        )),
        // The OpenAI-compatible wire every self-hosted server speaks — TGI,
        // vLLM, Ollama, llama.cpp, and Hugging Face's hosted router. The base
        // URL is deployment wiring, so it comes from the environment like a
        // key does; the token is optional because the common local server
        // needs none.
        #[cfg(feature = "providers")]
        "chat-completions" => {
            let base = key("CHAT_COMPLETIONS_BASE_URL").map_err(|_| {
                "CHAT_COMPLETIONS_BASE_URL is not set, and the manifest names the \
                 chat-completions provider. Point it at the server: Ollama is \
                 http://localhost:11434, vLLM http://localhost:8000, TGI \
                 http://localhost:8080, Hugging Face's router \
                 https://router.huggingface.co/v1"
                    .to_owned()
            })?;
            let mut driver = agentplane::model::chat_completions::ChatCompletions::new(base)
                .map_err(|e| e.to_string())?;
            if let Ok(token) = std::env::var("CHAT_COMPLETIONS_API_KEY") {
                driver = driver.bearer(token);
            }
            Ok(Arc::new(driver))
        }
        #[cfg(feature = "fake-model")]
        "fake" => Ok(agentplane::model::fake::FakeProvider::new()),
        other => Err(format!(
            "no driver for provider '{other}'. This binary ships {}; anything else is an \
             embedder's own driver, registered through RuntimeBuilder::provider",
            shipped_providers().join(", "),
        )),
    }
}

/// Every provider name *this* binary can construct.
///
/// Assembled from the same `cfg`s as the dispatch above, rather than written
/// out as prose. A hand-written list is true only for whichever feature set
/// the author had in mind: the moment a driver becomes opt-in, the sentence
/// starts telling a reader their build has something it does not, and the
/// compiler has nothing to say about a string. A list derived from the build
/// cannot disagree with the build.
fn shipped_providers() -> Vec<&'static str> {
    #[allow(unused_mut)]
    let mut names: Vec<&'static str> = Vec::new();
    #[cfg(feature = "providers")]
    names.extend(["anthropic", "chat-completions", "gemini", "openai"]);
    #[cfg(feature = "bedrock")]
    names.push("bedrock");
    #[cfg(feature = "fake-model")]
    names.push("fake");
    names.sort_unstable();
    names
}

#[cfg(feature = "providers")]
fn key(var: &str) -> Result<String, String> {
    std::env::var(var)
        .map_err(|_| format!("{var} is not set, and the manifest names a provider that needs it"))
}

#[cfg(test)]
mod tests {
    use super::cutoff_before;

    /// A window a person could type is answered; one that reaches past the
    /// calendar is a message rather than an abort.
    ///
    /// The subtraction `time` performs panics on underflow, so the unchecked
    /// form ended a retention command with `overflow subtracting duration from
    /// date` and no mention of the flag that caused it.
    #[test]
    fn a_retention_window_past_the_calendar_is_a_message_not_an_abort() {
        let now = time::macros::datetime!(2026-09-13 12:00:00 UTC);
        assert!(cutoff_before(now, 90).is_ok());
        let err = cutoff_before(now, u32::MAX).expect_err("a window of 11m years");
        assert!(err.contains("--older-than-days"), "{err}");
    }
}