mr-668: multi-graph startup + mode inference (PR 5/10)

PR 5 of the MR-668 multi-graph server work. This is the first PR that
makes multi mode actually usable end-to-end: operators invoking
`omnigraph-server --config omnigraph.yaml` with a non-empty `graphs:`
map and no single-mode selector now get a running multi-graph server.

Mode inference (MR-668 decision 2, four-rule matrix in
`load_server_settings`):
  1. CLI `<URI>` positional        → Single
  2. CLI `--target <name>`         → Single (URI from graphs.<name>)
  3. `server.graph` in config      → Single (URI from graphs.<name>)
  4. `--config` + non-empty `graphs:` + no single-mode selector
                                   → Multi (all entries in `graphs:`)
  5. otherwise                     → error with migration hint

Rule 5's error message names every escape hatch so operators can fix
their invocation without grepping docs.

Config schema extensions:
  - `TargetConfig.policy: PolicySettings` (per-graph Cedar policy file).
    `#[serde(default)]` so existing single-graph YAMLs keep parsing.
  - `ServerDefaults.policy: PolicySettings` (server-level Cedar policy
    for management endpoints — loaded in PR 5, wired into `GET /graphs`
    in PR 6b).
  - `OmnigraphConfig::resolve_target_policy_file(name)` and
    `resolve_server_policy_file()` helpers — both resolve relative to
    the config file's `base_dir`.

Public types added to `omnigraph-server`:
  - `ServerConfigMode { Single { uri, policy_file } | Multi { graphs,
    config_path, server_policy_file } }`.
  - `GraphStartupConfig { graph_id, uri, policy_file }` — one entry
    per graph in multi mode.

`ServerConfig` shape change:
  - WAS: `{ uri: String, bind, policy_file, allow_unauthenticated }`.
  - NOW: `{ mode: ServerConfigMode, bind, allow_unauthenticated }`.
  - Breaking for any code that constructs `ServerConfig` directly.
    `main.rs` is unaffected (uses `load_server_settings`).

`serve()` now forks on `ServerConfig.mode`:
  - Single: existing flow via `AppState::open_with_bearer_tokens_and_policy`.
  - Multi: parallel open via `futures::stream::iter(graphs)
    .map(open_single_graph).buffer_unordered(4).collect()`. Bound 4 is
    a rule-of-thumb for I/O-bound work — at N≤10 this trades startup
    latency for a small amount of concurrent S3/Lance open pressure.
    Fail-fast: first open error aborts startup; in-flight opens drop
    their engine via Arc (Lance datasets close cleanly).

New helper `open_single_graph(GraphStartupConfig)`:
  - Validates `GraphId` per the regex in PR 1.
  - `Omnigraph::open(uri).await` with descriptive error context.
  - Loads per-graph policy file and re-applies it via
    `Omnigraph::with_policy` (engine-layer enforcement, MR-722).
  - Returns `Arc<GraphHandle>` ready for the registry.

Routing middleware bug fix:
  - `Router::nest("/graphs/{graph_id}", inner)` rewrites
    `request.uri().path()` to the inner suffix (e.g. `/snapshot`).
    The previous middleware tried to parse `{graph_id}` from
    `request.uri().path()` and got 400 instead of 200. Fixed by reading
    from `axum::extract::OriginalUri` request extension, which preserves
    the pre-rewrite URI.
  - Caught by the two new tests
    `cluster_routes_dispatch_per_graph_handle` and
    `cluster_route_for_unknown_graph_returns_404`.

Tests (14 new, all passing):
  - Four-rule matrix: one test per branch + the joint case
    `mode_inference_cli_uri_overrides_graphs_map` + the empty-graphs-map
    error case.
  - Per-graph + server-level policy file path resolution.
  - Reserved `GraphId` rejection at startup.
  - End-to-end multi-graph routing: two graphs side by side, each
    cluster route hits the right engine.
  - Unknown graph id under cluster prefix → 404.
  - Flat routes 404 in multi mode.

