Merge pull request #192 from ModernRelay/refactor/server-modularize

refactor(server): modularize the test monolith and lib.rs — pure code movement

.github/workflows/ci.yml

@@ -351,7 +351,10 @@ jobs:
         run: cargo test --locked -p omnigraph-engine --test s3_storage -- --nocapture
 
       - name: Run RustFS server smoke
-        run: cargo test --locked -p omnigraph-server --test server server_opens_s3_repo_directly_and_serves_snapshot_and_read -- --nocapture
+        # The exact test name (not a loose substring): a filter that matches
+        # nothing passes vacuously, which silently ran zero tests here for a
+        # while (the old filter said s3_repo; the test says s3_graph).
+        run: cargo test --locked -p omnigraph-server --test s3 server_opens_s3_graph_directly_and_serves_snapshot_and_read -- --nocapture
 
       - name: Run RustFS CLI smoke
         run: cargo test --locked -p omnigraph-cli --test system_local local_cli_s3_end_to_end_init_load_read_flow -- --nocapture

crates/omnigraph-server/src/settings.rs

@@ -0,0 +1,988 @@
+//! Server settings: omnigraph.yaml/CLI/env resolution, mode inference
+//! (single vs multi vs cluster), bearer-token sources, and runtime-state
+//! classification (moved verbatim from lib.rs in the modularization).
+
+use super::*;
+
+/// Build serving settings from a cluster directory's applied revision
+/// (RFC-005 §D2): graphs at derived roots, stored queries from verified
+/// catalog blob content, policy bundles from blob paths with their applied
+/// bindings. Always multi-graph routing. The unauthenticated/env handling
+/// matches the omnigraph.yaml path.
+pub(crate) async fn load_cluster_settings(
+    cluster_dir: &PathBuf,
+    cli_bind: Option<String>,
+    cli_allow_unauthenticated: bool,
+) -> Result<ServerConfig> {
+    let snapshot = omnigraph_cluster::read_serving_snapshot(cluster_dir).await.map_err(|diagnostics| {
+        let details = diagnostics
+            .iter()
+            .map(|diagnostic| format!("[{}] {}: {}", diagnostic.code, diagnostic.path, diagnostic.message))
+            .collect::<Vec<_>>()
+            .join("\n  ");
+        eyre!("the cluster at '{}' is not ready to serve:\n  {details}", cluster_dir.display())
+    })?;
+
+    // Bindings -> Cedar slots. The serving pipeline loads one bundle per
+    // graph plus one server-level bundle; stacked bundles per scope are a
+    // later slice — refuse loudly rather than silently merging policy.
+    let mut server_policy_file: Option<PathBuf> = None;
+    let mut graph_policy_files: BTreeMap<String, PathBuf> = BTreeMap::new();
+    for policy in &snapshot.policies {
+        for binding in &policy.applies_to {
+            if binding == "cluster" {
+                if server_policy_file.replace(policy.blob_path.clone()).is_some() {
+                    bail!(
+                        "multiple policy bundles bind the cluster scope; cluster-mode serving supports one bundle per scope — split or merge bundles (multi-bundle scopes are a later slice)"
+                    );
+                }
+            } else if let Some(graph_id) = binding.strip_prefix("graph.") {
+                if graph_policy_files
+                    .insert(graph_id.to_string(), policy.blob_path.clone())
+                    .is_some()
+                {
+                    bail!(
+                        "multiple policy bundles bind graph '{graph_id}'; cluster-mode serving supports one bundle per scope — split or merge bundles (multi-bundle scopes are a later slice)"
+                    );
+                }
+            } else {
+                bail!("unrecognized policy binding '{binding}' in the applied revision");
+            }
+        }
+    }
+
+    let mut graphs = Vec::new();
+    for graph in &snapshot.graphs {
+        let specs: Vec<queries::RegistrySpec> = snapshot
+            .queries
+            .iter()
+            .filter(|query| query.graph_id == graph.graph_id)
+            .map(|query| queries::RegistrySpec {
+                name: query.name.clone(),
+                source: query.source.clone(),
+                // The §D5 bridge: the cluster registry has no expose flag
+                // (exposure becomes a policy decision in Phase 6) — cluster
+                // mode lists every stored query.
+                expose: true,
+                tool_name: None,
+            })
+            .collect();
+        let registry = QueryRegistry::from_specs(specs).map_err(|errors| {
+            let details = errors
+                .iter()
+                .map(|error| error.to_string())
+                .collect::<Vec<_>>()
+                .join("\n  ");
+            eyre!(
+                "stored queries in the applied revision failed to parse:\n  {details}\nrun `cluster refresh` then `cluster apply`, and restart"
+            )
+        })?;
+        graphs.push(GraphStartupConfig {
+            graph_id: graph.graph_id.clone(),
+            uri: graph.root.to_string_lossy().to_string(),
+            policy_file: graph_policy_files.get(&graph.graph_id).cloned(),
+            queries: registry,
+        });
+    }
+
+    let env_unauth = std::env::var("OMNIGRAPH_UNAUTHENTICATED")
+        .ok()
+        .map(|v| {
+            let trimmed = v.trim();
+            !trimmed.is_empty() && trimmed != "0" && !trimmed.eq_ignore_ascii_case("false")
+        })
+        .unwrap_or(false);
+
+    Ok(ServerConfig {
+        mode: ServerConfigMode::Multi {
+            graphs,
+            config_path: cluster_dir.clone(),
+            server_policy_file,
+        },
+        bind: cli_bind.unwrap_or_else(|| "127.0.0.1:8080".to_string()),
+        allow_unauthenticated: cli_allow_unauthenticated || env_unauth,
+    })
+}
+
+pub async fn load_server_settings(
+    config_path: Option<&PathBuf>,
+    cli_cluster: Option<&PathBuf>,
+    cli_uri: Option<String>,
+    cli_target: Option<String>,
+    cli_bind: Option<String>,
+    cli_allow_unauthenticated: bool,
+) -> Result<ServerConfig> {
+    // Rule 0 (RFC-005): --cluster is an exclusive boot source. It is checked
+    // before anything reads omnigraph.yaml — in cluster mode that file is
+    // never opened, not even the implicit current-directory search.
+    if let Some(cluster_dir) = cli_cluster {
+        if cli_uri.is_some() || cli_target.is_some() || config_path.is_some() {
+            bail!(
+                "--cluster is an exclusive boot source; it cannot combine with a graph URI, --target, or --config (axiom 15: a deployment serves from one source)"
+            );
+        }
+        return load_cluster_settings(cluster_dir, cli_bind, cli_allow_unauthenticated).await;
+    }
+    let config = load_config(config_path)?;
+    let bind = cli_bind.unwrap_or_else(|| config.server_bind().to_string());
+    // Either `--unauthenticated` or `OMNIGRAPH_UNAUTHENTICATED=1` flips
+    // this. Treat any non-empty, non-"0"/"false" string as truthy —
+    // standard 12-factor "any value is true" reading of the env var.
+    let env_unauth = std::env::var("OMNIGRAPH_UNAUTHENTICATED")
+        .ok()
+        .map(|v| {
+            let trimmed = v.trim();
+            !trimmed.is_empty() && trimmed != "0" && !trimmed.eq_ignore_ascii_case("false")
+        })
+        .unwrap_or(false);
+    let allow_unauthenticated = cli_allow_unauthenticated || env_unauth;
+
+    // MR-668 decision 2 — four-rule mode inference matrix.
+    //
+    //   1. CLI `<URI>` positional        → Single (URI = the value)
+    //   2. CLI `--target <name>`         → Single (URI = graphs.<name>.uri)
+    //   3. `server.graph` in config      → Single (URI = graphs.<server.graph>.uri)
+    //   4. `--config` + non-empty `graphs:` + no single-mode selector
+    //                                    → Multi (every entry in `graphs:`)
+    //   5. otherwise                     → error with migration hint
+    //
+    // Rules 1-3 are mutually compatible (CLI URI wins over `--target`
+    // wins over `server.graph`), reusing the existing
+    // `resolve_target_uri` precedence.
+    let has_cli_uri = cli_uri.is_some();
+    let has_cli_target = cli_target.is_some();
+    let has_server_graph = config.server_graph_name().is_some();
+    let has_graphs_map = !config.graphs.is_empty();
+    let has_explicit_config = config_path.is_some();
+
+    let mode = if has_cli_uri || has_cli_target || has_server_graph {
+        // Rules 1, 2, or 3 → Single mode.
+        let raw_uri = config.resolve_target_uri(
+            cli_uri,
+            cli_target.as_deref(),
+            config.server_graph_name(),
+        )?;
+        let uri = normalize_root_uri(&raw_uri).wrap_err_with(|| {
+            format!("normalize single-graph URI '{raw_uri}' from server settings")
+        })?;
+        // Config follows graph IDENTITY, not mode: a bare URI is anonymous
+        // (top-level config); a graph chosen by name uses its per-graph
+        // `graphs.<name>.{policy,queries}`. `resolve_target_uri` already
+        // errored on an unknown name, so a `Some(name)` here is a known graph.
+        let selected: Option<&str> = if has_cli_uri {
+            None
+        } else {
+            cli_target.as_deref().or_else(|| config.server_graph_name())
+        };
+        // A named selection must not leave a populated top-level block
+        // silently unused — refuse boot and point at the per-graph block. The
+        // same rule the CLI selection gate enforces, shared via one helper so
+        // the boot check and `omnigraph queries validate`/`list` can't drift.
+        config.ensure_top_level_blocks_honored(selected)?;
+        // Load + identity-check now (no engine needed); the schema
+        // type-check happens when the engine opens.
+        let policy_file = config.resolve_policy_file_for(selected);
+        let queries = QueryRegistry::load(&config, config.query_entries_for(selected))
+            .map_err(|errs| color_eyre::eyre::eyre!(format_registry_load_errors(&uri, &errs)))?;
+        let graph_id = graph_resource_id_for_selection(selected, &uri);
+        ServerConfigMode::Single {
+            uri,
+            graph_id,
+            policy_file,
+            queries,
+        }
+    } else if has_explicit_config && has_graphs_map {
+        // Multi mode: every graph uses its per-graph block; top-level
+        // policy/queries are never honored, so a populated one is an error.
+        let unhonored = config.populated_top_level_blocks();
+        if !unhonored.is_empty() {
+            bail!(
+                "multi-graph mode: top-level {} {} not honored — each graph uses its own \
+                 `graphs.<graph_id>.…` block. Move per-graph rules there (and any \
+                 `graph_list` policy to `server.policy.file`).",
+                unhonored.join(" and "),
+                if unhonored.len() == 1 { "is" } else { "are" },
+            );
+        }
+        // Rule 4 → Multi mode. Build a startup config per graph.
+        let mut graphs = Vec::with_capacity(config.graphs.len());
+        for (name, target) in &config.graphs {
+            // Validate the graph id can construct a `GraphId` newtype.
+            // Doing this here (not at registry insert) so a malformed
+            // omnigraph.yaml fails at startup with a clear error.
+            GraphId::try_from(name.clone()).map_err(|err| {
+                color_eyre::eyre::eyre!("invalid graph id '{name}' in omnigraph.yaml: {err}")
+            })?;
+            let raw_uri = config.resolve_uri_value(&target.uri);
+            let uri = normalize_root_uri(&raw_uri).wrap_err_with(|| {
+                format!("normalize URI '{raw_uri}' for graph '{name}' in omnigraph.yaml")
+            })?;
+            // Per-graph `queries:`, selected through the shared
+            // `query_entries_for` so server and CLI resolve identically.
+            // Load + identity-check now; the schema type-check happens
+            // when this graph's engine opens.
+            let queries = QueryRegistry::load(&config, config.query_entries_for(Some(name.as_str())))
+                .map_err(|errs| color_eyre::eyre::eyre!(format_registry_load_errors(name, &errs)))?;
+            graphs.push(GraphStartupConfig {
+                graph_id: name.clone(),
+                uri,
+                policy_file: config.resolve_target_policy_file(name),
+                queries,
+            });
+        }
+        let config_path = config_path
+            .cloned()
+            .expect("has_explicit_config implies config_path is Some");
+        let server_policy_file = config.resolve_server_policy_file();
+        ServerConfigMode::Multi {
+            graphs,
+            config_path,
+            server_policy_file,
+        }
+    } else {
+        // Rule 5 → error with migration hint.
+        bail!(
+            "no graph to serve: pass a URI (`omnigraph-server <URI>`), select a target \
+             (`--target <name> --config omnigraph.yaml`), set `server.graph: <name>` in \
+             omnigraph.yaml, or for multi-graph mode add a `graphs:` map to the config \
+             file referenced by `--config`."
+        );
+    };
+
+    Ok(ServerConfig {
+        mode,
+        bind,
+        allow_unauthenticated,
+    })
+}
+
+/// Whether the loaded config will run the server in multi-graph mode.
+/// Useful for the test that constructs `ServerConfig` directly.
+pub fn server_config_is_multi(config: &ServerConfig) -> bool {
+    matches!(config.mode, ServerConfigMode::Multi { .. })
+}
+
+/// MR-723 server runtime state, classified from the three-state matrix
+/// of (bearer tokens configured) × (policy file configured) at startup.
+///
+/// * **Open** — neither tokens nor policy; requires explicit
+///   `allow_unauthenticated`. Effectively a "trust the network" dev
+///   mode. `serve()` refuses to start in this shape without the flag,
+///   so the only way to reach this state at runtime is via deliberate
+///   operator opt-in.
+/// * **DefaultDeny** — tokens configured but no policy file. The
+///   server requires a valid bearer token; once authenticated, every
+///   action except `Read` is denied with 403. Closes the "tokens but
+///   forgot the policy file" trap.
+/// * **PolicyEnabled** — policy file configured and at least one
+///   bearer token configured. Cedar evaluates every authenticated
+///   request. Policy without tokens is rejected at startup —
+///   such a server would 401 every request, which is bug-shaped
+///   rather than feature-shaped (operators wanting "deny all
+///   unauthenticated traffic" should configure tokens plus a
+///   deny-all policy to get meaningful 403s with policy-decision
+///   logging instead).
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum ServerRuntimeState {
+    Open,
+    DefaultDeny,
+    PolicyEnabled,
+}
+
+/// Compute the [`ServerRuntimeState`] from the configured inputs.
+/// Pulled out as a pure function so the matrix is unit-testable
+/// without standing up the full server.
+///
+/// The classifier is the **single source of truth** for "should we
+/// start?" — both `serve()`'s single-mode and multi-mode branches
+/// call this before constructing their `AppState`. Adding a startup
+/// invariant here means both modes enforce it automatically; the
+/// alternative (per-constructor `bail!`) drifts the moment a third
+/// mode is added.
+pub fn classify_server_runtime_state(
+    has_tokens: bool,
+    has_policy: bool,
+    allow_unauthenticated: bool,
+) -> Result<ServerRuntimeState> {
+    match (has_tokens, has_policy, allow_unauthenticated) {
+        (false, false, false) => bail!(
+            "server has no bearer tokens and no policy file configured. This is a fully \
+             open server — pass `--unauthenticated` (or set OMNIGRAPH_UNAUTHENTICATED=1) \
+             if you actually want that, otherwise configure bearer tokens (see \
+             docs/user/server.md) and/or `policy.file` in omnigraph.yaml."
+        ),
+        (false, false, true) => Ok(ServerRuntimeState::Open),
+        (true, false, _) => Ok(ServerRuntimeState::DefaultDeny),
+        (false, true, _) => bail!(
+            "policy file is configured but no bearer tokens — every request would 401 \
+             because no token can ever match. Configure at least one bearer token (see \
+             docs/user/server.md), or remove the policy file. To deny all unauthenticated \
+             traffic deliberately, configure tokens plus a deny-all Cedar rule — that \
+             produces meaningful 403s with policy-decision logging instead of silent 401s."
+        ),
+        (true, true, _) => Ok(ServerRuntimeState::PolicyEnabled),
+    }
+}
+
+pub(crate) fn normalize_bearer_token(value: Option<String>) -> Option<String> {
+    value
+        .map(|value| value.trim().to_string())
+        .filter(|value| !value.is_empty())
+}
+
+pub(crate) fn normalize_bearer_actor(value: String) -> Result<String> {
+    let value = value.trim().to_string();
+    if value.is_empty() {
+        bail!("bearer token actor names must not be blank");
+    }
+    Ok(value)
+}
+
+pub(crate) fn parse_bearer_tokens_json(value: &str) -> Result<Vec<(String, String)>> {
+    let entries: HashMap<String, String> = serde_json::from_str(value)
+        .wrap_err("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON must be a JSON object of actor->token")?;
+    Ok(entries.into_iter().collect())
+}
+
+pub(crate) fn read_bearer_tokens_file(path: &str) -> Result<Vec<(String, String)>> {
+    let contents = fs::read_to_string(path)
+        .wrap_err_with(|| format!("failed to read bearer tokens file at {path}"))?;
+    parse_bearer_tokens_json(&contents)
+        .wrap_err_with(|| format!("failed to parse bearer tokens file at {path}"))
+}
+
+pub(crate) fn validate_bearer_tokens(entries: Vec<(String, String)>) -> Result<Vec<(String, String)>> {
+    let mut seen_actors = HashSet::new();
+    let mut seen_tokens = HashSet::new();
+    let mut normalized = Vec::with_capacity(entries.len());
+
+    for (actor, token) in entries {
+        let actor = normalize_bearer_actor(actor)?;
+        let Some(token) = normalize_bearer_token(Some(token)) else {
+            bail!("bearer token for actor '{actor}' must not be blank");
+        };
+        if !seen_actors.insert(actor.clone()) {
+            bail!("duplicate bearer token actor '{actor}'");
+        }
+        if !seen_tokens.insert(token.clone()) {
+            bail!("duplicate bearer token value configured");
+        }
+        normalized.push((actor, token));
+    }
+
+    normalized.sort_by(|(left, _), (right, _)| left.cmp(right));
+    Ok(normalized)
+}
+
+pub(crate) fn server_bearer_tokens_from_env() -> Result<Vec<(String, String)>> {
+    let mut entries = Vec::new();
+
+    if let Some(token) = normalize_bearer_token(std::env::var("OMNIGRAPH_SERVER_BEARER_TOKEN").ok())
+    {
+        entries.push(("default".to_string(), token));
+    }
+
+    if let Some(path) =
+        normalize_bearer_token(std::env::var("OMNIGRAPH_SERVER_BEARER_TOKENS_FILE").ok())
+    {
+        entries.extend(read_bearer_tokens_file(&path)?);
+    } else if let Some(json) =
+        normalize_bearer_token(std::env::var("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON").ok())
+    {
+        entries.extend(parse_bearer_tokens_json(&json)?);
+    }
+
+    validate_bearer_tokens(entries)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        GraphStartupConfig, ServerConfig, ServerConfigMode, ServerRuntimeState,
+        classify_server_runtime_state, hash_bearer_token, load_server_settings,
+        normalize_bearer_token, parse_bearer_tokens_json, serve, server_bearer_tokens_from_env,
+    };
+    use serial_test::serial;
+    use std::env;
+    use std::fs;
+    use tempfile::tempdir;
+
+    /// `authorize` returns the allow/deny **decision** (`Authz`) and reserves
+    /// `Err` for operational failures, so the invoke handler can hide a denial
+    /// as 404 without also masking a 401/500. Pins each outcome.
+    #[test]
+    fn authorize_splits_decision_from_operational_error() {
+        use super::{Authz, PolicyAction, PolicyCompiler, PolicyConfig, PolicyRequest, ResolvedActor, authorize};
+        use std::sync::Arc;
+
+        fn req(action: PolicyAction) -> PolicyRequest {
+            PolicyRequest { action, branch: None, target_branch: None }
+        }
+        let actor = ResolvedActor::cluster_static(Arc::from("act-alice"));
+
+        // --- No policy engine installed (open / default-deny modes) ---
+        // A server-scoped action is denied in every no-policy state.
+        assert!(matches!(
+            authorize(Some(&actor), None, req(PolicyAction::GraphList)).unwrap(),
+            Authz::Denied(_)
+        ));
+        // Authenticated actor + a non-read per-graph action → default-deny.
+        assert!(matches!(
+            authorize(Some(&actor), None, req(PolicyAction::Change)).unwrap(),
+            Authz::Denied(_)
+        ));
+        // `read` is the one per-graph action permitted without a policy.
+        assert!(matches!(
+            authorize(Some(&actor), None, req(PolicyAction::Read)).unwrap(),
+            Authz::Allowed
+        ));
+        // Open mode (no actor, no policy) → allowed.
+        assert!(matches!(
+            authorize(None, None, req(PolicyAction::Read)).unwrap(),
+            Authz::Allowed
+        ));
+
+        // --- Policy engine installed ---
+        let policy: PolicyConfig = serde_yaml::from_str(
+            "version: 1\n\
+             groups:\n  team: [act-alice]\n\
+             rules:\n  - id: team-read\n    allow:\n      actors: { group: team }\n      actions: [read]\n      branch_scope: any\n",
+        )
+        .unwrap();
+        let engine = PolicyCompiler::compile(&policy, "graph").unwrap();
+
+        // A matched allow rule → Allowed.
+        assert!(matches!(
+            authorize(
+                Some(&actor),
+                Some(&engine),
+                PolicyRequest { action: PolicyAction::Read, branch: Some("main".to_string()), target_branch: None },
+            )
+            .unwrap(),
+            Authz::Allowed
+        ));
+        // Known actor, no matching allow rule → Denied, carrying the decision message.
+        match authorize(
+            Some(&actor),
+            Some(&engine),
+            PolicyRequest { action: PolicyAction::Change, branch: Some("main".to_string()), target_branch: None },
+        )
+        .unwrap()
+        {
+            Authz::Denied(message) => assert!(!message.is_empty(), "a deny carries its decision message"),
+            Authz::Allowed => panic!("change must be denied: only read is allowed"),
+        }
+        // Policy installed but no actor → operational failure (`Err`), NOT a
+        // decision. This is the split that keeps a 401/500 from being masked
+        // as the denial's response in the invoke handler.
+        assert!(
+            authorize(None, Some(&engine), req(PolicyAction::Read)).is_err(),
+            "a missing actor with a policy installed is an operational error, not a deny"
+        );
+    }
+
+    #[test]
+    fn hash_bearer_token_produces_32_byte_output() {
+        let hash = hash_bearer_token("any-token");
+        assert_eq!(hash.len(), 32);
+    }
+
+    /// The single gate both open paths funnel through: it refuses a
+    /// schema breakage (naming the graph label + query), attaches a clean
+    /// registry, and collapses an empty one to `None`. Pure over its args
+    /// (no engine), so it covers the multi-graph path's logic too — the
+    /// only per-path difference is the `label`, asserted here.
+    #[test]
+    fn validate_and_attach_gates_on_schema_and_collapses_empty() {
+        use crate::queries::{QueryRegistry, RegistrySpec};
+        use omnigraph_compiler::catalog::build_catalog;
+        use omnigraph_compiler::schema::parser::parse_schema;
+
+        let schema = parse_schema("node User {\nname: String\n}\n").unwrap();
+        let catalog = build_catalog(&schema).unwrap();
+        let spec = |name: &str, source: &str| RegistrySpec {
+            name: name.to_string(),
+            source: source.to_string(),
+            expose: false,
+            tool_name: None,
+        };
+
+        // Empty registry → nothing attached, no error.
+        let empty =
+            super::validate_and_attach(QueryRegistry::default(), &catalog, "g").unwrap();
+        assert!(empty.is_none());
+
+        // A query that type-checks → attached.
+        let ok = QueryRegistry::from_specs(vec![spec(
+            "find_user",
+            "query find_user() { match { $u: User } return { $u.name } }",
+        )])
+        .unwrap();
+        assert!(super::validate_and_attach(ok, &catalog, "g").unwrap().is_some());
+
+        // A query referencing a type the schema lacks → boot refusal that
+        // names both the graph label and the offending query.
+        let broken = QueryRegistry::from_specs(vec![spec(
+            "ghost",
+            "query ghost() { match { $w: Widget } return { $w.name } }",
+        )])
+        .unwrap();
+        let err = super::validate_and_attach(broken, &catalog, "graph-x").unwrap_err();
+        let msg = err.to_string();
+        assert!(msg.contains("graph-x"), "labels the graph: {msg}");
+        assert!(msg.contains("ghost"), "names the query: {msg}");
+        assert!(msg.contains("schema check"), "mentions the schema check: {msg}");
+    }
+
+    #[test]
+    fn hash_bearer_token_is_deterministic() {
+        assert_eq!(
+            hash_bearer_token("stable-input"),
+            hash_bearer_token("stable-input"),
+        );
+    }
+
+    #[test]
+    fn hash_bearer_token_differs_for_different_inputs() {
+        assert_ne!(hash_bearer_token("token-a"), hash_bearer_token("token-b"));
+    }
+
+    #[test]
+    fn hash_bearer_token_matches_known_sha256_vector() {
+        // SHA-256("abc"). If this ever fails, the hash function was swapped.
+        let hash = hash_bearer_token("abc");
+        let hex: String = hash.iter().map(|b| format!("{:02x}", b)).collect();
+        assert_eq!(
+            hex,
+            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
+        );
+    }
+
+    #[tokio::test]
+    async fn server_settings_load_from_yaml_config() {
+        let temp = tempdir().unwrap();
+        let config = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config,
+            r#"
+graphs:
+  local:
+    uri: /tmp/demo.omni
+server:
+  graph: local
+  bind: 0.0.0.0:9090
+"#,
+        )
+        .unwrap();
+
+        let settings = load_server_settings(Some(&config), None, None, None, None, false).await.unwrap();
+        match &settings.mode {
+            ServerConfigMode::Single { uri, graph_id, .. } => {
+                assert_eq!(uri, "/tmp/demo.omni");
+                assert_eq!(graph_id, "local");
+            }
+            ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
+        }
+        assert_eq!(settings.bind, "0.0.0.0:9090");
+    }
+
+    #[tokio::test]
+    async fn server_settings_cli_flags_override_yaml_config() {
+        let temp = tempdir().unwrap();
+        let config = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config,
+            r#"
+graphs:
+  local:
+    uri: /tmp/demo.omni
+server:
+  graph: local
+  bind: 127.0.0.1:8080
+"#,
+        )
+        .unwrap();
+
+        let settings = load_server_settings(
+            Some(&config),
+            None,
+            Some("/tmp/override.omni".to_string()),
+            None,
+            Some("0.0.0.0:9999".to_string()),
+            false,
+        )
+        .await
+        .unwrap();
+        match &settings.mode {
+            ServerConfigMode::Single { uri, graph_id, .. } => {
+                assert_eq!(uri, "/tmp/override.omni");
+                assert_eq!(graph_id, "/tmp/override.omni");
+            }
+            ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
+        }
+        assert_eq!(settings.bind, "0.0.0.0:9999");
+    }
+
+    #[tokio::test]
+    async fn server_settings_can_resolve_named_target() {
+        let temp = tempdir().unwrap();
+        let config = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config,
+            r#"
+graphs:
+  local:
+    uri: ./demo.omni
+  dev:
+    uri: http://127.0.0.1:8080
+server:
+  graph: local
+  bind: 127.0.0.1:8080
+"#,
+        )
+        .unwrap();
+
+        let settings =
+            load_server_settings(Some(&config), None, None, Some("dev".to_string()), None, false)
+                .await
+                .unwrap();
+        match &settings.mode {
+            ServerConfigMode::Single { uri, graph_id, .. } => {
+                assert_eq!(uri, "http://127.0.0.1:8080");
+                assert_eq!(graph_id, "dev");
+            }
+            ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
+        }
+    }
+
+    #[tokio::test]
+    async fn server_settings_require_uri_from_cli_or_config() {
+        let error = load_server_settings(None, None, None, None, None, false).await.unwrap_err();
+        assert!(
+            error.to_string().contains("no graph to serve"),
+            "expected mode-inference error, got: {error}",
+        );
+    }
+
+    #[test]
+    fn classify_open_requires_explicit_unauthenticated_flag() {
+        // State 1: no tokens, no policy, no flag → refuse to start.
+        let error = classify_server_runtime_state(false, false, false).unwrap_err();
+        let msg = error.to_string();
+        assert!(
+            msg.contains("--unauthenticated"),
+            "expected refusal message mentioning --unauthenticated, got: {msg}"
+        );
+
+        // Same matrix cell but with the flag set → Open mode permitted.
+        assert_eq!(
+            classify_server_runtime_state(false, false, true).unwrap(),
+            ServerRuntimeState::Open
+        );
+    }
+
+    #[test]
+    fn classify_tokens_without_policy_is_default_deny() {
+        // State 2: tokens configured, no policy → DefaultDeny regardless
+        // of the flag (the flag opts into the fully-open dev mode; it
+        // doesn't downgrade default-deny back to open).
+        assert_eq!(
+            classify_server_runtime_state(true, false, false).unwrap(),
+            ServerRuntimeState::DefaultDeny
+        );
+        assert_eq!(
+            classify_server_runtime_state(true, false, true).unwrap(),
+            ServerRuntimeState::DefaultDeny
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn serve_refuses_to_start_with_policy_but_no_tokens_multi_mode() {
+        // Bug 2 from the bot-review pass: multi-mode startup was missing
+        // the "policy requires tokens" check that single-mode enforces.
+        // After centralizing the check in `classify_server_runtime_state`,
+        // both modes get the same enforcement. This test guards the
+        // multi-mode propagation path.
+        //
+        // Sibling test below pins single mode. Together they pin that
+        // the classifier is called from both branches of `serve()`.
+        let _guard = EnvGuard::set(&[
+            ("OMNIGRAPH_SERVER_BEARER_TOKEN", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_FILE", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_AWS_SECRET", None),
+            ("OMNIGRAPH_UNAUTHENTICATED", None),
+        ]);
+        let temp = tempdir().unwrap();
+        // The classifier reads `has_policy_configured` from the config
+        // shape (does the Option contain a path?), not from file
+        // existence, so we can hand it a path without writing a real
+        // policy file — the bail fires before policy load.
+        let policy_path = temp.path().join("server-policy.yaml");
+        let config = ServerConfig {
+            mode: ServerConfigMode::Multi {
+                graphs: vec![GraphStartupConfig {
+                    graph_id: "alpha".to_string(),
+                    uri: temp
+                        .path()
+                        .join("alpha.omni")
+                        .to_string_lossy()
+                        .into_owned(),
+                    policy_file: None,
+                    queries: crate::queries::QueryRegistry::default(),
+                }],
+                config_path: temp.path().join("omnigraph.yaml"),
+                server_policy_file: Some(policy_path),
+            },
+            bind: "127.0.0.1:0".to_string(),
+            allow_unauthenticated: false,
+        };
+        let result = serve(config).await;
+        let err = result
+            .expect_err("serve should refuse to start in multi mode with policy but no tokens");
+        let msg = format!("{:?}", err);
+        assert!(
+            msg.contains("policy file is configured but no bearer tokens"),
+            "expected policy-without-tokens rejection in multi mode, got: {msg}",
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn serve_refuses_to_start_in_state_1_without_unauthenticated() {
+        // MR-723 PR A: pin the integration boundary that the classifier
+        // is actually called by `serve()` before any side-effecting
+        // work (Lance dataset open, TcpListener::bind). The classifier
+        // itself is unit-tested above; this test guards the propagation
+        // path from `classify_server_runtime_state` through serve's
+        // `?` so a future refactor that drops the call returns red.
+        //
+        // Marked `#[serial]` because we have to clear all bearer-token
+        // env vars, and another test in this module setting any of them
+        // concurrently would corrupt the read inside `resolve_token_source`.
+        let _guard = EnvGuard::set(&[
+            ("OMNIGRAPH_SERVER_BEARER_TOKEN", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_FILE", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON", None),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_AWS_SECRET", None),
+            ("OMNIGRAPH_UNAUTHENTICATED", None),
+        ]);
+        let temp = tempdir().unwrap();
+        // Graph path doesn't need to exist — classifier fires before
+        // `AppState::open_with_bearer_tokens_and_policy`.
+        let config = ServerConfig {
+            mode: ServerConfigMode::Single {
+                uri: temp
+                    .path()
+                    .join("graph.omni")
+                    .to_string_lossy()
+                    .into_owned(),
+                graph_id: "default".to_string(),
+                policy_file: None,
+                queries: crate::queries::QueryRegistry::default(),
+            },
+            bind: "127.0.0.1:0".to_string(),
+            allow_unauthenticated: false,
+        };
+        let result = serve(config).await;
+        let err =
+            result.expect_err("serve should refuse to start in State 1 without --unauthenticated");
+        let msg = format!("{:?}", err);
+        assert!(
+            msg.contains("no bearer tokens") || msg.contains("policy file"),
+            "expected refusal message naming the misconfiguration, got: {msg}",
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn unauthenticated_env_var_classification() {
+        // MR-723 PR A: closes the gap where the env-var read path inside
+        // `load_server_settings` was structurally implemented but not
+        // exercised by any test. Three properties to pin, all in one
+        // sequential test because `cargo test` runs the mod test suite
+        // in parallel and `OMNIGRAPH_UNAUTHENTICATED` is process-global
+        // — interleaving with another test that sets the same env var
+        // (concurrent classifier tests, even the bearer-token suite
+        // sharing `EnvGuard`) corrupts the read. Sequential within one
+        // test fn is the simplest race-free shape.
+        let temp = tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  local:
+    uri: /tmp/demo-unauth.omni
+server:
+  graph: local
+"#,
+        )
+        .unwrap();
+
+        // Truthy values flip Open mode on, even with CLI flag off.
+        for value in ["1", "true", "yes", "TRUE", "anything"] {
+            let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some(value))]);
+            let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
+                .expect("settings load should succeed");
+            assert!(
+                settings.allow_unauthenticated,
+                "OMNIGRAPH_UNAUTHENTICATED={value:?} should enable Open mode",
+            );
+        }
+
+        // Falsy values keep refusal behavior, even with CLI flag off.
+        for value in ["0", "false", "FALSE", ""] {
+            let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some(value))]);
+            let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
+                .expect("settings load should succeed");
+            assert!(
+                !settings.allow_unauthenticated,
+                "OMNIGRAPH_UNAUTHENTICATED={value:?} should NOT enable Open mode",
+            );
+        }
+
+        // Unset env var: also false.
+        let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", None)]);
+        let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
+            .expect("settings load should succeed");
+        assert!(
+            !settings.allow_unauthenticated,
+            "OMNIGRAPH_UNAUTHENTICATED unset should NOT enable Open mode",
+        );
+        drop(_guard);
+
+        // CLI flag wins even when env is falsy — `serve()` honors the
+        // OR of both inputs.
+        let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some("0"))]);
+        let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await
+            .expect("settings load should succeed");
+        assert!(
+            settings.allow_unauthenticated,
+            "--unauthenticated CLI flag should win even when env is falsy",
+        );
+    }
+
+    #[test]
+    fn classify_policy_enabled_requires_tokens() {
+        // State 3: tokens + policy → PolicyEnabled, regardless of the
+        // `allow_unauthenticated` flag (Cedar evaluates the bearer,
+        // the flag is moot once tokens exist).
+        assert_eq!(
+            classify_server_runtime_state(true, true, false).unwrap(),
+            ServerRuntimeState::PolicyEnabled
+        );
+        assert_eq!(
+            classify_server_runtime_state(true, true, true).unwrap(),
+            ServerRuntimeState::PolicyEnabled
+        );
+    }
+
+    #[test]
+    fn classify_policy_without_tokens_is_rejected() {
+        // Closes the "policy installed but no tokens → silent 401 on
+        // every request" footgun. The same shape that single-mode
+        // `open_with_bearer_tokens_and_policy` used to bail on
+        // privately is now rejected by the classifier so both single
+        // and multi mode get the same enforcement from one source of
+        // truth.
+        for allow_unauthenticated in [false, true] {
+            let err =
+                classify_server_runtime_state(false, true, allow_unauthenticated).unwrap_err();
+            let msg = err.to_string();
+            assert!(
+                msg.contains("policy file is configured but no bearer tokens"),
+                "expected policy-without-tokens rejection message; got: {msg}"
+            );
+            assert!(
+                msg.contains("every request would 401"),
+                "rejection message must name the failure mode; got: {msg}"
+            );
+        }
+    }
+
+    #[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")
+        );
+    }
+
+    struct EnvGuard {
+        saved: Vec<(&'static str, Option<String>)>,
+    }
+
+    impl EnvGuard {
+        fn set(vars: &[(&'static str, Option<&str>)]) -> Self {
+            let saved = vars
+                .iter()
+                .map(|(name, _)| (*name, env::var(name).ok()))
+                .collect::<Vec<_>>();
+            for (name, value) in vars {
+                unsafe {
+                    match value {
+                        Some(value) => env::set_var(name, value),
+                        None => env::remove_var(name),
+                    }
+                }
+            }
+            Self { saved }
+        }
+    }
+
+    impl Drop for EnvGuard {
+        fn drop(&mut self) {
+            for (name, value) in self.saved.drain(..) {
+                unsafe {
+                    match value {
+                        Some(value) => env::set_var(name, value),
+                        None => env::remove_var(name),
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn parse_bearer_tokens_json_reads_actor_token_map() {
+        let tokens = parse_bearer_tokens_json(r#"{"alice":" token-a ","bob":"token-b"}"#).unwrap();
+        assert_eq!(tokens.len(), 2);
+        assert!(tokens.contains(&("alice".to_string(), " token-a ".to_string())));
+        assert!(tokens.contains(&("bob".to_string(), "token-b".to_string())));
+    }
+
+    #[test]
+    #[serial]
+    fn server_bearer_tokens_from_env_reads_legacy_token_and_token_file() {
+        let temp = tempdir().unwrap();
+        let tokens_path = temp.path().join("tokens.json");
+        fs::write(
+            &tokens_path,
+            r#"{"team-01":"token-one","team-02":"token-two"}"#,
+        )
+        .unwrap();
+
+        let _guard = EnvGuard::set(&[
+            ("OMNIGRAPH_SERVER_BEARER_TOKEN", Some(" legacy-token ")),
+            (
+                "OMNIGRAPH_SERVER_BEARER_TOKENS_FILE",
+                Some(tokens_path.to_str().unwrap()),
+            ),
+            ("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON", None),
+        ]);
+
+        let tokens = server_bearer_tokens_from_env().unwrap();
+        assert_eq!(
+            tokens,
+            vec![
+                ("default".to_string(), "legacy-token".to_string()),
+                ("team-01".to_string(), "token-one".to_string()),
+                ("team-02".to_string(), "token-two".to_string()),
+            ]
+        );
+    }
+}

crates/omnigraph-server/tests/auth_policy.rs

@@ -0,0 +1,915 @@
+//! Bearer auth, actor resolution, Cedar policy decisions, admission.
+//! Moved verbatim from tests/server.rs in the modularization.
+
+use std::env;
+use std::fs;
+use std::sync::Arc;
+
+use axum::body::Body;
+use axum::http::header::AUTHORIZATION;
+use axum::http::{Method, Request, StatusCode};
+use omnigraph::db::{Omnigraph, ReadTarget};
+use omnigraph::error::OmniError;
+use omnigraph::loader::LoadMode;
+use omnigraph_server::api::{
+    BranchCreateRequest, BranchMergeRequest, ChangeRequest, ErrorOutput, ExportRequest, ReadRequest, SchemaApplyRequest,
+};
+use omnigraph_server::{AppState, build_app};
+use serde_json::{Value, json};
+use tower::ServiceExt;
+
+
+mod support;
+use support::*;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn healthz_succeeds_after_startup() {
+    let (_temp, app) = app_for_loaded_graph().await;
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/healthz")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body["status"], "ok");
+    assert_eq!(body["version"], env!("CARGO_PKG_VERSION"));
+    match option_env!("OMNIGRAPH_SOURCE_VERSION") {
+        Some(source_version) => assert_eq!(body["source_version"], source_version),
+        None => assert!(body.get("source_version").is_none()),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn protected_routes_require_bearer_token() {
+    let (_temp, app) = app_for_loaded_graph_with_auth("demo-token").await;
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+
+    let error: ErrorOutput = serde_json::from_value(body).unwrap();
+    assert_eq!(status, StatusCode::UNAUTHORIZED);
+    assert_eq!(
+        error.code,
+        Some(omnigraph_server::api::ErrorCode::Unauthorized)
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn protected_routes_accept_valid_bearer_token_while_healthz_stays_open() {
+    let (_temp, app) = app_for_loaded_graph_with_auth("demo-token").await;
+
+    let health = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .uri("/healthz")
+                .method(Method::GET)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(health.status(), StatusCode::OK);
+
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches")
+            .method(Method::GET)
+            .header("authorization", "Bearer demo-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert!(body["branches"].is_array());
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn protected_routes_accept_any_configured_team_bearer_token() {
+    let (_temp, app) = app_for_loaded_graph_with_auth_tokens(&[
+        ("team-01", "token-one"),
+        ("team-02", "token-two"),
+    ])
+    .await;
+
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-two")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert!(body["branches"].is_array());
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn bearer_token_resolves_to_correct_actor_for_policy_decisions() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(
+        &policy_path,
+        r#"
+version: 1
+groups:
+  readers: [act-a]
+  writers: [act-b]
+protected_branches: [main]
+rules:
+  - id: readers-only
+    allow:
+      actors: { group: readers }
+      actions: [read]
+      branch_scope: any
+"#,
+    )
+    .unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![
+            ("act-a".to_string(), "token-a".to_string()),
+            ("act-b".to_string(), "token-b".to_string()),
+        ],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    // act-a is authenticated AND authorized.
+    let (ok_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-a")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(ok_status, StatusCode::OK);
+
+    // act-b is authenticated but policy rejects — proves the resolved actor
+    // (not some default) was the policy subject.
+    let (denied_status, denied_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-b")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let denied_error: ErrorOutput = serde_json::from_value(denied_body).unwrap();
+    assert_eq!(denied_status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        denied_error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden)
+    );
+
+    // Unknown token: 401, never reaches the policy engine.
+    let (bad_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer wrong-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(bad_status, StatusCode::UNAUTHORIZED);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn actor_id_resolves_from_bearer_token_ignoring_client_supplied_headers() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let policy_path = temp.path().join("policy.yaml");
+    // Same readers/writers split as
+    // `bearer_token_resolves_to_correct_actor_for_policy_decisions` —
+    // `act-a` can read main, `act-b` cannot. The asymmetry is what
+    // makes the spoof-up/spoof-down distinction observable.
+    fs::write(
+        &policy_path,
+        r#"
+version: 1
+groups:
+  readers: [act-a]
+  writers: [act-b]
+protected_branches: [main]
+rules:
+  - id: readers-only
+    allow:
+      actors: { group: readers }
+      actions: [read]
+      branch_scope: any
+"#,
+    )
+    .unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![
+            ("act-a".to_string(), "token-a".to_string()),
+            ("act-b".to_string(), "token-b".to_string()),
+        ],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    // (1) Spoof-up: bearer for act-b (denied) + X-Actor-Id: act-a (allowed).
+    // If the server were trusting the header, this would succeed as
+    // act-a. The contract is: the bearer wins. Expect 403 because
+    // act-b can't read.
+    let (spoof_up_status, spoof_up_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-b")
+            .header("x-actor-id", "act-a")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let spoof_up_error: ErrorOutput = serde_json::from_value(spoof_up_body).unwrap();
+    assert_eq!(
+        spoof_up_status,
+        StatusCode::FORBIDDEN,
+        "X-Actor-Id must not promote a denied bearer to an allowed actor",
+    );
+    assert_eq!(
+        spoof_up_error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden),
+    );
+
+    // (2) Spoof-down: bearer for act-a (allowed) + X-Actor-Id: act-b (denied).
+    // If the server were trusting the header, this would fail as act-b.
+    // The contract is: the bearer wins. Expect 200 because act-a can read.
+    let (spoof_down_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-a")
+            .header("x-actor-id", "act-b")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(
+        spoof_down_status,
+        StatusCode::OK,
+        "X-Actor-Id must not demote an allowed bearer to a denied actor",
+    );
+
+    // (3) Empty-string spoof attempt: an X-Actor-Id of "" must not
+    // leak through as the policy subject. Same expectation as (1):
+    // bearer for act-b is denied regardless of what the header tries.
+    let (empty_spoof_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-b")
+            .header("x-actor-id", "")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(
+        empty_spoof_status,
+        StatusCode::FORBIDDEN,
+        "empty X-Actor-Id must not clear the resolved actor",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_allows_read_but_distinguishes_401_from_403() {
+    let (_temp, app) = app_for_loaded_graph_with_auth_tokens_and_policy(
+        &[("act-bruno", "team-token"), ("act-ragnor", "admin-token")],
+        POLICY_YAML,
+    )
+    .await;
+
+    let (missing_status, missing_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let missing_error: ErrorOutput = serde_json::from_value(missing_body).unwrap();
+    assert_eq!(missing_status, StatusCode::UNAUTHORIZED);
+    assert_eq!(
+        missing_error.code,
+        Some(omnigraph_server::api::ErrorCode::Unauthorized)
+    );
+
+    let (snapshot_status, snapshot_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer team-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(snapshot_status, StatusCode::OK);
+    assert_eq!(snapshot_body["branch"], "main");
+
+    let export_request = ExportRequest {
+        branch: Some("main".to_string()),
+        type_names: Vec::new(),
+        table_keys: Vec::new(),
+    };
+    let (forbidden_status, forbidden_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/export")
+            .method(Method::POST)
+            .header("authorization", "Bearer team-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&export_request).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    let forbidden_error: ErrorOutput = serde_json::from_value(forbidden_body).unwrap();
+    assert_eq!(forbidden_status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        forbidden_error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden)
+    );
+
+    let response = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .uri("/export")
+                .method(Method::POST)
+                .header("authorization", "Bearer admin-token")
+                .header("content-type", "application/json")
+                .body(Body::from(serde_json::to_vec(&export_request).unwrap()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_uses_resolved_branch_for_snapshot_reads() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let snapshot_id = {
+        let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+        db.resolve_snapshot("main").await.unwrap().to_string()
+    };
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, POLICY_PROTECTED_READ_YAML).unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![("act-bruno".to_string(), "team-token".to_string())],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    let read = ReadRequest {
+        query_source: fs::read_to_string(fixture("test.gq")).unwrap(),
+        query_name: Some("get_person".to_string()),
+        params: Some(json!({ "name": "Alice" })),
+        branch: None,
+        snapshot: Some(snapshot_id),
+    };
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/read")
+            .method(Method::POST)
+            .header("authorization", "Bearer team-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&read).unwrap()))
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body["target"]["branch"], Value::Null);
+    assert_eq!(
+        body["target"]["snapshot"].as_str(),
+        read.snapshot.as_deref()
+    );
+    assert_eq!(body["row_count"], 1);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_blocks_change_on_protected_main_but_allows_unprotected_branch() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    db.branch_create_from(ReadTarget::branch("main"), "feature")
+        .await
+        .unwrap();
+    drop(db);
+
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, POLICY_YAML).unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![("act-bruno".to_string(), "team-token".to_string())],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    let main_change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "Mina", "age": 28 })),
+        branch: Some("main".to_string()),
+    };
+    let (main_status, main_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header("authorization", "Bearer team-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&main_change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    let main_error: ErrorOutput = serde_json::from_value(main_body).unwrap();
+    assert_eq!(main_status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        main_error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden)
+    );
+
+    let feature_change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "Mina", "age": 28 })),
+        branch: Some("feature".to_string()),
+    };
+    let (feature_status, feature_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header("authorization", "Bearer team-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&feature_change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(feature_status, StatusCode::OK);
+    assert_eq!(feature_body["branch"], "feature");
+    assert_eq!(feature_body["affected_nodes"], 1);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_blocks_non_admin_merge_to_main_and_allows_admin() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    db.branch_create_from(ReadTarget::branch("main"), "feature")
+        .await
+        .unwrap();
+    db.load(
+        "feature",
+        r#"{"type":"Person","data":{"name":"Zoe","age":33}}"#,
+        LoadMode::Append,
+    )
+    .await
+    .unwrap();
+    drop(db);
+
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, POLICY_YAML).unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![
+            ("act-bruno".to_string(), "team-token".to_string()),
+            ("act-ragnor".to_string(), "admin-token".to_string()),
+        ],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    let merge = BranchMergeRequest {
+        source: "feature".to_string(),
+        target: Some("main".to_string()),
+    };
+    let (deny_status, deny_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches/merge")
+            .method(Method::POST)
+            .header("authorization", "Bearer team-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&merge).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    let deny_error: ErrorOutput = serde_json::from_value(deny_body).unwrap();
+    assert_eq!(deny_status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        deny_error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden)
+    );
+
+    let (allow_status, allow_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches/merge")
+            .method(Method::POST)
+            .header("authorization", "Bearer admin-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&merge).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(allow_status, StatusCode::OK);
+    assert_eq!(allow_body["actor_id"], "act-ragnor");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn authenticated_change_stamps_actor_on_commits() {
+    // With the Run state machine removed, actor_id is recorded
+    // directly on the commit graph (no intermediate run record).
+    let (_temp, app) = app_for_loaded_graph_with_auth_tokens(&[("act-andrew", "token-one")]).await;
+
+    let change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "Mina", "age": 28 })),
+        branch: Some("main".to_string()),
+    };
+    let (change_status, change_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header("authorization", "Bearer token-one")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(change_status, StatusCode::OK);
+    assert_eq!(change_body["actor_id"], "act-andrew");
+
+    let (commits_status, commits_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/commits?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-one")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(commits_status, StatusCode::OK);
+    let head = commits_body["commits"]
+        .as_array()
+        .unwrap()
+        .last()
+        .expect("head commit should exist");
+    assert_eq!(head["actor_id"], "act-andrew");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
+    let (_temp, app) = app_for_loaded_graph_with_auth_tokens(&[
+        ("act-andrew", "token-one"),
+        ("act-ragnor", "token-two"),
+    ])
+    .await;
+
+    let create = BranchCreateRequest {
+        from: Some("main".to_string()),
+        name: "feature".to_string(),
+    };
+    let (create_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches")
+            .method(Method::POST)
+            .header("authorization", "Bearer token-one")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&create).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(create_status, StatusCode::OK);
+
+    let change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "Zoe", "age": 33 })),
+        branch: Some("feature".to_string()),
+    };
+    let (change_status, _) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header("authorization", "Bearer token-one")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(change_status, StatusCode::OK);
+
+    let merge = BranchMergeRequest {
+        source: "feature".to_string(),
+        target: Some("main".to_string()),
+    };
+    let (merge_status, merge_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches/merge")
+            .method(Method::POST)
+            .header("authorization", "Bearer token-two")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&merge).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(merge_status, StatusCode::OK);
+    assert_eq!(merge_body["actor_id"], "act-ragnor");
+
+    let (commit_status, commit_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/commits?branch=main")
+            .method(Method::GET)
+            .header("authorization", "Bearer token-two")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(commit_status, StatusCode::OK);
+    let head = commit_body["commits"]
+        .as_array()
+        .unwrap()
+        .last()
+        .expect("head commit should exist");
+    assert_eq!(head["actor_id"], "act-ragnor");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn engine_layer_policy_fires_via_direct_arc_omnigraph_from_new_single() {
+    use omnigraph_server::GraphRouting;
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+
+    // Permit `act-allowed` for change actions; `act-blocked` is not in
+    // any allowed group — every change request from them must deny.
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, permit_all_policy_yaml(&["act-allowed"])).unwrap();
+    let policy_engine =
+        omnigraph_server::PolicyEngine::load_graph(&policy_path, graph.to_string_lossy().as_ref())
+            .unwrap();
+
+    let workload = omnigraph_server::workload::WorkloadController::new(100, 1_000_000_000);
+    let state = AppState::new_single(
+        graph.to_string_lossy().to_string(),
+        db,
+        vec![("act-blocked".to_string(), "block-token".to_string())],
+        Some(policy_engine),
+        workload,
+    );
+
+    // Reach into the routing and pull the engine the same way an
+    // embedded consumer holding `Arc<Omnigraph>` would. If `new_single`
+    // failed to apply `with_policy` to the engine, this `mutate_as`
+    // would succeed — the HTTP-layer is bypassed entirely.
+    let handle = match state.routing() {
+        GraphRouting::Single { handle } => Arc::clone(handle),
+        GraphRouting::Multi { .. } => panic!("expected single-mode routing"),
+    };
+    let engine = Arc::clone(&handle.engine);
+
+    let mut params: omnigraph_compiler::ParamMap = Default::default();
+    params.insert(
+        "name".to_string(),
+        omnigraph_compiler::Literal::String("EngineLayerBlocked".to_string()),
+    );
+    params.insert("age".to_string(), omnigraph_compiler::Literal::Integer(30));
+    let result = engine
+        .mutate_as(
+            "main",
+            MUTATION_QUERIES,
+            "insert_person",
+            &params,
+            Some("act-blocked"),
+        )
+        .await;
+    match result {
+        Err(OmniError::Policy(_)) => { /* expected — engine-layer gate fired */ }
+        Ok(_) => panic!(
+            "engine-layer policy did NOT fire — act-blocked successfully ran mutate_as via \
+             the engine pulled from the registry handle. AppState::new_single failed to apply \
+             with_policy to the underlying Omnigraph engine. This is the B2 footgun the \
+             with_policy_engine deletion was supposed to close."
+        ),
+        Err(other) => panic!("expected OmniError::Policy, got: {other:?}"),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn oversized_request_body_returns_payload_too_large() {
+    let (_temp, app) = app_for_loaded_graph().await;
+    let oversized = "x".repeat(1_100_000);
+    let response = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .uri("/read")
+                .method(Method::POST)
+                .header("content-type", "application/json")
+                .body(Body::from(oversized))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn default_deny_mode_allows_read_for_authenticated_actor() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_only(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-andrew", "demo-token")],
+    )
+    .await;
+
+    let (status, _body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot")
+            .method(Method::GET)
+            .header(AUTHORIZATION, "Bearer demo-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn default_deny_mode_rejects_change_with_forbidden() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_only(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-andrew", "demo-token")],
+    )
+    .await;
+
+    let change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "DefaultDeny", "age": 1 })),
+        branch: Some("main".to_string()),
+    };
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header(AUTHORIZATION, "Bearer demo-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    let error: ErrorOutput = serde_json::from_value(body).unwrap();
+    assert!(
+        error.error.contains("default-deny"),
+        "expected default-deny in error message, got: {}",
+        error.error
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn default_deny_mode_rejects_schema_apply_with_forbidden() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_only(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-andrew", "demo-token")],
+    )
+    .await;
+
+    let req = SchemaApplyRequest {
+        schema_source: additive_schema_with_nickname(),
+        ..Default::default()
+    };
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/schema/apply")
+            .method(Method::POST)
+            .header(AUTHORIZATION, "Bearer demo-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&req).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    let error: ErrorOutput = serde_json::from_value(body).unwrap();
+    assert!(
+        error.error.contains("default-deny"),
+        "expected default-deny in error message, got: {}",
+        error.error
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_change_admin_on_main_allowed() {
+    // (act-ragnor, change, main) — admins-change-anywhere rule applies.
+    // Both SDK and HTTP must allow. Each path uses its own fresh graph
+    // because allow→side-effects.
+    let (_t1, graph1, policy1) = build_parity_graph().await;
+    let sdk = sdk_change_decision(&graph1, &policy1, "act-ragnor").await;
+    let (_t2, graph2, policy2) = build_parity_graph().await;
+    let http = http_change_decision(&graph2, &policy2, "act-ragnor", "ragnor-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Allow) && matches!(http, ParityDecision::Allow),
+        "SDK={sdk:?} HTTP={http:?} — should both Allow",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_change_team_on_main_denied() {
+    // (act-bruno, change, main) — no rule grants bruno change on
+    // protected. Both SDK and HTTP must deny. Same graph is reusable
+    // because deny→no side-effects.
+    let (_temp, graph, policy) = build_parity_graph().await;
+    let sdk = sdk_change_decision(&graph, &policy, "act-bruno").await;
+    let http = http_change_decision(&graph, &policy, "act-bruno", "bruno-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Deny) && matches!(http, ParityDecision::Deny),
+        "SDK={sdk:?} HTTP={http:?} — should both Deny",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_branch_merge_admin_allowed() {
+    // (act-ragnor, branch_merge, feature→main) — admins-merge-to-protected
+    // rule applies. Both Allow. Each path uses its own fresh graph —
+    // a successful merge consumes the feature branch's commit on main.
+    let (_t1, graph1, policy1) = build_parity_graph().await;
+    let sdk = sdk_merge_decision(&graph1, &policy1, "act-ragnor").await;
+    let (_t2, graph2, policy2) = build_parity_graph().await;
+    let http = http_merge_decision(&graph2, &policy2, "act-ragnor", "ragnor-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Allow) && matches!(http, ParityDecision::Allow),
+        "SDK={sdk:?} HTTP={http:?} — should both Allow",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_branch_merge_team_denied() {
+    // (act-bruno, branch_merge, feature→main) — no rule grants bruno
+    // branch_merge. Both Deny.
+    let (_temp, graph, policy) = build_parity_graph().await;
+    let sdk = sdk_merge_decision(&graph, &policy, "act-bruno").await;
+    let http = http_merge_decision(&graph, &policy, "act-bruno", "bruno-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Deny) && matches!(http, ParityDecision::Deny),
+        "SDK={sdk:?} HTTP={http:?} — should both Deny",
+    );
+}

crates/omnigraph-server/tests/multi_graph.rs

@@ -0,0 +1,584 @@
+//! Cluster-mode boot and the concurrent branch-ops matrix.
+//! Moved verbatim from tests/server.rs in the modularization.
+
+use std::fs;
+
+use axum::body::{Body, to_bytes};
+use axum::http::{Method, Request, StatusCode};
+use omnigraph_server::api::ErrorOutput;
+use omnigraph_server::{AppState, build_app};
+use serde_json::Value;
+use tower::ServiceExt;
+
+
+mod support;
+use support::*;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn concurrent_branch_ops_morphological_matrix() {
+    // Cell a: Merge × Merge, distinct targets.
+    // Pre-fix on b09a097/22d76db: branch_merge_impl's swap-restore race
+    // landed feature_a's content in target_b instead of target_a (and
+    // vice versa — symmetric swap). Identity asserts catch both
+    // asymmetric and symmetric variants.
+    {
+        let cell = "a:merge×merge:distinct-targets";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "feature-a-cella").await;
+        h.insert_person("feature-a-cella", "EveA-cella", 22).await;
+        h.create_branch("main", "feature-b-cella").await;
+        h.insert_person("feature-b-cella", "FrankB-cella", 33).await;
+        h.create_branch("main", "target-a-cella").await;
+        h.create_branch("main", "target-b-cella").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("feature-a-cella".to_string(), "target-a-cella".to_string()),
+                matrix::op_merge("feature-b-cella".to_string(), "target-b-cella".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] merge a", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] merge b", cell);
+        h.assert_persons("target-a-cella", cell, &["EveA-cella"], &["FrankB-cella"])
+            .await;
+        h.assert_persons("target-b-cella", cell, &["FrankB-cella"], &["EveA-cella"])
+            .await;
+        h.assert_post_op_sentinel(cell, "sentinel-cella").await;
+    }
+
+    // Cell b: Merge × Merge, same target / distinct sources.
+    // Both want to land in main. merge_exclusive serializes; both should
+    // succeed and main should contain BOTH sources' contributions.
+    {
+        let cell = "b:merge×merge:same-target-distinct-sources";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "src-x-cellb").await;
+        h.insert_person("src-x-cellb", "Xavier-cellb", 41).await;
+        h.create_branch("main", "src-y-cellb").await;
+        h.insert_person("src-y-cellb", "Yvonne-cellb", 42).await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("src-x-cellb".to_string(), "main".to_string()),
+                matrix::op_merge("src-y-cellb".to_string(), "main".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] merge x", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] merge y", cell);
+        h.assert_persons("main", cell, &["Xavier-cellb", "Yvonne-cellb"], &[])
+            .await;
+        h.assert_post_op_sentinel(cell, "sentinel-cellb").await;
+    }
+
+    // Cell c: Merge × Merge, same source / distinct targets (fanout).
+    // One source merged into two targets simultaneously. merge_exclusive
+    // serializes; both targets should reflect the source's content.
+    {
+        let cell = "c:merge×merge:same-source-distinct-targets";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "src-shared-cellc").await;
+        h.insert_person("src-shared-cellc", "Sharon-cellc", 50)
+            .await;
+        h.create_branch("main", "tgt-1-cellc").await;
+        h.create_branch("main", "tgt-2-cellc").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("src-shared-cellc".to_string(), "tgt-1-cellc".to_string()),
+                matrix::op_merge("src-shared-cellc".to_string(), "tgt-2-cellc".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] merge into tgt-1", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] merge into tgt-2", cell);
+        h.assert_persons("tgt-1-cellc", cell, &["Sharon-cellc"], &[])
+            .await;
+        h.assert_persons("tgt-2-cellc", cell, &["Sharon-cellc"], &[])
+            .await;
+        h.assert_post_op_sentinel(cell, "sentinel-cellc").await;
+    }
+
+    // Cell d: Merge × Change, both touching main. C2 permits both
+    // succeed, or exactly one clean 409 if the merge detects target
+    // movement after planning but before acquiring the queue.
+    {
+        let cell = "d:merge×change:into-target";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "feature-celld").await;
+        h.insert_person("feature-celld", "EveD-celld", 22).await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("feature-celld".to_string(), "main".to_string()),
+                matrix::op_change_insert("main".to_string(), "FrankD-celld".to_string(), 33),
+            )
+            .await;
+        assert_eq!(sb.status, StatusCode::OK, "[{}] change", cell);
+        assert!(
+            sa.status == StatusCode::OK || sa.status == StatusCode::CONFLICT,
+            "[{}] merge must be 200 or clean 409, got {}",
+            cell,
+            sa.status
+        );
+        if sa.status == StatusCode::OK {
+            h.assert_persons("main", cell, &["EveD-celld", "FrankD-celld"], &[])
+                .await;
+        } else {
+            let error: ErrorOutput = serde_json::from_slice(&sa.body).unwrap();
+            let conflict = error
+                .manifest_conflict
+                .expect("merge 409 must include manifest_conflict");
+            assert_eq!(
+                conflict.table_key, "node:Person",
+                "[{}] conflict table",
+                cell
+            );
+            h.assert_persons("main", cell, &["FrankD-celld"], &["EveD-celld"])
+                .await;
+        }
+        h.assert_post_op_sentinel(cell, "sentinel-celld").await;
+    }
+
+    // Cell e: Merge × BranchCreateFrom-target. Concurrent fork off the
+    // merge target while the merge runs. Both should succeed; the new
+    // branch should have a coherent view (either pre- or post-merge,
+    // both valid). After both, target = main has the merged content.
+    {
+        let cell = "e:merge×branch_create_from:target";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "src-celle").await;
+        h.insert_person("src-celle", "Eve-celle", 22).await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("src-celle".to_string(), "main".to_string()),
+                matrix::op_branch_create("main".to_string(), "fork-celle".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] merge", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] branch_create_from", cell);
+        // Main definitely has Eve.
+        h.assert_persons("main", cell, &["Eve-celle"], &[]).await;
+        // fork-celle was forked off main at SOME version; main's current
+        // count is 5 (4 seeded + Eve). fork-celle has either 4 (pre-merge
+        // snapshot) or 5 (post-merge snapshot); both are valid timings.
+        let fork_count = h.person_count("fork-celle").await;
+        assert!(
+            fork_count == 4 || fork_count == 5,
+            "[{}] fork-celle row count must be pre- or post-merge view (4 or 5), got {}",
+            cell,
+            fork_count
+        );
+        h.assert_post_op_sentinel(cell, "sentinel-celle").await;
+    }
+
+    // Cell f: BranchCreateFrom × BranchCreateFrom, distinct parents.
+    // Pre-fix on f925ad1: swap-restore race in branch_create_from_impl
+    // forked the new branch off the wrong parent. Identity asserts pin
+    // that fork-from-A inherits A's content, fork-from-B inherits B's.
+    {
+        let cell = "f:branch_create_from×branch_create_from:distinct-parents";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "alpha-cellf").await;
+        h.insert_person("alpha-cellf", "Eve-cellf", 22).await;
+        h.create_branch("main", "beta-cellf").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_branch_create("alpha-cellf".to_string(), "gamma-cellf".to_string()),
+                matrix::op_branch_create("beta-cellf".to_string(), "delta-cellf".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] gamma create", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] delta create", cell);
+        // gamma forks off alpha → must contain Eve.
+        h.assert_persons("gamma-cellf", cell, &["Eve-cellf"], &[])
+            .await;
+        // delta forks off beta → must NOT contain Eve.
+        h.assert_persons("delta-cellf", cell, &[], &["Eve-cellf"])
+            .await;
+        h.assert_post_op_sentinel(cell, "sentinel-cellf").await;
+    }
+
+    // Cell g: BranchCreateFrom × BranchDelete, unrelated branches.
+    // Disjoint branches; both should complete cleanly without
+    // interference.
+    {
+        let cell = "g:branch_create_from×branch_delete:unrelated";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "doomed-cellg").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_branch_create("main".to_string(), "newborn-cellg".to_string()),
+                matrix::op_branch_delete("doomed-cellg".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] create newborn", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] delete doomed", cell);
+        // newborn-cellg exists with main's content.
+        h.assert_persons("newborn-cellg", cell, &["Alice"], &[])
+            .await;
+        h.assert_post_op_sentinel(cell, "sentinel-cellg").await;
+    }
+
+    // Cell h: BranchDelete × BranchDelete, distinct branches. Both call
+    // refresh() internally; verify no deadlock and both deletes land.
+    {
+        let cell = "h:branch_delete×branch_delete:distinct";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "doomed1-cellh").await;
+        h.create_branch("main", "doomed2-cellh").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_branch_delete("doomed1-cellh".to_string()),
+                matrix::op_branch_delete("doomed2-cellh".to_string()),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] delete 1", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] delete 2", cell);
+        // Verify both gone via /branches list (snapshot would still work
+        // for a deleted branch via parent fallback in some paths, so we
+        // use the explicit list).
+        let r = h
+            .app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/branches")
+                    .method(Method::GET)
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(r.status(), StatusCode::OK);
+        let body = to_bytes(r.into_body(), usize::MAX).await.unwrap();
+        let list_body: Value = serde_json::from_slice(&body).unwrap();
+        let branches: Vec<&str> = list_body["branches"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .filter_map(|v| v.as_str())
+            .collect();
+        assert!(
+            !branches.contains(&"doomed1-cellh"),
+            "[{}] doomed1 still in branch list: {:?}",
+            cell,
+            branches
+        );
+        assert!(
+            !branches.contains(&"doomed2-cellh"),
+            "[{}] doomed2 still in branch list: {:?}",
+            cell,
+            branches
+        );
+        h.assert_post_op_sentinel(cell, "sentinel-cellh").await;
+    }
+
+    // Cell i: BranchDelete × Change, on a different branch. Delete one
+    // branch while a /change runs on main. Both should succeed.
+    {
+        let cell = "i:branch_delete×change:distinct-branch";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "doomed-celli").await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_branch_delete("doomed-celli".to_string()),
+                matrix::op_change_insert("main".to_string(), "Pat-celli".to_string(), 44),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] delete", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] change", cell);
+        h.assert_persons("main", cell, &["Pat-celli"], &[]).await;
+        h.assert_post_op_sentinel(cell, "sentinel-celli").await;
+    }
+
+    // Cell j: BranchCreateFrom × Change, both on main. The fork timing
+    // determines whether the new branch sees the change (pre or post).
+    // Both valid. Main must contain the inserted row.
+    {
+        let cell = "j:branch_create_from×change:on-source";
+        let h = matrix::Harness::new().await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_branch_create("main".to_string(), "twin-cellj".to_string()),
+                matrix::op_change_insert("main".to_string(), "Quincy-cellj".to_string(), 55),
+            )
+            .await;
+        assert_eq!(sa.status, StatusCode::OK, "[{}] branch_create", cell);
+        assert_eq!(sb.status, StatusCode::OK, "[{}] change", cell);
+        h.assert_persons("main", cell, &["Quincy-cellj"], &[]).await;
+        // twin-cellj has either pre-change view (no Quincy) or
+        // post-change view (with Quincy); either is valid.
+        let twin_has_quincy = h.person_exists("twin-cellj", "Quincy-cellj").await;
+        let _ = twin_has_quincy; // either valid timing — just ensure no panic
+        h.assert_post_op_sentinel(cell, "sentinel-cellj").await;
+    }
+
+    // Cell k: reopen consistency. Run a representative concurrent pair,
+    // drop the engine, reopen on a separate handle, verify state matches.
+    {
+        let cell = "k:reopen-after-pair";
+        let h = matrix::Harness::new().await;
+        h.create_branch("main", "src-cellk").await;
+        h.insert_person("src-cellk", "Rita-cellk", 36).await;
+
+        let (sa, sb) = h
+            .run_pair(
+                matrix::op_merge("src-cellk".to_string(), "main".to_string()),
+                matrix::op_change_insert("main".to_string(), "Steve-cellk".to_string(), 37),
+            )
+            .await;
+        assert_eq!(sb.status, StatusCode::OK, "[{}] change", cell);
+        assert!(
+            sa.status == StatusCode::OK || sa.status == StatusCode::CONFLICT,
+            "[{}] merge must be 200 or clean 409, got {}",
+            cell,
+            sa.status
+        );
+        if sa.status == StatusCode::OK {
+            h.assert_persons("main", cell, &["Rita-cellk", "Steve-cellk"], &[])
+                .await;
+        } else {
+            let error: ErrorOutput = serde_json::from_slice(&sa.body).unwrap();
+            let conflict = error
+                .manifest_conflict
+                .expect("merge 409 must include manifest_conflict");
+            assert_eq!(
+                conflict.table_key, "node:Person",
+                "[{}] conflict table",
+                cell
+            );
+            h.assert_persons("main", cell, &["Steve-cellk"], &["Rita-cellk"])
+                .await;
+        }
+
+        // Reopen via a fresh AppState on the same graph.
+        let graph_uri = format!("{}/server.omni", h._temp.path().display());
+        let reopened = AppState::open(graph_uri.clone()).await.unwrap();
+        let app2 = build_app(reopened);
+        // Sanity: the same identity check via the new app must see
+        // Rita and Steve.
+        let r = app2
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/snapshot?branch=main")
+                    .method(Method::GET)
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(r.status(), StatusCode::OK, "[{}] reopen snapshot", cell);
+        let body = to_bytes(r.into_body(), usize::MAX).await.unwrap();
+        let v: Value = serde_json::from_slice(&body).unwrap();
+        let person_rows = v["tables"]
+            .as_array()
+            .and_then(|tables| {
+                tables
+                    .iter()
+                    .find(|t| t["table_key"].as_str() == Some("node:Person"))
+            })
+            .and_then(|t| t["row_count"].as_u64())
+            .expect("reopen snapshot must include node:Person row_count");
+        let expected_rows = if sa.status == StatusCode::OK { 6 } else { 5 };
+        assert_eq!(
+            person_rows, expected_rows,
+            "[{}] reopened main should include seed (4) + committed concurrent writes",
+            cell,
+        );
+    }
+}
+
+#[tokio::test]
+async fn cluster_boot_serves_applied_state() {
+    let temp = converged_cluster_dir("").await;
+    let settings = cluster_settings(temp.path()).await.unwrap();
+    let omnigraph_server::ServerConfigMode::Multi {
+        graphs,
+        config_path,
+        server_policy_file,
+    } = settings.mode
+    else {
+        panic!("cluster boot must select multi-graph routing");
+    };
+    assert_eq!(graphs.len(), 1);
+    assert_eq!(graphs[0].graph_id, "knowledge");
+    assert!(server_policy_file.is_none());
+
+    let state =
+        omnigraph_server::open_multi_graph_state(graphs, Vec::new(), None, config_path)
+            .await
+            .unwrap();
+    let app = build_app(state);
+
+    // The management surface keeps its closed-by-default contract: without a
+    // cluster-scoped policy bundle there is no server-level Cedar engine, so
+    // GET /graphs refuses even in cluster mode.
+    let (status, body) = json_response(
+        &app,
+        Request::builder().uri("/graphs").body(Body::empty()).unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN, "{body}");
+
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/graphs/knowledge/queries")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    assert!(
+        body["queries"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|q| q["name"] == "find_person"),
+        "{body}"
+    );
+
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/graphs/knowledge/queries/find_person")
+            .header("content-type", "application/json")
+            .body(Body::from(r#"{"params":{"name":"nobody"}}"#))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+}
+
+#[tokio::test]
+async fn cluster_boot_wires_policy_bindings_into_cedar_slots() {
+    let temp = tempfile::tempdir().unwrap();
+    drop(temp);
+    let policy_block = r#"policies:
+  graph_rules:
+    file: ./graph.policy.yaml
+    applies_to: [knowledge]
+  cluster_rules:
+    file: ./cluster.policy.yaml
+    applies_to: [cluster]
+"#;
+    let temp = {
+        let temp = tempfile::tempdir().unwrap();
+        fs::write(
+            temp.path().join("people.pg"),
+            "\nnode Person {\n  name: String @key\n}\n",
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join("people.gq"),
+            "\nquery find_person($name: String) {\n  match { $p: Person { name: $name } }\n  return { $p.name }\n}\n",
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join("graph.policy.yaml"),
+            permit_all_policy_yaml(&["default"]),
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join("cluster.policy.yaml"),
+            permit_all_policy_yaml(&["default"]).replace("protected_branches: [main]\n", "protected_branches: [main]\nkind: server\n"),
+        )
+        .unwrap();
+        fs::write(
+            temp.path().join("cluster.yaml"),
+            format!(
+                r#"
+version: 1
+graphs:
+  knowledge:
+    schema: ./people.pg
+    queries:
+      find_person:
+        file: ./people.gq
+{policy_block}"#
+            ),
+        )
+        .unwrap();
+        let import = omnigraph_cluster::import_config_dir(temp.path()).await;
+        assert!(import.ok, "{:?}", import.diagnostics);
+        let apply = omnigraph_cluster::apply_config_dir(temp.path()).await;
+        assert!(apply.ok && apply.converged, "{:?}", apply.diagnostics);
+        temp
+    };
+
+    let settings = cluster_settings(temp.path()).await.unwrap();
+    let omnigraph_server::ServerConfigMode::Multi {
+        graphs,
+        server_policy_file,
+        ..
+    } = settings.mode
+    else {
+        panic!("cluster boot must select multi-graph routing");
+    };
+    let graph_policy = graphs[0].policy_file.as_ref().expect("graph-bound bundle");
+    assert!(
+        graph_policy
+            .to_string_lossy()
+            .contains("__cluster/resources/policy/graph_rules/"),
+        "{graph_policy:?}"
+    );
+    let server_policy = server_policy_file.expect("cluster-bound bundle");
+    assert!(
+        server_policy
+            .to_string_lossy()
+            .contains("__cluster/resources/policy/cluster_rules/"),
+        "{server_policy:?}"
+    );
+}
+
+#[tokio::test]
+async fn cluster_boot_refusals() {
+    // Mutual exclusion with --config / URI.
+    let temp = converged_cluster_dir("").await;
+    let dir = temp.path().to_path_buf();
+    let err = omnigraph_server::load_server_settings(
+        Some(&dir.join("omnigraph.yaml")),
+        Some(&dir),
+        None,
+        None,
+        None,
+        true,
+    )
+    .await
+    .unwrap_err();
+    assert!(err.to_string().contains("exclusive boot source"), "{err}");
+    let err = omnigraph_server::load_server_settings(
+        None,
+        Some(&dir),
+        Some("file:///tmp/x.omni".to_string()),
+        None,
+        None,
+        true,
+    )
+    .await
+    .unwrap_err();
+    assert!(err.to_string().contains("exclusive boot source"), "{err}");
+
+    // Tampered catalog blob refuses boot with the remedy.
+    let blob_dir = dir.join("__cluster/resources/query/knowledge/find_person");
+    let blob = fs::read_dir(&blob_dir).unwrap().next().unwrap().unwrap().path();
+    fs::write(&blob, "tampered").unwrap();
+    let err = cluster_settings(&dir).await.unwrap_err();
+    assert!(
+        err.to_string().contains("catalog_payload_digest_mismatch"),
+        "{err}"
+    );
+    assert!(err.to_string().contains("cluster refresh"), "{err}");
+
+    // Missing state refuses with the import/apply remedy.
+    let empty = tempfile::tempdir().unwrap();
+    let err = cluster_settings(empty.path()).await.unwrap_err();
+    assert!(err.to_string().contains("cluster_state_missing"), "{err}");
+}

crates/omnigraph-server/tests/s3.rs

@@ -0,0 +1,77 @@
+//! S3-backed single-graph serving (gated on OMNIGRAPH_S3_TEST_BUCKET).
+//! Moved verbatim from tests/server.rs in the modularization.
+
+use std::fs;
+
+use axum::body::Body;
+use axum::http::{Method, Request, StatusCode};
+use omnigraph::db::Omnigraph;
+use omnigraph::loader::{LoadMode, load_jsonl};
+use omnigraph_server::api::ReadRequest;
+use omnigraph_server::{AppState, build_app};
+use serde_json::json;
+
+
+mod support;
+use support::*;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn server_opens_s3_graph_directly_and_serves_snapshot_and_read() {
+    let Some(uri) = s3_test_graph_uri("server") else {
+        eprintln!("skipping s3 server test: OMNIGRAPH_S3_TEST_BUCKET is not set");
+        return;
+    };
+
+    Omnigraph::init(&uri, &fs::read_to_string(fixture("test.pg")).unwrap())
+        .await
+        .unwrap();
+    let mut db = Omnigraph::open(&uri).await.unwrap();
+    load_jsonl(
+        &mut db,
+        &fs::read_to_string(fixture("test.jsonl")).unwrap(),
+        LoadMode::Overwrite,
+    )
+    .await
+    .unwrap();
+
+    let app = build_app(
+        AppState::open_with_bearer_token(uri.clone(), Some("s3-token".to_string()))
+            .await
+            .unwrap(),
+    );
+
+    let (snapshot_status, snapshot_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot")
+            .method(Method::GET)
+            .header("authorization", "Bearer s3-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(snapshot_status, StatusCode::OK);
+    assert!(snapshot_body["tables"].is_array());
+
+    let read = ReadRequest {
+        query_source: fs::read_to_string(fixture("test.gq")).unwrap(),
+        query_name: Some("get_person".to_string()),
+        params: Some(json!({ "name": "Alice" })),
+        branch: Some("main".to_string()),
+        snapshot: None,
+    };
+    let (read_status, read_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/read")
+            .method(Method::POST)
+            .header("authorization", "Bearer s3-token")
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&read).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(read_status, StatusCode::OK);
+    assert_eq!(read_body["row_count"], 1);
+    assert_eq!(read_body["rows"][0]["p.name"], "Alice");
+}

crates/omnigraph-server/tests/schema_routes.rs

@@ -0,0 +1,830 @@
+//! Schema read/apply routes: migrations over HTTP, drift, gating.
+//! Moved verbatim from tests/server.rs in the modularization.
+
+use std::fs;
+
+use axum::body::Body;
+use axum::http::{Method, Request, StatusCode};
+use lance::index::DatasetIndexExt;
+use omnigraph::db::{Omnigraph, ReadTarget};
+use omnigraph::loader::LoadMode;
+use omnigraph_server::api::{
+    ChangeRequest, ErrorOutput, ReadRequest, SchemaApplyRequest, SchemaOutput,
+};
+use omnigraph_server::{AppState, build_app};
+use serde_json::json;
+
+
+mod support;
+use support::*;
+
+#[tokio::test]
+async fn schema_apply_route_updates_graph_for_authorized_admin() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let schema = additive_schema_with_nickname();
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: schema,
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+    let graph = graph_path(temp.path());
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    assert!(
+        reopened.catalog().node_types["Person"]
+            .properties
+            .contains_key("nickname")
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_rejects_stored_query_breakage_before_publish() {
+    let (temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, true)],
+        &[("act-ragnor", "admin-token")],
+        STORED_QUERY_SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: renamed_age_schema(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "body: {payload}");
+    let message = payload["error"].as_str().unwrap_or_default();
+    assert!(
+        message.contains("find_person") && message.contains("schema check"),
+        "registry breakage should name the stored query; body: {payload}"
+    );
+
+    let reopened = Omnigraph::open(graph_path(temp.path()).to_str().unwrap())
+        .await
+        .unwrap();
+    let person = &reopened.catalog().node_types["Person"];
+    assert!(person.properties.contains_key("age"));
+    assert!(!person.properties.contains_key("years"));
+
+    let (invoke_status, invoke_body) = json_response(
+        &app,
+        invoke_request(
+            "find_person",
+            "admin-token",
+            json!({ "params": { "name": "Alice" } }),
+        ),
+    )
+    .await;
+    assert_eq!(invoke_status, StatusCode::OK, "body: {invoke_body}");
+    assert_eq!(invoke_body["row_count"], 1);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_noop_keeps_valid_stored_query_registry() {
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, true)],
+        &[("act-ragnor", "admin-token")],
+        STORED_QUERY_SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: fs::read_to_string(fixture("test.pg")).unwrap(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+    assert_eq!(status, StatusCode::OK, "body: {payload}");
+    assert_eq!(payload["applied"], false);
+}
+
+#[tokio::test]
+async fn schema_apply_route_requires_schema_apply_policy_permission() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: additive_schema_with_nickname(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        payload["code"],
+        serde_json::to_value(omnigraph_server::api::ErrorCode::Forbidden).unwrap()
+    );
+}
+
+#[tokio::test]
+async fn schema_apply_route_requires_bearer_token_when_policy_enabled() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: additive_schema_with_nickname(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::UNAUTHORIZED);
+    assert_eq!(
+        payload["code"],
+        serde_json::to_value(omnigraph_server::api::ErrorCode::Unauthorized).unwrap()
+    );
+}
+
+#[tokio::test]
+async fn schema_apply_route_can_rename_type() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: renamed_person_schema(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+    let graph = graph_path(temp.path());
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    let snapshot = reopened
+        .snapshot_of(ReadTarget::branch("main"))
+        .await
+        .unwrap();
+    assert!(snapshot.entry("node:Human").is_some());
+    assert!(snapshot.entry("node:Person").is_none());
+}
+
+#[tokio::test]
+async fn schema_apply_route_can_rename_property() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: renamed_age_schema(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+    let graph = graph_path(temp.path());
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    let person = &reopened.catalog().node_types["Person"];
+    assert!(person.properties.contains_key("years"));
+    assert!(!person.properties.contains_key("age"));
+}
+
+#[tokio::test]
+async fn schema_apply_route_can_add_index() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let graph = graph_path(temp.path());
+    let before_index_count = {
+        let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+        let snapshot = db.snapshot_of(ReadTarget::branch("main")).await.unwrap();
+        let dataset = snapshot.open("node:Person").await.unwrap();
+        dataset.load_indices().await.unwrap().len()
+    };
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: indexed_name_schema(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    let snapshot = reopened
+        .snapshot_of(ReadTarget::branch("main"))
+        .await
+        .unwrap();
+    let dataset = snapshot.open("node:Person").await.unwrap();
+    let after_index_count = dataset.load_indices().await.unwrap().len();
+    assert!(after_index_count > before_index_count);
+}
+
+#[tokio::test]
+async fn schema_apply_route_rejects_unsupported_plan() {
+    let (_temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: unsupported_schema_change(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::BAD_REQUEST);
+    assert_eq!(
+        payload["code"],
+        serde_json::to_value(omnigraph_server::api::ErrorCode::BadRequest).unwrap()
+    );
+}
+
+#[tokio::test]
+async fn schema_apply_route_rejects_when_non_main_branch_exists() {
+    let temp = init_graph_with_schema(&fs::read_to_string(fixture("test.pg")).unwrap()).await;
+    let graph = graph_path(temp.path());
+    let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    db.branch_create("feature").await.unwrap();
+    drop(db);
+
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, SCHEMA_APPLY_POLICY_YAML).unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![("act-ragnor".to_string(), "admin-token".to_string())],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/schema/apply")
+        .header("content-type", "application/json")
+        .header("authorization", "Bearer admin-token")
+        .body(Body::from(
+            serde_json::to_vec(&SchemaApplyRequest {
+                schema_source: additive_schema_with_nickname(),
+                ..Default::default()
+            })
+            .unwrap(),
+        ))
+        .unwrap();
+    let (status, payload) = json_response(&app, request).await;
+
+    assert_eq!(status, StatusCode::CONFLICT);
+    assert_eq!(
+        payload["code"],
+        serde_json::to_value(omnigraph_server::api::ErrorCode::Conflict).unwrap()
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_drift_returns_conflict_for_snapshot_read_and_change() {
+    let (temp, app) = app_for_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    fs::write(graph.join("_schema.pg"), drifted_test_schema()).unwrap();
+
+    let (snapshot_status, snapshot_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/snapshot?branch=main")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let snapshot_error: ErrorOutput = serde_json::from_value(snapshot_body).unwrap();
+    assert_eq!(snapshot_status, StatusCode::CONFLICT);
+    assert_eq!(
+        snapshot_error.code,
+        Some(omnigraph_server::api::ErrorCode::Conflict)
+    );
+    assert!(
+        snapshot_error
+            .error
+            .contains("schema evolution is locked down in phase 1")
+    );
+
+    let read = ReadRequest {
+        query_source: fs::read_to_string(fixture("test.gq")).unwrap(),
+        query_name: Some("get_person".to_string()),
+        params: Some(json!({ "name": "Alice" })),
+        branch: Some("main".to_string()),
+        snapshot: None,
+    };
+    let (read_status, read_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/read")
+            .method(Method::POST)
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&read).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    let read_error: ErrorOutput = serde_json::from_value(read_body).unwrap();
+    assert_eq!(read_status, StatusCode::CONFLICT);
+    assert_eq!(
+        read_error.code,
+        Some(omnigraph_server::api::ErrorCode::Conflict)
+    );
+    assert!(
+        read_error
+            .error
+            .contains("schema evolution is locked down in phase 1")
+    );
+
+    let change = ChangeRequest {
+        query: MUTATION_QUERIES.to_string(),
+        name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "Mina", "age": 28 })),
+        branch: Some("main".to_string()),
+    };
+    let (change_status, change_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&change).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    let change_error: ErrorOutput = serde_json::from_value(change_body).unwrap();
+    assert_eq!(change_status, StatusCode::CONFLICT);
+    assert_eq!(
+        change_error.code,
+        Some(omnigraph_server::api::ErrorCode::Conflict)
+    );
+    assert!(
+        change_error
+            .error
+            .contains("schema evolution is locked down in phase 1")
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_route_returns_current_source() {
+    let (_temp, app) = app_for_loaded_graph().await;
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/schema")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::OK);
+    let output: SchemaOutput = serde_json::from_value(body).unwrap();
+    assert!(output.schema_source.contains("node Person"));
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_route_requires_bearer_token_when_auth_configured() {
+    let (_temp, app) = app_for_loaded_graph_with_auth("demo-token").await;
+
+    let (missing_status, missing_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/schema")
+            .method(Method::GET)
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let missing_error: ErrorOutput = serde_json::from_value(missing_body).unwrap();
+    assert_eq!(missing_status, StatusCode::UNAUTHORIZED);
+    assert_eq!(
+        missing_error.code,
+        Some(omnigraph_server::api::ErrorCode::Unauthorized)
+    );
+
+    let (ok_status, ok_body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/schema")
+            .method(Method::GET)
+            .header("authorization", "Bearer demo-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(ok_status, StatusCode::OK);
+    let output: SchemaOutput = serde_json::from_value(ok_body).unwrap();
+    assert!(!output.schema_source.is_empty());
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_route_denied_when_actor_lacks_read_permission() {
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let policy_path = temp.path().join("policy.yaml");
+    // Policy grants branch_create only — no read action for act-bruno.
+    fs::write(&policy_path, INGEST_CREATE_ONLY_POLICY_YAML).unwrap();
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        graph.to_string_lossy().to_string(),
+        vec![("act-bruno".to_string(), "team-token".to_string())],
+        Some(&policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+
+    let (status, body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/schema")
+            .method(Method::GET)
+            .header("authorization", "Bearer team-token")
+            .body(Body::empty())
+            .unwrap(),
+    )
+    .await;
+    let error: ErrorOutput = serde_json::from_value(body).unwrap();
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert_eq!(
+        error.code,
+        Some(omnigraph_server::api::ErrorCode::Forbidden)
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_soft_drops_property_via_http() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    // Load a row that has the column we're about to drop.
+    let graph = graph_path(temp.path());
+    {
+        let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+        db.load(
+            "main",
+            r#"{"type":"Person","data":{"name":"PreDrop","age":42}}"#,
+            LoadMode::Append,
+        )
+        .await
+        .unwrap();
+    }
+    let pre_version = manifest_dataset_version(&graph).await;
+
+    let (status, payload) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/schema/apply")
+            .header("content-type", "application/json")
+            .header("authorization", "Bearer admin-token")
+            .body(Body::from(
+                serde_json::to_vec(&SchemaApplyRequest {
+                    schema_source: schema_without_age(),
+                    ..Default::default()
+                })
+                .unwrap(),
+            ))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+
+    // Catalog reflects the drop: `age` is gone from the live schema.
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    assert!(
+        !reopened.catalog().node_types["Person"]
+            .properties
+            .contains_key("age"),
+        "catalog should not contain `age` after drop"
+    );
+
+    // Soft drop preserves the prior version — `age` is still readable
+    // via time travel to the pre-drop manifest version. Mirrors the
+    // SDK-side assertion in `apply_schema_drops_a_nullable_property_softly_preserves_prior_version`.
+    let pre_drop_snapshot = reopened.snapshot_at_version(pre_version).await.unwrap();
+    let pre_drop_ds = pre_drop_snapshot.open("node:Person").await.unwrap();
+    let pre_drop_fields = pre_drop_ds
+        .schema()
+        .fields
+        .iter()
+        .map(|f| f.name.clone())
+        .collect::<Vec<_>>();
+    assert!(
+        pre_drop_fields.iter().any(|f| f == "age"),
+        "soft drop should leave the pre-drop dataset's `age` column \
+         time-travel-reachable; got fields {pre_drop_fields:?}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_soft_drops_node_type_via_http() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let graph = graph_path(temp.path());
+
+    let (status, payload) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/schema/apply")
+            .header("content-type", "application/json")
+            .header("authorization", "Bearer admin-token")
+            .body(Body::from(
+                serde_json::to_vec(&SchemaApplyRequest {
+                    schema_source: schema_without_company(),
+                    ..Default::default()
+                })
+                .unwrap(),
+            ))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    assert!(
+        !reopened.catalog().node_types.contains_key("Company"),
+        "catalog should not contain `Company` after drop"
+    );
+    assert!(
+        !reopened.catalog().edge_types.contains_key("WorksAt"),
+        "catalog should not contain `WorksAt` after cascade"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_hard_drops_property_with_allow_data_loss() {
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let graph = graph_path(temp.path());
+    {
+        let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+        db.load(
+            "main",
+            r#"{"type":"Person","data":{"name":"PreDropHard","age":50}}"#,
+            LoadMode::Append,
+        )
+        .await
+        .unwrap();
+    }
+
+    // Apply with allow_data_loss=true → Hard mode promotion.
+    let (status, payload) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/schema/apply")
+            .header("content-type", "application/json")
+            .header("authorization", "Bearer admin-token")
+            .body(Body::from(
+                serde_json::to_vec(&SchemaApplyRequest {
+                    schema_source: schema_without_age(),
+                    allow_data_loss: true,
+                })
+                .unwrap(),
+            ))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+
+    // Catalog reflects the drop.
+    let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    assert!(
+        !reopened.catalog().node_types["Person"]
+            .properties
+            .contains_key("age"),
+        "catalog should not contain `age` after Hard drop"
+    );
+    // Plan steps should show DropMode::Hard for property drops.
+    let steps = payload["steps"].as_array().expect("steps array");
+    let drop_step = steps
+        .iter()
+        .find(|s| s["kind"] == "drop_property")
+        .expect("plan should include drop_property step");
+    let mode = &drop_step["mode"];
+    assert_eq!(
+        mode, "hard",
+        "expected hard mode under allow_data_loss=true"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_keeps_drops_soft_without_flag() {
+    // Symmetric to the Hard test: same schema change, but no
+    // allow_data_loss flag → drops stay Soft (prior column data
+    // remains time-travel-reachable). Pins the default semantics
+    // against accidental Hard promotion.
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let graph = graph_path(temp.path());
+
+    let (status, payload) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/schema/apply")
+            .header("content-type", "application/json")
+            .header("authorization", "Bearer admin-token")
+            .body(Body::from(
+                serde_json::to_vec(&SchemaApplyRequest {
+                    schema_source: schema_without_age(),
+                    allow_data_loss: false,
+                })
+                .unwrap(),
+            ))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+
+    let steps = payload["steps"].as_array().expect("steps array");
+    let drop_step = steps
+        .iter()
+        .find(|s| s["kind"] == "drop_property")
+        .expect("plan should include drop_property step");
+    let mode = &drop_step["mode"];
+    assert_eq!(mode, "soft", "expected soft mode without allow_data_loss");
+    let _ = graph;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn schema_apply_route_additive_property_preserves_existing_rows() {
+    // SDK suite covers rename and drop data preservation. Additive
+    // AddProperty wasn't pinned with a row-count check anywhere.
+    // Load N rows, apply schema adding nullable property, verify
+    // every row is still readable and the new column is null.
+    let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
+        &fs::read_to_string(fixture("test.pg")).unwrap(),
+        &[("act-ragnor", "admin-token")],
+        SCHEMA_APPLY_POLICY_YAML,
+    )
+    .await;
+    let graph = graph_path(temp.path());
+
+    // Standard fixture data: 4 Persons + 1 Company. Load it.
+    let pre_count = {
+        let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+        db.load(
+            "main",
+            &fs::read_to_string(fixture("test.jsonl")).unwrap(),
+            LoadMode::Append,
+        )
+        .await
+        .unwrap();
+        let snap = db
+            .snapshot_of(omnigraph::db::ReadTarget::branch("main"))
+            .await
+            .unwrap();
+        snap.entry("node:Person").expect("Person").row_count
+    };
+    assert!(pre_count > 0, "fixture should have loaded Person rows");
+
+    let (status, payload) = json_response(
+        &app,
+        Request::builder()
+            .method(Method::POST)
+            .uri("/schema/apply")
+            .header("content-type", "application/json")
+            .header("authorization", "Bearer admin-token")
+            .body(Body::from(
+                serde_json::to_vec(&SchemaApplyRequest {
+                    schema_source: additive_schema_with_nickname(),
+                    ..Default::default()
+                })
+                .unwrap(),
+            ))
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(payload["applied"], true);
+
+    // Row count preserved.
+    let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
+    let snap = db
+        .snapshot_of(omnigraph::db::ReadTarget::branch("main"))
+        .await
+        .unwrap();
+    let post_count = snap.entry("node:Person").expect("Person").row_count;
+    assert_eq!(
+        post_count, pre_count,
+        "AddProperty should preserve row count",
+    );
+}

crates/omnigraph-server/tests/stored_queries.rs

@@ -0,0 +1,329 @@
+//! Stored-query registry boot, /queries listing, and invocation routes.
+//! Moved verbatim from tests/server.rs in the modularization.
+
+
+use axum::body::Body;
+use axum::http::StatusCode;
+use omnigraph_server::AppState;
+use serde_json::json;
+
+
+mod support;
+use support::*;
+
+#[tokio::test]
+async fn server_boots_with_a_valid_stored_query_registry() {
+    // A stored query that type-checks against the fixture schema
+    // (`Person { name, age }`) must let the server boot.
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let registry = stored_query_registry(&[(
+        "find_person",
+        "query find_person($name: String) { match { $p: Person { name: $name } } return { $p.age } }",
+        false,
+    )]);
+    let state = AppState::open_single_with_queries(
+        graph.to_string_lossy().to_string(),
+        vec![],
+        None,
+        registry,
+    )
+    .await;
+    assert!(state.is_ok(), "valid registry should boot: {:?}", state.err());
+}
+
+#[tokio::test]
+async fn server_refuses_boot_on_type_broken_stored_query() {
+    // A stored query referencing a type not in the schema (`Widget`)
+    // must abort boot, naming the offending query.
+    let temp = init_loaded_graph().await;
+    let graph = graph_path(temp.path());
+    let registry = stored_query_registry(&[(
+        "ghost",
+        "query ghost() { match { $w: Widget } return { $w.name } }",
+        false,
+    )]);
+    let result = AppState::open_single_with_queries(
+        graph.to_string_lossy().to_string(),
+        vec![],
+        None,
+        registry,
+    )
+    .await;
+    // `AppState` is not `Debug`, so match rather than `expect_err`.
+    let err = match result {
+        Ok(_) => panic!("type-broken stored query must refuse boot"),
+        Err(err) => err,
+    };
+    let msg = err.to_string();
+    assert!(msg.contains("ghost"), "error should name the broken query: {msg}");
+    assert!(
+        msg.contains("schema check"),
+        "error should mention the schema check: {msg}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_stored_read_returns_rows() {
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, false)],
+        &[("act-invoke", "t-invoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+    let (status, body) = json_response(
+        &app,
+        invoke_request("find_person", "t-invoke", json!({ "params": { "name": "Alice" } })),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+    assert_eq!(body["query_name"], "find_person");
+    assert_eq!(body["row_count"], 1, "Alice is in the fixture; body: {body}");
+    assert!(body["rows"].is_array(), "read envelope shape; body: {body}");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_stored_read_accepts_absent_or_empty_body() {
+    let no_param_query = "query list_people() { match { $p: Person } return { $p.name } }";
+    let (_temp, app) = app_with_stored_queries(
+        &[("list_people", no_param_query, false)],
+        &[("act-invoke", "t-invoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+
+    let (status, body) = json_response(
+        &app,
+        invoke_request_bytes("list_people", "t-invoke", Body::empty(), None),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+    assert_eq!(body["query_name"], "list_people");
+
+    let (status, body) = json_response(
+        &app,
+        invoke_request_bytes(
+            "list_people",
+            "t-invoke",
+            Body::empty(),
+            Some("application/json"),
+        ),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+
+    let (status, body) = json_response(
+        &app,
+        invoke_request_bytes(
+            "list_people",
+            "t-invoke",
+            Body::from("{}"),
+            Some("application/json"),
+        ),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+
+    let (status, body) = json_response(
+        &app,
+        invoke_request_bytes(
+            "list_people",
+            "t-invoke",
+            Body::from("{"),
+            Some("application/json"),
+        ),
+    )
+    .await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "body: {body}");
+    assert!(
+        body["error"]
+            .as_str()
+            .unwrap_or_default()
+            .contains("invalid stored-query invocation body"),
+        "malformed JSON should be rejected as bad request; body: {body}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_stored_mutation_double_gates_on_change() {
+    let specs: &[(&str, &str, bool)] = &[(
+        "add_person",
+        "query add_person($name: String) { insert Person { name: $name } }",
+        false,
+    )];
+    let (_temp, app) = app_with_stored_queries(
+        specs,
+        &[("act-invoke", "t-invoke"), ("act-full", "t-full")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+
+    // Has invoke_query but NOT change → the inner change gate denies (403).
+    let (status, body) = json_response(
+        &app,
+        invoke_request("add_person", "t-invoke", json!({ "params": { "name": "Eve" } })),
+    )
+    .await;
+    assert_eq!(
+        status,
+        StatusCode::FORBIDDEN,
+        "invoke_query without change must 403; body: {body}"
+    );
+
+    // Has invoke_query + change → applied.
+    let (status, body) = json_response(
+        &app,
+        invoke_request("add_person", "t-full", json!({ "params": { "name": "Eve" } })),
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+    assert_eq!(body["affected_nodes"], 1, "body: {body}");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_stored_query_bad_param_is_400() {
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, false)],
+        &[("act-invoke", "t-invoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+    // `name` is declared String; pass a number.
+    let (status, body) = json_response(
+        &app,
+        invoke_request("find_person", "t-invoke", json!({ "params": { "name": 123 } })),
+    )
+    .await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "body: {body}");
+    assert!(
+        body["error"].as_str().unwrap_or_default().contains("name"),
+        "400 should name the offending param; body: {body}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_unknown_query_and_denied_actor_return_identical_404() {
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, false)],
+        &[("act-invoke", "t-invoke"), ("act-noinvoke", "t-noinvoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+
+    // Authorized actor, unknown query name → 404.
+    let (unknown_status, unknown_body) =
+        json_response(&app, invoke_request("does_not_exist", "t-invoke", json!({}))).await;
+    // Denied actor (no invoke_query), real query name → 404.
+    let (denied_status, denied_body) = json_response(
+        &app,
+        invoke_request("find_person", "t-noinvoke", json!({ "params": { "name": "Alice" } })),
+    )
+    .await;
+
+    assert_eq!(unknown_status, StatusCode::NOT_FOUND);
+    assert_eq!(denied_status, StatusCode::NOT_FOUND);
+    assert_eq!(
+        unknown_body, denied_body,
+        "deny must be byte-identical to a missing query (no catalog probing)"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn invoke_query_holder_without_read_sees_403_not_404() {
+    // The 404-hiding is for callers WITHOUT invoke_query. An actor that
+    // HOLDS invoke_query but lacks `read` clears the boundary gate, then the
+    // inner read gate denies → 403 for an EXISTING read query, vs 404 for an
+    // unknown one. Existence is visible to grant-holders by design (the
+    // documented double-gate); this pins that actual contract.
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, false)],
+        &[("act-invokeonly", "t-invokeonly")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+    let (exists_status, _) = json_response(
+        &app,
+        invoke_request("find_person", "t-invokeonly", json!({ "params": { "name": "Alice" } })),
+    )
+    .await;
+    let (absent_status, _) =
+        json_response(&app, invoke_request("does_not_exist", "t-invokeonly", json!({}))).await;
+    assert_eq!(
+        exists_status,
+        StatusCode::FORBIDDEN,
+        "an existing read query the holder can't read → inner-gate 403"
+    );
+    assert_eq!(absent_status, StatusCode::NOT_FOUND, "unknown query still 404s");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn list_queries_returns_only_exposed_with_typed_params() {
+    let (_temp, app) = app_with_stored_queries(
+        &[
+            ("find_person", FIND_PERSON_GQ, true),
+            (
+                "add_person",
+                "query add_person($name: String) { insert Person { name: $name } }",
+                true,
+            ),
+            ("hidden", "query hidden() { match { $p: Person } return { $p.name } }", false),
+        ],
+        &[("act-invoke", "t-invoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+    let (status, body) = json_response(&app, get_request("/queries", "t-invoke")).await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+
+    let entries = body["queries"].as_array().unwrap();
+    let names: Vec<&str> = entries.iter().map(|q| q["name"].as_str().unwrap()).collect();
+    assert!(
+        names.contains(&"find_person") && names.contains(&"add_person"),
+        "exposed queries listed: {names:?}"
+    );
+    assert!(!names.contains(&"hidden"), "non-exposed query hidden from the catalog: {names:?}");
+
+    let fp = entries.iter().find(|q| q["name"] == "find_person").unwrap();
+    assert_eq!(fp["mutation"], false);
+    assert_eq!(fp["tool_name"], "find_person");
+    assert_eq!(fp["params"][0]["name"], "name");
+    assert_eq!(fp["params"][0]["kind"], "string");
+    let ap = entries.iter().find(|q| q["name"] == "add_person").unwrap();
+    assert_eq!(ap["mutation"], true, "stored insert → mutation");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn list_queries_is_read_gated_so_a_non_invoker_can_list() {
+    // The catalog is read-gated (not invoke_query-gated), so a reader who
+    // lacks invoke_query still enumerates the exposed queries — the
+    // documented probe-oracle gap until per-query Cedar filtering lands.
+    let (_temp, app) = app_with_stored_queries(
+        &[("find_person", FIND_PERSON_GQ, true)],
+        &[("act-noinvoke", "t-noinvoke")],
+        INVOKE_POLICY_YAML,
+    )
+    .await;
+    let (status, body) = json_response(&app, get_request("/queries", "t-noinvoke")).await;
+    assert_eq!(status, StatusCode::OK, "read-gated catalog; body: {body}");
+    let names: Vec<&str> = body["queries"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|q| q["name"].as_str().unwrap())
+        .collect();
+    assert!(
+        names.contains(&"find_person"),
+        "a reader lists the catalog despite lacking invoke_query: {names:?}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn list_queries_is_empty_when_no_registry() {
+    let (_temp, app) = app_for_loaded_graph_with_auth("demo-token").await;
+    let (status, body) = json_response(&app, get_request("/queries", "demo-token")).await;
+    assert_eq!(status, StatusCode::OK, "body: {body}");
+    assert!(
+        body["queries"].as_array().unwrap().is_empty(),
+        "no stored-query registry → empty catalog"
+    );
+}