refactor(cli): split main.rs into cli/helpers/output modules

Verbatim moves: the clap surface (every command/subcommand/arg struct) to
cli.rs, resolution helpers (config/actor/graph/branch/query, remote HTTP,
env/token, scaffolding) to helpers.rs, human/JSON formatting to output.rs,
the in-source test mod to main_tests.rs via #[path]. main.rs (1,184 lines)
keeps main() and the dispatch match. Visibility bumps only; 22 binary
tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/omnigraph-cli/src/cli.rs

@@ -0,0 +1,650 @@
+//! The clap surface: every command, subcommand, and argument struct
+//! (moved verbatim from main.rs in the modularization).
+
+use super::*;
+
+pub(crate) const DEFAULT_BEARER_TOKEN_ENV: &str = "OMNIGRAPH_BEARER_TOKEN";
+
+#[derive(Debug, Parser)]
+#[command(name = "omnigraph")]
+#[command(about = "Omnigraph graph database CLI")]
+#[command(version = env!("CARGO_PKG_VERSION"), disable_version_flag = true)]
+pub(crate) struct Cli {
+    /// Actor identity for direct-engine writes (MR-722). Overrides
+    /// `cli.actor` from `omnigraph.yaml`. When the configured policy
+    /// is in effect, Cedar evaluates this actor against the requested
+    /// action and scope; with policy configured but neither this flag
+    /// nor `cli.actor` set, the engine-layer footgun guard fires and
+    /// the write is denied (no silent bypass). Has no effect on remote
+    /// HTTP writes — those resolve their actor server-side from the
+    /// bearer token.
+    #[arg(long = "as", global = true, value_name = "ACTOR")]
+    pub(crate) as_actor: Option<String>,
+
+    #[command(subcommand)]
+    pub(crate) command: Command,
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum Command {
+    /// Print the CLI version
+    Version,
+    /// Generate, clean, or refresh explicit seed embeddings
+    Embed(EmbedArgs),
+    /// Initialize a new graph from a schema
+    Init {
+        #[arg(long)]
+        schema: PathBuf,
+        /// Graph URI (local path or s3://)
+        uri: String,
+        /// Overwrite existing schema artifacts at the URI. Without
+        /// this flag, init refuses to touch a URI that already holds
+        /// `_schema.pg`, `_schema.ir.json`, or `__schema_state.json`
+        /// — closes the re-init footgun (MR-668 follow-up). With the
+        /// flag, the operator opts in to destructive semantics.
+        #[arg(long)]
+        force: bool,
+    },
+    /// Load data into a graph (local or remote)
+    Load {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        data: PathBuf,
+        /// Target branch (defaults to main). Without --from it must exist.
+        #[arg(long)]
+        branch: Option<String>,
+        /// Base branch to fork --branch from when it doesn't exist yet.
+        /// Without this flag a missing branch is an error, never a fork.
+        #[arg(long)]
+        from: Option<String>,
+        /// How existing rows are handled: overwrite | append | merge.
+        /// Required — overwrite is destructive, so there is no default.
+        #[arg(long)]
+        mode: CliLoadMode,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Deprecated alias of `load --from <base>` (defaults: --mode merge, --from main)
+    Ingest {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        data: PathBuf,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long)]
+        from: Option<String>,
+        #[arg(long, default_value = "merge")]
+        mode: CliLoadMode,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Branch operations
+    Branch {
+        #[command(subcommand)]
+        command: BranchCommand,
+    },
+    /// Schema planning operations
+    Schema {
+        #[command(subcommand)]
+        command: SchemaCommand,
+    },
+    /// Validate queries against a schema (offline) or repo (repo-backed).
+    ///
+    /// Canonical name is `lint` (matches the `omnigraph_compiler::lint`
+    /// module and the `OG-XXX-NNN` lint-code vocabulary). Replaces the
+    /// deprecated `omnigraph query lint` / `omnigraph query check` /
+    /// `omnigraph check` invocations — each is kept as an argv-level
+    /// shim that prints a one-line stderr warning and rewrites to
+    /// `omnigraph lint`. Aliases are deliberately *not* exposed via
+    /// clap's `visible_alias` because that would advertise two
+    /// equivalent canonical names, which agents emit interchangeably
+    /// (see MR-981).
+    Lint {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        query: PathBuf,
+        #[arg(long)]
+        schema: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Operate on the server-side stored-query registry (`queries:`).
+    Queries {
+        #[command(subcommand)]
+        command: QueriesCommand,
+    },
+    /// Show graph snapshot
+    Snapshot {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Export a full graph snapshot as JSONL
+    Export {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long, hide = true)]
+        jsonl: bool,
+        #[arg(long = "type")]
+        type_names: Vec<String>,
+        #[arg(long = "table")]
+        table_keys: Vec<String>,
+    },
+    /// Commit history operations
+    Commit {
+        #[command(subcommand)]
+        command: CommitCommand,
+    },
+    /// Execute a read query against a branch or snapshot.
+    ///
+    /// Canonical read endpoint. The previous name `omnigraph read` is
+    /// kept as a visible alias and prints a one-line deprecation warning
+    /// when used. Pairs with `omnigraph mutate` on the write side.
+    #[command(visible_alias = "read")]
+    Query {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(hide = true)]
+        legacy_uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long, conflicts_with_all = ["query", "query_string"])]
+        alias: Option<String>,
+        #[arg(long, conflicts_with_all = ["alias", "query_string"])]
+        query: Option<PathBuf>,
+        /// Inline GQ source — alternative to `--query <path>` and `--alias <name>`.
+        #[arg(short = 'e', long = "query-string", value_name = "GQ", conflicts_with_all = ["query", "alias"])]
+        query_string: Option<String>,
+        #[arg(long)]
+        name: Option<String>,
+        #[command(flatten)]
+        params: ParamsArgs,
+        #[arg(long, conflicts_with = "snapshot")]
+        branch: Option<String>,
+        #[arg(long, conflicts_with = "branch")]
+        snapshot: Option<String>,
+        #[arg(long, conflicts_with = "json")]
+        format: Option<ReadOutputFormat>,
+        #[arg(long, conflicts_with = "format")]
+        json: bool,
+        #[arg()]
+        alias_args: Vec<String>,
+    },
+    /// Execute a graph mutation query against a branch.
+    ///
+    /// Canonical mutation endpoint. The previous name `omnigraph change`
+    /// is kept as a visible alias and prints a one-line deprecation
+    /// warning when used. Pairs with `omnigraph query` on the read side.
+    #[command(visible_alias = "change")]
+    Mutate {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(hide = true)]
+        legacy_uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long, conflicts_with_all = ["query", "query_string"])]
+        alias: Option<String>,
+        #[arg(long, conflicts_with_all = ["alias", "query_string"])]
+        query: Option<PathBuf>,
+        /// Inline GQ source — alternative to `--query <path>` and `--alias <name>`.
+        #[arg(short = 'e', long = "query-string", value_name = "GQ", conflicts_with_all = ["query", "alias"])]
+        query_string: Option<String>,
+        #[arg(long)]
+        name: Option<String>,
+        #[command(flatten)]
+        params: ParamsArgs,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long)]
+        json: bool,
+        #[arg()]
+        alias_args: Vec<String>,
+    },
+    /// Policy administration and diagnostics
+    Policy {
+        #[command(subcommand)]
+        command: PolicyCommand,
+    },
+    /// Compact small Lance fragments in every table of the graph
+    Optimize {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Classify and explicitly repair manifest/head drift
+    Repair {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        /// Publish verified maintenance drift. Without this flag, repair only
+        /// previews what it would do.
+        #[arg(long)]
+        confirm: bool,
+        /// Also publish suspicious or unverifiable drift. Requires
+        /// `--confirm`; use only after operator review.
+        #[arg(long, requires = "confirm")]
+        force: bool,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Remove old Lance versions from every table of the graph (destructive)
+    Cleanup {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        /// Number of recent versions to keep per table. Either `--keep` or
+        /// `--older-than` (or both) must be set.
+        #[arg(long)]
+        keep: Option<u32>,
+        /// Only remove versions older than this duration. Accepts Go-style
+        /// durations: `7d`, `24h`, `90m`. At least one of --keep / --older-than.
+        #[arg(long)]
+        older_than: Option<String>,
+        /// Required to actually run; without it, prints what would be removed
+        #[arg(long)]
+        confirm: bool,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Validate and plan read-only cluster configuration.
+    Cluster {
+        #[command(subcommand)]
+        command: ClusterCommand,
+    },
+    /// Manage graphs on a multi-graph server (MR-668)
+    Graphs {
+        #[command(subcommand)]
+        command: GraphsCommand,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum ClusterCommand {
+    /// Validate cluster.yaml and referenced schemas, queries, and policy files.
+    Validate {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Produce a read-only plan by diffing cluster.yaml against __cluster/state.json.
+    Plan {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Apply the config-only (query/policy) subset of the plan to the local
+    /// cluster catalog. Graph/schema changes are deferred to a later stage.
+    Apply {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Record a digest-bound approval for a gated (irreversible) change,
+    /// e.g. a graph delete. Requires the global --as actor.
+    Approve {
+        /// Typed resource address of the gated change (e.g. graph.scratch).
+        resource: String,
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Read the local JSON state ledger without scanning live graph resources.
+    Status {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Refresh existing local JSON state from declared graph observations.
+    Refresh {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Import initial local JSON state from declared graph observations.
+    Import {
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Remove a held local JSON state lock after operator confirmation.
+    ForceUnlock {
+        /// Exact lock id from cluster status or a state_lock_held diagnostic.
+        lock_id: String,
+        /// Cluster config directory containing cluster.yaml.
+        #[arg(long, default_value = ".")]
+        config: PathBuf,
+        /// Emit JSON instead of human text.
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+/// Operations on the graph registry of a multi-graph server (MR-668).
+///
+/// All operations target a remote multi-graph server URL (http:// or
+/// https://). Local-URI invocations return a clear error. To add or
+/// remove graphs, operators edit `omnigraph.yaml` directly and restart
+/// the server — runtime mutation is not exposed in v0.6.0.
+#[derive(Debug, Subcommand)]
+pub(crate) enum GraphsCommand {
+    /// List every graph registered with the multi-graph server.
+    List {
+        /// Remote server URL (e.g. `https://server.example.com`).
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum BranchCommand {
+    /// Create a new branch
+    Create {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        from: Option<String>,
+        name: String,
+        #[arg(long)]
+        json: bool,
+    },
+    /// List branches
+    List {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Delete a branch
+    Delete {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        name: String,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Merge a source branch into a target branch
+    Merge {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        source: String,
+        #[arg(long)]
+        into: Option<String>,
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum SchemaCommand {
+    /// Plan a schema migration against the accepted persisted schema
+    Plan {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        schema: PathBuf,
+        #[arg(long)]
+        json: bool,
+        /// Show the plan as it would execute with `--allow-data-loss`.
+        /// Promotes every `DropMode::Soft` step to `DropMode::Hard`
+        /// so the plan output reflects the destructive intent.
+        #[arg(long, default_value_t = false)]
+        allow_data_loss: bool,
+    },
+    /// Apply a supported schema migration
+    Apply {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        schema: PathBuf,
+        #[arg(long)]
+        json: bool,
+        /// Allow destructive (data-loss) schema changes.
+        ///
+        /// Without this flag, drops are "soft": the column or table
+        /// is removed from the current manifest version but prior
+        /// versions are retained, so `snapshot_at_version(pre_drop)`
+        /// can still read the dropped data until `omnigraph cleanup`
+        /// runs. With this flag, drops are "hard": `cleanup_old_versions`
+        /// runs on the affected datasets immediately after the apply,
+        /// making the prior data unreachable.
+        #[arg(long, default_value_t = false)]
+        allow_data_loss: bool,
+    },
+    /// Show the current accepted schema source
+    #[command(alias = "get")]
+    Show {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+
+pub(crate) enum CommitCommand {
+    /// List graph commits
+    List {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Show a graph commit
+    Show {
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        commit_id: String,
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum PolicyCommand {
+    /// Validate policy YAML and compiled Cedar policy state
+    Validate {
+        #[arg(long)]
+        config: Option<PathBuf>,
+    },
+    /// Run declarative policy tests from policy.tests.yaml
+    Test {
+        #[arg(long)]
+        config: Option<PathBuf>,
+    },
+    /// Explain one policy decision locally
+    Explain {
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        actor: String,
+        #[arg(long)]
+        action: PolicyAction,
+        #[arg(long)]
+        branch: Option<String>,
+        #[arg(long = "target-branch")]
+        target_branch: Option<String>,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum QueriesCommand {
+    /// Type-check the stored-query registry against the live schema.
+    ///
+    /// Distinct from `omnigraph lint` (which lints one `.gq` file):
+    /// this validates the whole `queries:` registry — opening the graph
+    /// to read its schema and confirming every stored query still
+    /// type-checks. Exits non-zero on any breakage.
+    Validate {
+        /// Graph URI
+        uri: Option<String>,
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+    /// List the registered stored queries (name, MCP exposure, params).
+    List {
+        #[arg(long)]
+        target: Option<String>,
+        #[arg(long)]
+        config: Option<PathBuf>,
+        #[arg(long)]
+        json: bool,
+    },
+}
+
+#[derive(Debug, Args, Clone)]
+pub(crate) struct ParamsArgs {
+    #[arg(long, conflicts_with = "params_file")]
+    pub(crate) params: Option<String>,
+    #[arg(long, conflicts_with = "params")]
+    pub(crate) params_file: Option<PathBuf>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, ValueEnum)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum CliLoadMode {
+    Overwrite,
+    Append,
+    Merge,
+}
+
+impl From<CliLoadMode> for LoadMode {
+    fn from(value: CliLoadMode) -> Self {
+        match value {
+            CliLoadMode::Overwrite => LoadMode::Overwrite,
+            CliLoadMode::Append => LoadMode::Append,
+            CliLoadMode::Merge => LoadMode::Merge,
+        }
+    }
+}
+
+impl CliLoadMode {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            CliLoadMode::Overwrite => "overwrite",
+            CliLoadMode::Append => "append",
+            CliLoadMode::Merge => "merge",
+        }
+    }
+}
+

crates/omnigraph-cli/src/main_tests.rs

@@ -0,0 +1,416 @@
+//! In-source test suite for the CLI binary (moved verbatim from
+//! main.rs; `use super::*` resolves through the #[path] declaration).
+
+    use std::fs;
+
+    use super::{
+        DEFAULT_BEARER_TOKEN_ENV, apply_bearer_token, bearer_token_from_env_file,
+        legacy_change_request_body, load_cli_config, load_env_file_into_process,
+        normalize_bearer_token, parse_env_assignment, resolve_cli_graph, resolve_policy_context,
+        resolve_remote_bearer_token,
+    };
+    use omnigraph_server::load_config;
+    use reqwest::header::AUTHORIZATION;
+    use serde_json::json;
+    use tempfile::tempdir;
+
+    #[test]
+    fn legacy_change_request_body_uses_legacy_field_names() {
+        // `execute_change_remote` hits `POST /change`, which old
+        // `omnigraph-server` builds deserialize as `ChangeRequest` with
+        // **required** `query_source` and optional `query_name` keys.
+        // Newer servers accept both spellings via serde alias, but a
+        // newer CLI must still emit the legacy keys on the wire so it
+        // can talk to an old server during a rolling upgrade.
+        let body = legacy_change_request_body(
+            "query insert_person($n: String) { insert Person { name: $n } }",
+            Some("insert_person"),
+            "main",
+            Some(&json!({ "n": "Alice" })),
+        );
+        assert_eq!(
+            body["query_source"].as_str(),
+            Some("query insert_person($n: String) { insert Person { name: $n } }"),
+        );
+        assert_eq!(body["query_name"].as_str(), Some("insert_person"));
+        assert_eq!(body["branch"].as_str(), Some("main"));
+        assert_eq!(body["params"]["n"].as_str(), Some("Alice"));
+        // Crucially, the **new** field names must NOT appear -- old
+        // servers would silently treat them as unknown fields and then
+        // fail on missing required `query_source`.
+        assert!(
+            body.get("query").is_none(),
+            "legacy /change body must not carry the renamed `query` key; got {body}"
+        );
+        assert!(
+            body.get("name").is_none(),
+            "legacy /change body must not carry the renamed `name` key; got {body}"
+        );
+    }
+
+    #[test]
+    fn legacy_change_request_body_omits_optional_fields_when_unset() {
+        let body = legacy_change_request_body(
+            "query find() { match { $p: Person } return { $p.name } }",
+            None,
+            "main",
+            None,
+        );
+        assert_eq!(body["branch"].as_str(), Some("main"));
+        assert!(body.get("query_name").is_none());
+        assert!(body.get("params").is_none());
+    }
+
+    #[test]
+    fn apply_bearer_token_adds_header_when_configured() {
+        let client = reqwest::Client::new();
+        let request = apply_bearer_token(client.get("http://example.com"), Some("demo-token"))
+            .build()
+            .unwrap();
+        assert_eq!(
+            request
+                .headers()
+                .get(AUTHORIZATION)
+                .and_then(|value| value.to_str().ok()),
+            Some("Bearer demo-token")
+        );
+    }
+
+    #[test]
+    fn apply_bearer_token_leaves_request_unchanged_when_not_configured() {
+        let client = reqwest::Client::new();
+        let request = apply_bearer_token(client.get("http://example.com"), None)
+            .build()
+            .unwrap();
+        assert!(request.headers().get(AUTHORIZATION).is_none());
+    }
+
+    #[test]
+    fn normalize_bearer_token_trims_and_filters_blank_values() {
+        assert_eq!(normalize_bearer_token(None), None);
+        assert_eq!(normalize_bearer_token(Some("   ".to_string())), None);
+        assert_eq!(
+            normalize_bearer_token(Some(" demo-token ".to_string())).as_deref(),
+            Some("demo-token")
+        );
+    }
+
+    #[test]
+    fn parse_env_assignment_supports_plain_and_exported_values() {
+        assert_eq!(
+            parse_env_assignment("DEMO_TOKEN=demo-token"),
+            Some(("DEMO_TOKEN".to_string(), "demo-token".to_string()))
+        );
+        assert_eq!(
+            parse_env_assignment("export DEMO_TOKEN=\"quoted-token\""),
+            Some(("DEMO_TOKEN".to_string(), "quoted-token".to_string()))
+        );
+        assert_eq!(parse_env_assignment("# comment"), None);
+        assert_eq!(parse_env_assignment("   "), None);
+    }
+
+    #[test]
+    fn bearer_token_from_env_file_reads_named_value() {
+        let temp = tempdir().unwrap();
+        let env_file = temp.path().join(".env.omni");
+        fs::write(
+            &env_file,
+            "FIRST=ignore\nexport DEMO_TOKEN=\" demo-token \"\n",
+        )
+        .unwrap();
+
+        assert_eq!(
+            bearer_token_from_env_file(&env_file, "DEMO_TOKEN")
+                .unwrap()
+                .as_deref(),
+            Some("demo-token")
+        );
+        assert_eq!(
+            bearer_token_from_env_file(&env_file, "MISSING").unwrap(),
+            None
+        );
+    }
+
+    #[test]
+    fn load_env_file_into_process_sets_missing_values_without_overriding_existing_ones() {
+        let temp = tempdir().unwrap();
+        let env_file = temp.path().join(".env.omni");
+        fs::write(
+            &env_file,
+            "AUTOLOAD_ONLY=from-file\nAUTOLOAD_PRESET=from-file\n",
+        )
+        .unwrap();
+
+        let missing_key = "AUTOLOAD_ONLY";
+        let preset_key = "AUTOLOAD_PRESET";
+        let previous_missing = std::env::var_os(missing_key);
+        let previous_preset = std::env::var_os(preset_key);
+
+        unsafe {
+            std::env::remove_var(missing_key);
+            std::env::set_var(preset_key, "from-env");
+        }
+
+        load_env_file_into_process(&env_file).unwrap();
+
+        assert_eq!(std::env::var(missing_key).unwrap(), "from-file");
+        assert_eq!(std::env::var(preset_key).unwrap(), "from-env");
+
+        unsafe {
+            if let Some(value) = previous_missing {
+                std::env::set_var(missing_key, value);
+            } else {
+                std::env::remove_var(missing_key);
+            }
+
+            if let Some(value) = previous_preset {
+                std::env::set_var(preset_key, value);
+            } else {
+                std::env::remove_var(preset_key);
+            }
+        }
+    }
+
+    #[test]
+    fn resolve_remote_bearer_token_uses_scoped_env_file_with_global_fallback() {
+        let temp = tempdir().unwrap();
+        fs::write(
+            temp.path().join("omnigraph.yaml"),
+            r#"
+graphs:
+  demo:
+    uri: https://example.com
+    bearer_token_env: DEMO_TOKEN
+auth:
+  env_file: .env.omni
+cli:
+  graph: demo
+"#,
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join(".env.omni"),
+            "DEMO_TOKEN=scoped-token\nOMNIGRAPH_BEARER_TOKEN=global-token\n",
+        )
+        .unwrap();
+
+        let previous = std::env::var_os(DEFAULT_BEARER_TOKEN_ENV);
+        unsafe {
+            std::env::remove_var(DEFAULT_BEARER_TOKEN_ENV);
+        }
+
+        let config_path = temp.path().join("omnigraph.yaml");
+        let config = load_config(Some(&config_path)).unwrap();
+
+        assert_eq!(
+            resolve_remote_bearer_token(&config, None, Some("demo"))
+                .unwrap()
+                .as_deref(),
+            Some("scoped-token")
+        );
+        assert_eq!(
+            resolve_remote_bearer_token(&config, Some("https://override.example.com"), None)
+                .unwrap()
+                .as_deref(),
+            Some("global-token")
+        );
+
+        unsafe {
+            if let Some(value) = previous {
+                std::env::set_var(DEFAULT_BEARER_TOKEN_ENV, value);
+            } else {
+                std::env::remove_var(DEFAULT_BEARER_TOKEN_ENV);
+            }
+        }
+    }
+
+    #[test]
+    fn load_cli_config_autoloads_env_file_into_process() {
+        let temp = tempdir().unwrap();
+        fs::write(
+            temp.path().join("omnigraph.yaml"),
+            r#"
+auth:
+  env_file: .env.omni
+graphs:
+  demo:
+    uri: s3://bucket/prefix
+"#,
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join(".env.omni"),
+            "AUTOLOAD_FROM_CONFIG=loaded\n",
+        )
+        .unwrap();
+
+        let key = "AUTOLOAD_FROM_CONFIG";
+        let previous = std::env::var_os(key);
+        unsafe {
+            std::env::remove_var(key);
+        }
+
+        let config_path = temp.path().join("omnigraph.yaml");
+        let config = load_cli_config(Some(&config_path)).unwrap();
+
+        assert_eq!(
+            config.resolve_target_uri(None, Some("demo"), None).unwrap(),
+            "s3://bucket/prefix"
+        );
+        assert_eq!(std::env::var(key).unwrap(), "loaded");
+
+        unsafe {
+            if let Some(value) = previous {
+                std::env::set_var(key, value);
+            } else {
+                std::env::remove_var(key);
+            }
+        }
+    }
+
+    #[test]
+    fn graph_identity_resolve_policy_context_named_cli_graph_uses_graph_key_not_project_name_or_uri()
+     {
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+project:
+  name: misleading-project
+graphs:
+  local:
+    uri: /tmp/local-policy-graph.omni
+    policy:
+      file: ./policy.yaml
+cli:
+  graph: local
+"#,
+        )
+        .unwrap();
+
+        let config = load_config(Some(&config_path)).unwrap();
+        let context = resolve_policy_context(&config).unwrap();
+        assert_eq!(context.graph_id, "local");
+    }
+
+    #[test]
+    fn graph_identity_resolve_policy_context_server_graph_uses_graph_key_when_cli_graph_absent() {
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+project:
+  name: misleading-project
+graphs:
+  local:
+    uri: /tmp/local-policy-graph.omni
+    policy:
+      file: ./server-policy.yaml
+server:
+  graph: local
+"#,
+        )
+        .unwrap();
+
+        let config = load_config(Some(&config_path)).unwrap();
+        let context = resolve_policy_context(&config).unwrap();
+        assert_eq!(context.graph_id, "local");
+        assert!(context.policy_file.ends_with("server-policy.yaml"));
+    }
+
+    #[test]
+    fn graph_identity_resolve_policy_context_anonymous_uses_top_level_default_identity() {
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+project:
+  name: misleading-project
+graphs:
+  local:
+    uri: /tmp/local-policy-graph.omni
+policy:
+  file: ./top-policy.yaml
+"#,
+        )
+        .unwrap();
+
+        let config = load_config(Some(&config_path)).unwrap();
+        let context = resolve_policy_context(&config).unwrap();
+        assert_eq!(context.graph_id, "default");
+        assert!(context.policy_file.ends_with("top-policy.yaml"));
+    }
+
+    #[test]
+    fn graph_identity_resolve_cli_graph_named_target_uses_graph_key_not_project_name_or_uri() {
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+project:
+  name: misleading-project
+graphs:
+  prod:
+    uri: s3://bucket/prod-graph/
+    policy:
+      file: ./prod-policy.yaml
+"#,
+        )
+        .unwrap();
+
+        let config = load_config(Some(&config_path)).unwrap();
+        let graph = resolve_cli_graph(&config, None, Some("prod")).unwrap();
+        assert_eq!(graph.selected(), Some("prod"));
+        assert_eq!(graph.graph_id, "prod");
+        assert_eq!(graph.uri, "s3://bucket/prod-graph/");
+    }
+
+    #[test]
+    fn graph_identity_resolve_cli_graph_positional_uri_uses_anonymous_normalized_uri() {
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+project:
+  name: misleading-project
+graphs:
+  local:
+    uri: /tmp/configured-graph.omni
+    policy:
+      file: ./policy.yaml
+cli:
+  graph: local
+"#,
+        )
+        .unwrap();
+
+        let config = load_config(Some(&config_path)).unwrap();
+        let local_graph_path = temp.path().join("explicit-graph.omni");
+        let local_graph = resolve_cli_graph(
+            &config,
+            Some(format!("file://{}", local_graph_path.display())),
+            None,
+        )
+        .unwrap();
+        assert_eq!(local_graph.selected(), None);
+        assert_eq!(
+            local_graph.graph_id,
+            local_graph_path.to_string_lossy().as_ref()
+        );
+        assert_eq!(local_graph.policy_file, None);
+
+        let s3_graph = resolve_cli_graph(
+            &config,
+            Some("s3://bucket/anonymous-graph/".to_string()),
+            None,
+        )
+        .unwrap();
+        assert_eq!(s3_graph.selected(), None);
+        assert_eq!(s3_graph.graph_id, "s3://bucket/anonymous-graph");
+        assert_eq!(s3_graph.policy_file, None);
+    }

crates/omnigraph-cli/src/output.rs

@@ -0,0 +1,830 @@
+//! Human/JSON output formatting for every command (moved verbatim from
+//! main.rs in the modularization).
+
+use super::*;
+
+#[derive(Debug, Serialize)]
+pub(crate) struct LoadOutput {
+    pub(crate) uri: String,
+    pub(crate) branch: String,
+    pub(crate) mode: &'static str,
+    /// Present only when `--from` was given; echoes the requested base.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) base_branch: Option<String>,
+    pub(crate) branch_created: bool,
+    pub(crate) nodes_loaded: usize,
+    pub(crate) edges_loaded: usize,
+    pub(crate) node_types_loaded: usize,
+    pub(crate) edge_types_loaded: usize,
+}
+
+pub(crate) fn load_output_from_tables(
+    uri: &str,
+    branch: &str,
+    mode: CliLoadMode,
+    output: &IngestOutput,
+) -> LoadOutput {
+    let mut nodes_loaded = 0;
+    let mut edges_loaded = 0;
+    let mut node_types_loaded = 0;
+    let mut edge_types_loaded = 0;
+    for table in &output.tables {
+        if table.table_key.starts_with("node:") {
+            nodes_loaded += table.rows_loaded;
+            node_types_loaded += 1;
+        } else if table.table_key.starts_with("edge:") {
+            edges_loaded += table.rows_loaded;
+            edge_types_loaded += 1;
+        }
+    }
+    LoadOutput {
+        uri: uri.to_string(),
+        branch: branch.to_string(),
+        mode: mode.as_str(),
+        base_branch: output.base_branch.clone(),
+        branch_created: output.branch_created,
+        nodes_loaded,
+        edges_loaded,
+        node_types_loaded,
+        edge_types_loaded,
+    }
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct SchemaPlanOutput<'a> {
+    pub(crate) uri: &'a str,
+    pub(crate) supported: bool,
+    pub(crate) step_count: usize,
+    pub(crate) steps: &'a [SchemaMigrationStep],
+}
+
+pub(crate) fn print_schema_apply_human(output: &SchemaApplyOutput) {
+    println!("schema apply for {}", output.uri);
+    println!("supported: {}", if output.supported { "yes" } else { "no" });
+    println!("applied: {}", if output.applied { "yes" } else { "no" });
+    println!("manifest_version: {}", output.manifest_version);
+    if output.steps.is_empty() {
+        println!("no schema changes");
+        return;
+    }
+    for step in &output.steps {
+        println!("- {}", render_schema_plan_step(step));
+    }
+}
+
+pub(crate) fn query_kind_label(kind: QueryLintQueryKind) -> &'static str {
+    match kind {
+        QueryLintQueryKind::Read => "read",
+        QueryLintQueryKind::Mutation => "mutation",
+    }
+}
+
+pub(crate) fn severity_label(severity: QueryLintSeverity) -> &'static str {
+    match severity {
+        QueryLintSeverity::Error => "ERROR",
+        QueryLintSeverity::Warning => "WARN ",
+        QueryLintSeverity::Info => "INFO ",
+    }
+}
+
+pub(crate) fn print_query_lint_human(output: &QueryLintOutput) {
+    for result in &output.results {
+        match result.status {
+            QueryLintStatus::Ok => {
+                println!(
+                    "OK    query `{}` ({})",
+                    result.name,
+                    query_kind_label(result.kind)
+                );
+            }
+            QueryLintStatus::Error => {
+                println!(
+                    "ERROR query `{}`: {}",
+                    result.name,
+                    result.error.as_deref().unwrap_or("unknown error")
+                );
+            }
+        }
+
+        for warning in &result.warnings {
+            println!("WARN  query `{}`: {}", result.name, warning);
+        }
+    }
+
+    for finding in &output.findings {
+        println!("{} {}", severity_label(finding.severity), finding.message);
+    }
+
+    println!(
+        "INFO  Lint complete: {} queries processed ({} error(s), {} warning(s), {} info item(s))",
+        output.queries_processed, output.errors, output.warnings, output.infos
+    );
+}
+
+pub(crate) fn finish_query_lint(output: &QueryLintOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_query_lint_human(output);
+    }
+
+    if output.status == QueryLintStatus::Error {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+
+    Ok(())
+}
+
+pub(crate) fn print_json<T: Serialize>(value: &T) -> Result<()> {
+    println!("{}", serde_json::to_string_pretty(value)?);
+    Ok(())
+}
+
+pub(crate) fn print_cluster_validate_human(output: &ValidateOutput) {
+    if output.ok {
+        println!(
+            "cluster config valid: {} resource(s), {} dependency edge(s)",
+            output.resources.len(),
+            output.dependencies.len()
+        );
+    } else {
+        println!("cluster config invalid");
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn print_cluster_plan_human(output: &PlanOutput) {
+    if output.ok {
+        println!(
+            "cluster plan: {} change(s), {} approval gate(s)",
+            output.changes.len(),
+            output.approvals_required.len()
+        );
+        for change in &output.changes {
+            let bindings = if change.binding_change { " [bindings]" } else { "" };
+            println!("  {:?} {}{bindings}", change.operation, change.resource);
+            if let Some(migration) = &change.migration {
+                if !migration.supported {
+                    println!("      migration UNSUPPORTED:");
+                }
+                for step in &migration.steps {
+                    println!(
+                        "      {}",
+                        serde_json::to_string(step).unwrap_or_else(|_| format!("{step:?}"))
+                    );
+                }
+            }
+        }
+        if output.changes.is_empty() {
+            println!("  no changes");
+        }
+    } else {
+        println!("cluster plan failed");
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn print_cluster_apply_human(output: &ApplyOutput) {
+    if output.ok {
+        println!(
+            "cluster apply: {} applied, {} deferred/blocked",
+            output.applied_count, output.deferred_count
+        );
+    } else {
+        println!("cluster apply failed");
+    }
+    // The change list prints on failure too: an operator debugging a partial
+    // apply (payload or state-write error) needs to see what was attempted.
+    print_cluster_apply_changes(&output.changes);
+    if output.ok {
+        let state = &output.state_observations;
+        println!(
+            "  state: revision {}, converged: {}, written: {}",
+            state.state_revision, output.converged, output.state_written
+        );
+        println!("  note: cluster-booted servers (--cluster) serve this on their next restart; omnigraph.yaml deployments are unaffected");
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn print_cluster_apply_changes(changes: &[omnigraph_cluster::PlanChange]) {
+    for change in changes {
+        let bindings = if change.binding_change { " [bindings]" } else { "" };
+        match (&change.disposition, change.reason.as_deref()) {
+            (Some(disposition), Some(reason)) => println!(
+                "  {:?} {}{bindings} [{disposition:?}: {reason}]",
+                change.operation, change.resource
+            ),
+            (Some(disposition), None) => println!(
+                "  {:?} {}{bindings} [{disposition:?}]",
+                change.operation, change.resource
+            ),
+            _ => println!("  {:?} {}{bindings}", change.operation, change.resource),
+        }
+    }
+    if changes.is_empty() {
+        println!("  no changes");
+    }
+}
+
+pub(crate) fn print_cluster_status_human(output: &StatusOutput) {
+    if output.ok {
+        let state = &output.state_observations;
+        if state.state_found {
+            println!(
+                "cluster state: revision {}, {} resource(s)",
+                state.state_revision, state.resource_count
+            );
+            if let Some(digest) = state.applied_config_digest.as_deref() {
+                println!("  applied config: {digest}");
+            }
+            if state.locked {
+                println!("  lock: held{}", cluster_lock_summary(state));
+            } else {
+                println!("  lock: not held");
+            }
+        } else {
+            println!("cluster state missing");
+        }
+    } else {
+        println!("cluster status failed");
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn print_cluster_state_sync_human(output: &StateSyncOutput) {
+    let operation = match output.operation {
+        omnigraph_cluster::StateSyncOperation::Refresh => "refresh",
+        omnigraph_cluster::StateSyncOperation::Import => "import",
+    };
+    if output.ok {
+        let state = &output.state_observations;
+        println!(
+            "cluster {operation}: revision {}, {} resource(s)",
+            state.state_revision, state.resource_count
+        );
+        if let Some(cas) = state.state_cas.as_deref() {
+            println!("  state_cas: {cas}");
+        }
+        if state.locked {
+            println!("  lock: acquired{}", cluster_lock_summary(state));
+        } else {
+            println!("  lock: not acquired");
+        }
+    } else {
+        println!("cluster {operation} failed");
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn print_cluster_force_unlock_human(output: &ForceUnlockOutput) {
+    if output.ok {
+        if output.lock_removed {
+            println!(
+                "cluster force-unlock: removed lock{}",
+                cluster_lock_summary(&output.state_observations)
+            );
+        } else {
+            println!("cluster force-unlock: no lock removed");
+        }
+    } else {
+        println!("cluster force-unlock failed");
+        if output.state_observations.locked {
+            println!(
+                "  lock: held{}",
+                cluster_lock_summary(&output.state_observations)
+            );
+        }
+    }
+    print_cluster_diagnostics(&output.diagnostics);
+}
+
+pub(crate) fn cluster_lock_summary(state: &omnigraph_cluster::StateObservations) -> String {
+    let Some(lock_id) = state.lock_id.as_deref() else {
+        return String::new();
+    };
+    let mut parts = vec![format!("id={lock_id}")];
+    if let Some(operation) = state.lock_operation.as_deref() {
+        parts.push(format!("operation={operation}"));
+    }
+    if let Some(pid) = state.lock_pid {
+        parts.push(format!("pid={pid}"));
+    }
+    if let Some(created_at) = state.lock_created_at.as_deref() {
+        parts.push(format!("created_at={created_at}"));
+    }
+    if let Some(age_seconds) = state.lock_age_seconds {
+        parts.push(format!("age_seconds={age_seconds}"));
+    }
+    format!(" ({})", parts.join(", "))
+}
+
+pub(crate) fn print_cluster_diagnostics(diagnostics: &[omnigraph_cluster::Diagnostic]) {
+    for diagnostic in diagnostics {
+        let label = match diagnostic.severity {
+            DiagnosticSeverity::Error => "ERROR",
+            DiagnosticSeverity::Warning => "WARN ",
+        };
+        println!(
+            "{label} {} {}: {}",
+            diagnostic.code, diagnostic.path, diagnostic.message
+        );
+    }
+}
+
+pub(crate) fn finish_cluster_validate(output: &ValidateOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_validate_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_plan(output: &PlanOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_plan_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_apply(output: &ApplyOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_apply_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_approve(output: &ApproveOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else if output.ok {
+        println!(
+            "cluster approve: {} {} approved by {} (approval {})",
+            output
+                .operation
+                .as_ref()
+                .map(|operation| format!("{operation:?}").to_lowercase())
+                .unwrap_or_default(),
+            output.resource.as_deref().unwrap_or("?"),
+            output.approved_by.as_deref().unwrap_or("?"),
+            output.approval_id.as_deref().unwrap_or("?"),
+        );
+        print_cluster_diagnostics(&output.diagnostics);
+    } else {
+        println!("cluster approve failed");
+        print_cluster_diagnostics(&output.diagnostics);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_status(output: &StatusOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_status_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_state_sync(output: &StateSyncOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_state_sync_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn finish_cluster_force_unlock(output: &ForceUnlockOutput, json: bool) -> Result<()> {
+    if json {
+        print_json(output)?;
+    } else {
+        print_cluster_force_unlock_human(output);
+    }
+    if !output.ok {
+        io::stdout().flush()?;
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+pub(crate) fn print_load_human(payload: &LoadOutput) {
+    println!(
+        "loaded {} on branch {} with {}: {} nodes across {} node types, {} edges across {} edge types",
+        payload.uri,
+        payload.branch,
+        payload.mode,
+        payload.nodes_loaded,
+        payload.node_types_loaded,
+        payload.edges_loaded,
+        payload.edge_types_loaded
+    );
+    if payload.branch_created {
+        if let Some(base) = &payload.base_branch {
+            println!("branch {} created from {}", payload.branch, base);
+        }
+    }
+}
+
+pub(crate) fn print_ingest_human(output: &IngestOutput) {
+    println!(
+        "ingested {} into branch {} from {} with {} ({})",
+        output.uri,
+        output.branch,
+        output.base_branch.as_deref().unwrap_or("main"),
+        output.mode.as_str(),
+        if output.branch_created {
+            "branch created"
+        } else {
+            "branch exists"
+        }
+    );
+    for table in &output.tables {
+        println!("{} rows_loaded={}", table.table_key, table.rows_loaded);
+    }
+    if let Some(actor_id) = &output.actor_id {
+        println!("actor_id: {}", actor_id);
+    }
+}
+
+pub(crate) fn print_schema_plan_human(uri: &str, plan: &SchemaMigrationPlan) {
+    println!("schema plan for {}", uri);
+    println!("supported: {}", if plan.supported { "yes" } else { "no" });
+    if plan.steps.is_empty() {
+        println!("no schema changes");
+        return;
+    }
+    for step in &plan.steps {
+        println!("- {}", render_schema_plan_step(step));
+    }
+}
+
+pub(crate) fn render_schema_plan_step(step: &SchemaMigrationStep) -> String {
+    match step {
+        SchemaMigrationStep::AddType { type_kind, name } => {
+            format!("add {} type '{}'", schema_type_kind_label(*type_kind), name)
+        }
+        SchemaMigrationStep::RenameType {
+            type_kind,
+            from,
+            to,
+        } => format!(
+            "rename {} type '{}' -> '{}'",
+            schema_type_kind_label(*type_kind),
+            from,
+            to
+        ),
+        SchemaMigrationStep::AddProperty {
+            type_kind,
+            type_name,
+            property_name,
+            property_type,
+        } => format!(
+            "add property '{}.{}' ({}) on {} '{}'",
+            type_name,
+            property_name,
+            render_prop_type(property_type),
+            schema_type_kind_label(*type_kind),
+            type_name
+        ),
+        SchemaMigrationStep::RenameProperty {
+            type_kind,
+            type_name,
+            from,
+            to,
+        } => format!(
+            "rename property '{}.{}' -> '{}.{}' on {} '{}'",
+            type_name,
+            from,
+            type_name,
+            to,
+            schema_type_kind_label(*type_kind),
+            type_name
+        ),
+        SchemaMigrationStep::AddConstraint {
+            type_kind,
+            type_name,
+            constraint,
+        } => format!(
+            "add constraint {} on {} '{}'",
+            render_constraint(constraint),
+            schema_type_kind_label(*type_kind),
+            type_name
+        ),
+        SchemaMigrationStep::UpdateTypeMetadata {
+            type_kind,
+            name,
+            annotations,
+        } => format!(
+            "update metadata on {} '{}' ({})",
+            schema_type_kind_label(*type_kind),
+            name,
+            render_annotations(annotations)
+        ),
+        SchemaMigrationStep::UpdatePropertyMetadata {
+            type_kind,
+            type_name,
+            property_name,
+            annotations,
+        } => format!(
+            "update metadata on property '{}.{}' of {} '{}' ({})",
+            type_name,
+            property_name,
+            schema_type_kind_label(*type_kind),
+            type_name,
+            render_annotations(annotations)
+        ),
+        SchemaMigrationStep::DropType {
+            type_kind,
+            name,
+            mode,
+        } => format!(
+            "drop {} type '{}' ({} mode)",
+            schema_type_kind_label(*type_kind),
+            name,
+            drop_mode_label(*mode),
+        ),
+        SchemaMigrationStep::DropProperty {
+            type_kind,
+            type_name,
+            property_name,
+            mode,
+        } => format!(
+            "drop property '{}.{}' of {} '{}' ({} mode)",
+            type_name,
+            property_name,
+            schema_type_kind_label(*type_kind),
+            type_name,
+            drop_mode_label(*mode),
+        ),
+        SchemaMigrationStep::UnsupportedChange { entity, reason, .. } => {
+            // When a schema-lint code is attached, render code + tier
+            // so operators see at-a-glance the kind of risk (destructive
+            // / validated / safe) — not just the rule identifier.
+            // Reach the diagnostic via the `diagnostic()` helper so the
+            // CLI doesn't need to know how the lookup works.
+            match step.diagnostic() {
+                Some(diag) => format!(
+                    "unsupported change on {} [{}, {}]: {}",
+                    entity,
+                    diag.code,
+                    schema_lint_tier_label(diag.tier),
+                    reason,
+                ),
+                None => format!("unsupported change on {}: {}", entity, reason),
+            }
+        }
+    }
+}
+
+pub(crate) fn schema_type_kind_label(kind: omnigraph_compiler::SchemaTypeKind) -> &'static str {
+    match kind {
+        omnigraph_compiler::SchemaTypeKind::Interface => "interface",
+        omnigraph_compiler::SchemaTypeKind::Node => "node",
+        omnigraph_compiler::SchemaTypeKind::Edge => "edge",
+    }
+}
+
+pub(crate) fn schema_lint_tier_label(tier: omnigraph_compiler::SafetyTier) -> &'static str {
+    match tier {
+        omnigraph_compiler::SafetyTier::Safe => "safe",
+        omnigraph_compiler::SafetyTier::Validated => "validated",
+        omnigraph_compiler::SafetyTier::Destructive => "destructive",
+    }
+}
+
+pub(crate) fn drop_mode_label(mode: omnigraph_compiler::DropMode) -> &'static str {
+    match mode {
+        omnigraph_compiler::DropMode::Soft => "soft",
+        omnigraph_compiler::DropMode::Hard => "hard",
+    }
+}
+
+pub(crate) fn render_prop_type(prop_type: &omnigraph_compiler::PropType) -> String {
+    let base = if let Some(values) = &prop_type.enum_values {
+        format!("Enum({})", values.join("|"))
+    } else {
+        prop_type.scalar.to_string()
+    };
+    let base = if prop_type.list {
+        format!("[{}]", base)
+    } else {
+        base
+    };
+    if prop_type.nullable {
+        format!("{}?", base)
+    } else {
+        base
+    }
+}
+
+pub(crate) fn render_constraint(constraint: &omnigraph_compiler::schema::ast::Constraint) -> String {
+    match constraint {
+        omnigraph_compiler::schema::ast::Constraint::Key(columns) => {
+            format!("@key({})", columns.join(", "))
+        }
+        omnigraph_compiler::schema::ast::Constraint::Unique(columns) => {
+            format!("@unique({})", columns.join(", "))
+        }
+        omnigraph_compiler::schema::ast::Constraint::Index(columns) => {
+            format!("@index({})", columns.join(", "))
+        }
+        omnigraph_compiler::schema::ast::Constraint::Range { property, min, max } => {
+            format!("@range({}, {:?}, {:?})", property, min, max)
+        }
+        omnigraph_compiler::schema::ast::Constraint::Check { property, pattern } => {
+            format!("@check({}, {:?})", property, pattern)
+        }
+    }
+}
+
+pub(crate) fn render_annotations(annotations: &[omnigraph_compiler::schema::ast::Annotation]) -> String {
+    annotations
+        .iter()
+        .map(|annotation| match &annotation.value {
+            Some(value) => format!("@{}({})", annotation.name, value),
+            None => format!("@{}", annotation.name),
+        })
+        .collect::<Vec<_>>()
+        .join(", ")
+}
+
+pub(crate) fn print_embed_human(output: &EmbedOutput) {
+    println!(
+        "embedded {} rows (selected {}, cleaned {}) from {} -> {} [{} {}d]",
+        output.embedded_rows,
+        output.selected_rows,
+        output.cleaned_rows,
+        output.input,
+        output.output,
+        output.mode,
+        output.dimension
+    );
+}
+
+pub(crate) fn print_snapshot_human(branch: &str, manifest_version: u64, entries: &[SnapshotTableOutput]) {
+    println!("branch: {}", branch);
+    println!("manifest_version: {}", manifest_version);
+    for entry in entries {
+        println!(
+            "{} v{} branch={} rows={}",
+            entry.table_key,
+            entry.table_version,
+            entry.table_branch.as_deref().unwrap_or("main"),
+            entry.row_count
+        );
+    }
+}
+
+pub(crate) fn print_read_output(
+    output: &ReadOutput,
+    format: ReadOutputFormat,
+    config: &OmnigraphConfig,
+) -> Result<()> {
+    println!(
+        "{}",
+        render_read(
+            output,
+            format,
+            &ReadRenderOptions {
+                max_column_width: config.table_max_column_width(),
+                cell_layout: config.table_cell_layout(),
+            },
+        )?
+    );
+    Ok(())
+}
+
+pub(crate) fn print_change_human(output: &ChangeOutput) {
+    println!(
+        "changed {} via {}: {} nodes, {} edges",
+        output.branch, output.query_name, output.affected_nodes, output.affected_edges
+    );
+    if let Some(actor_id) = &output.actor_id {
+        println!("actor_id: {}", actor_id);
+    }
+}
+
+pub(crate) fn print_commit_list_human(commits: &[CommitOutput]) {
+    for commit in commits {
+        let branch = commit.manifest_branch.as_deref().unwrap_or("main");
+        println!(
+            "{} branch={} version={}{}",
+            commit.graph_commit_id,
+            branch,
+            commit.manifest_version,
+            commit
+                .actor_id
+                .as_deref()
+                .map(|actor| format!(" actor={}", actor))
+                .unwrap_or_default()
+        );
+    }
+}
+
+pub(crate) fn print_commit_human(commit: &CommitOutput) {
+    println!("graph_commit_id: {}", commit.graph_commit_id);
+    println!(
+        "manifest_branch: {}",
+        commit.manifest_branch.as_deref().unwrap_or("main")
+    );
+    println!("manifest_version: {}", commit.manifest_version);
+    if let Some(parent_commit_id) = &commit.parent_commit_id {
+        println!("parent_commit_id: {}", parent_commit_id);
+    }
+    if let Some(merged_parent_commit_id) = &commit.merged_parent_commit_id {
+        println!("merged_parent_commit_id: {}", merged_parent_commit_id);
+    }
+    if let Some(actor_id) = &commit.actor_id {
+        println!("actor_id: {}", actor_id);
+    }
+    println!("created_at: {}", commit.created_at);
+}
+
+pub(crate) fn print_policy_explain(decision: &PolicyDecision, actor_id: &str, request: &PolicyRequest) {
+    println!(
+        "decision: {}",
+        if decision.allowed { "allow" } else { "deny" }
+    );
+    println!("actor: {}", actor_id);
+    println!("action: {}", request.action);
+    if let Some(branch) = &request.branch {
+        println!("branch: {}", branch);
+    }
+    if let Some(target_branch) = &request.target_branch {
+        println!("target_branch: {}", target_branch);
+    }
+    if let Some(rule_id) = &decision.matched_rule_id {
+        println!("matched_rule: {}", rule_id);
+    }
+    println!("message: {}", decision.message);
+}
+
+pub(crate) fn yaml_string(value: &str) -> String {
+    format!("'{}'", value.replace('\'', "''"))
+}
+
+#[derive(serde::Serialize)]
+pub(crate) struct QueriesIssue {
+    pub(crate) query: String,
+    pub(crate) message: String,
+}
+
+#[derive(serde::Serialize)]
+pub(crate) struct QueriesValidateOutput {
+    pub(crate) ok: bool,
+    pub(crate) breakages: Vec<QueriesIssue>,
+    pub(crate) warnings: Vec<QueriesIssue>,
+}
+
+#[derive(serde::Serialize)]
+pub(crate) struct QueriesParam {
+    pub(crate) name: String,
+    #[serde(rename = "type")]
+    pub(crate) type_name: String,
+    pub(crate) nullable: bool,
+}
+
+#[derive(serde::Serialize)]
+pub(crate) struct QueriesListItem {
+    pub(crate) name: String,
+    pub(crate) mcp_expose: bool,
+    pub(crate) tool_name: Option<String>,
+    pub(crate) mutation: bool,
+    pub(crate) params: Vec<QueriesParam>,
+}
+
+#[derive(serde::Serialize)]
+pub(crate) struct QueriesListOutput {
+    pub(crate) queries: Vec<QueriesListItem>,
+}