Inline `ServerConfig` test (`serve_refuses_to_start_in_state_1_without_unauthenticated`)
and three `server_settings_*` tests updated to the new `mode` shape.

Result: 211 server tests green (74 lib + 71 integration + 66 openapi),
MR-731 regression test still pinned and passing.

LOC: +45 config.rs, +281 lib.rs (net), +395 tests/server.rs.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/omnigraph-server/tests/server.rs

@@ -4333,3 +4333,398 @@ async fn schema_apply_route_additive_property_preserves_existing_rows() {
         "AddProperty should preserve row count",
     );
 }
+
+// ─── MR-668 PR 5: multi-graph startup ─────────────────────────────────────
+
+mod multi_graph_startup {
+    use super::*;
+    use omnigraph_server::{
+        GraphHandle, GraphId, GraphKey, ServerConfig, ServerConfigMode, load_server_settings,
+    };
+    use std::sync::Arc;
+
+    async fn build_multi_mode_app(graph_ids: &[&str]) -> (Vec<tempfile::TempDir>, Router) {
+        let mut dirs = Vec::with_capacity(graph_ids.len());
+        let mut handles = Vec::with_capacity(graph_ids.len());
+        for id in graph_ids {
+            let dir = tempfile::tempdir().unwrap();
+            let graph_uri = dir.path().join(id).to_str().unwrap().to_string();
+            let schema = fs::read_to_string(fixture("test.pg")).unwrap();
+            let engine = Omnigraph::init(&graph_uri, &schema).await.unwrap();
+            handles.push(Arc::new(GraphHandle {
+                key: GraphKey::cluster(GraphId::try_from(*id).unwrap()),
+                uri: graph_uri,
+                engine: Arc::new(engine),
+                policy: None,
+            }));
+            dirs.push(dir);
+        }
+        let workload = omnigraph_server::workload::WorkloadController::from_env();
+        let state = AppState::new_multi(handles, Vec::new(), None, workload, None).unwrap();
+        let app = build_app(state);
+        (dirs, app)
+    }
+
+    /// Cluster route `/graphs/{graph_id}/snapshot` resolves to the right
+    /// engine. Two graphs side by side; assert each responds to its own
+    /// id and does NOT respond to the other's URL.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn cluster_routes_dispatch_per_graph_handle() {
+        let (_dirs, app) = build_multi_mode_app(&["alpha", "beta"]).await;
+        for id in ["alpha", "beta"] {
+            let resp = app
+                .clone()
+                .oneshot(
+                    Request::builder()
+                        .method(Method::GET)
+                        .uri(format!("/graphs/{id}/snapshot?branch=main"))
+                        .body(Body::empty())
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            assert_eq!(
+                resp.status(),
+                StatusCode::OK,
+                "graph '{id}' must respond OK on its cluster snapshot route"
+            );
+        }
+    }
+
+    /// Unknown graph id under the cluster prefix yields 404 (not 500,
+    /// not 410 — `Gone` is reserved for the future DELETE flow).
+    #[tokio::test(flavor = "multi_thread")]
+    async fn cluster_route_for_unknown_graph_returns_404() {
+        let (_dirs, app) = build_multi_mode_app(&["alpha"]).await;
+        let resp = app
+            .oneshot(
+                Request::builder()
+                    .method(Method::GET)
+                    .uri("/graphs/nonexistent/snapshot?branch=main")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+    }
+
+    /// Flat routes 404 in multi mode — the router only mounts under
+    /// `/graphs/{graph_id}/...` so `/snapshot` doesn't resolve.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn flat_routes_404_in_multi_mode() {
+        let (_dirs, app) = build_multi_mode_app(&["alpha"]).await;
+        let resp = app
+            .oneshot(
+                Request::builder()
+                    .method(Method::GET)
+                    .uri("/snapshot?branch=main")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+    }
+
+    /// `GraphId` validation runs at startup — a reserved name in
+    /// `omnigraph.yaml` produces a clear error rather than getting
+    /// rejected per-request.
+    #[test]
+    fn load_server_settings_rejects_reserved_graph_id() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  policies:
+    uri: /tmp/g1.omni
+"#,
+        )
+        .unwrap();
+        let err = load_server_settings(Some(&config_path), None, None, None, false).unwrap_err();
+        assert!(
+            err.to_string().contains("invalid graph id 'policies'"),
+            "expected reserved-name rejection, got: {err}"
+        );
+    }
+
+    // ── Four-rule mode inference matrix ───────────────────────────────
+
+    /// Rule 1: CLI positional URI → Single.
+    #[test]
+    fn mode_inference_cli_uri_is_single() {
+        let settings = load_server_settings(
+            None,
+            Some("/tmp/cli.omni".to_string()),
+            None,
+            None,
+            true, // allow unauth so we get past the runtime-state check
+        )
+        .unwrap();
+        match settings.mode {
+            ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/cli.omni"),
+            ServerConfigMode::Multi { .. } => panic!("expected Single (rule 1), got Multi"),
+        }
+    }
+
+    /// Rule 2: --target picks one graph from `graphs:` map → Single.
+    #[test]
+    fn mode_inference_cli_target_is_single() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+  beta:
+    uri: /tmp/beta.omni
+"#,
+        )
+        .unwrap();
+        let settings =
+            load_server_settings(Some(&config_path), None, Some("alpha".into()), None, true)
+                .unwrap();
+        match settings.mode {
+            ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/alpha.omni"),
+            ServerConfigMode::Multi { .. } => panic!("expected Single (rule 2), got Multi"),
+        }
+    }
+
+    /// Rule 3: `server.graph` set → Single (target picked from config).
+    #[test]
+    fn mode_inference_server_graph_is_single() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+  beta:
+    uri: /tmp/beta.omni
+server:
+  graph: beta
+"#,
+        )
+        .unwrap();
+        let settings =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap();
+        match settings.mode {
+            ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/beta.omni"),
+            ServerConfigMode::Multi { .. } => panic!("expected Single (rule 3), got Multi"),
+        }
+    }
+
+    /// Rule 4: `--config` + non-empty `graphs:` + no single-mode selector → Multi.
+    #[test]
+    fn mode_inference_config_plus_graphs_is_multi() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+  beta:
+    uri: /tmp/beta.omni
+"#,
+        )
+        .unwrap();
+        let settings =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap();
+        match settings.mode {
+            ServerConfigMode::Multi { graphs, .. } => {
+                let ids: Vec<&str> = graphs.iter().map(|g| g.graph_id.as_str()).collect();
+                // BTreeMap iteration order is alphabetical.
+                assert_eq!(ids, vec!["alpha", "beta"]);
+            }
+            ServerConfigMode::Single { .. } => panic!("expected Multi (rule 4), got Single"),
+        }
+    }
+
+    /// Rule 5: nothing → error with migration hint.
+    #[test]
+    fn mode_inference_no_inputs_errors_with_migration_hint() {
+        let err = load_server_settings(None, None, None, None, true).unwrap_err();
+        let msg = err.to_string();
+        assert!(
+            msg.contains("no graph to serve"),
+            "expected migration-hint error, got: {msg}"
+        );
+    }
+
+    /// Rule 4 sub-case: `--config` with empty `graphs:` map and no
+    /// single-mode selector → rule 5 fires (no graph to serve).
+    #[test]
+    fn mode_inference_empty_graphs_map_errors() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(&config_path, "server:\n  bind: 127.0.0.1:8080\n").unwrap();
+        let err =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap_err();
+        assert!(err.to_string().contains("no graph to serve"));
+    }
+
+    /// `--config` + `<URI>` together: URI wins → Single (the CLI URI
+    /// takes precedence over the config's graphs map).
+    #[test]
+    fn mode_inference_cli_uri_overrides_graphs_map() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+"#,
+        )
+        .unwrap();
+        let settings = load_server_settings(
+            Some(&config_path),
+            Some("/tmp/cli-override.omni".to_string()),
+            None,
+            None,
+            true,
+        )
+        .unwrap();
+        match settings.mode {
+            ServerConfigMode::Single { uri, .. } => {
+                assert_eq!(
+                    uri, "/tmp/cli-override.omni",
+                    "CLI URI must win over graphs: map"
+                );
+            }
+            ServerConfigMode::Multi { .. } => {
+                panic!("expected Single (CLI URI wins), got Multi")
+            }
+        }
+    }
+
+    /// Per-graph `policy.file` is resolved relative to the config base_dir.
+    #[test]
+    fn per_graph_policy_file_is_resolved_relative_to_base_dir() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+    policy:
+      file: ./policies/alpha.yaml
+  beta:
+    uri: /tmp/beta.omni
+"#,
+        )
+        .unwrap();
+        let settings =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap();
+        let graphs = match settings.mode {
+            ServerConfigMode::Multi { graphs, .. } => graphs,
+            _ => panic!("expected Multi"),
+        };
+        // graphs is BTreeMap-iter order (alphabetical).
+        let alpha = &graphs[0];
+        let beta = &graphs[1];
+        assert_eq!(alpha.graph_id, "alpha");
+        assert_eq!(
+            alpha.policy_file.as_ref().unwrap(),
+            &temp.path().join("policies/alpha.yaml")
+        );
+        assert_eq!(beta.graph_id, "beta");
+        assert!(beta.policy_file.is_none());
+    }
+
+    /// `server.policy.file` resolves alongside the graphs map.
+    #[test]
+    fn server_policy_file_is_resolved_relative_to_base_dir() {
+        let temp = tempfile::tempdir().unwrap();
+        let config_path = temp.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            r#"
+server:
+  policy:
+    file: ./server-policy.yaml
+graphs:
+  alpha:
+    uri: /tmp/alpha.omni
+"#,
+        )
+        .unwrap();
+        let settings =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap();
+        match settings.mode {
+            ServerConfigMode::Multi {
+                server_policy_file, ..
+            } => {
+                assert_eq!(
+                    server_policy_file.unwrap(),
+                    temp.path().join("server-policy.yaml")
+                );
+            }
+            _ => panic!("expected Multi"),
+        }
+    }
+
+    /// End-to-end: load an `omnigraph.yaml` with two graphs and serve
+    /// them. Both graphs must be queryable via cluster routes.
+    ///
+    /// Uses `_` placeholders for tempdirs so they live until end-of-test.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn server_settings_drive_multi_graph_startup_end_to_end() {
+        let cfg_dir = tempfile::tempdir().unwrap();
+        // Real graph storage dirs (the URIs in the config must point to
+        // a graph init-able location).
+        let alpha_dir = cfg_dir.path().join("alpha.omni");
+        let beta_dir = cfg_dir.path().join("beta.omni");
+        let schema = fs::read_to_string(fixture("test.pg")).unwrap();
+        Omnigraph::init(alpha_dir.to_str().unwrap(), &schema)
+            .await
+            .unwrap();
+        Omnigraph::init(beta_dir.to_str().unwrap(), &schema)
+            .await
+            .unwrap();
+
+        let config_path = cfg_dir.path().join("omnigraph.yaml");
+        fs::write(
+            &config_path,
+            format!(
+                r#"
+graphs:
+  alpha:
+    uri: {alpha}
+  beta:
+    uri: {beta}
+"#,
+                alpha = alpha_dir.display(),
+                beta = beta_dir.display(),
+            ),
+        )
+        .unwrap();
+
+        let settings: ServerConfig =
+            load_server_settings(Some(&config_path), None, None, None, true).unwrap();
+        assert!(matches!(settings.mode, ServerConfigMode::Multi { .. }));
+
+        // We don't actually call `serve()` (would bind a socket); we
+        // just confirm the settings are well-formed and the inferred
+        // mode lists both graphs.
+        match settings.mode {
+            ServerConfigMode::Multi { graphs, .. } => {
+                assert_eq!(graphs.len(), 2);
+                let ids: Vec<&str> = graphs.iter().map(|g| g.graph_id.as_str()).collect();
+                assert_eq!(ids, vec!["alpha", "beta"]);
+            }
+            _ => unreachable!(),
+        }
+    }
+}