fix(mcp-oauth): resilient config resolution + reduce config_source spawn amplification + cache OAuth discovery (TASK-326)

Author: Shooksie

crates/animus-mcp-oauth/Cargo.toml

@@ -33,6 +33,7 @@ rmcp = { version = "1.7", features = [
 
 [dev-dependencies]
 protocol = { workspace = true, features = ["test-utils"] }
+orchestrator-config = { workspace = true, features = ["test-utils"] }
 tempfile = "3"
 axum = "0.8"
 http = "1"

crates/animus-mcp-oauth/src/client.rs

@@ -59,33 +59,55 @@ impl McpSession {
         timeout: Duration,
     ) -> Result<Self> {
         crate::ensure_crypto_provider();
+        // The ONE `config_source` resolution per `mcp call`. The resolved URL +
+        // flow are threaded to the spawned proxy (`--url` + `--auth-code`) so the
+        // proxy TRUSTS them and does NOT re-resolve — collapsing the historical
+        // 3× config_source spawn amplification (CLI + proxy-bin + proxy-connect)
+        // down to this single touch.
         let resolution = resolve_server_url(project_root, server, url_override)?;
         // OAuth servers (every flow) are served through the local proxy, which
         // resolves + injects the live bearer itself; plain-HTTP servers are
         // connected directly.
         let uses_oauth = resolution.is_authorization_code || resolution.broker_oauth.is_some();
         if uses_oauth {
-            Self::connect_via_proxy(project_root, server, url_override, timeout).await
+            Self::connect_via_proxy(project_root, server, &resolution, timeout).await
         } else {
             Self::connect_http(&resolution.url, timeout).await
         }
     }
 
-    /// Spawn `animus-mcp-proxy --server <server> [--url <url>] --project-root
-    /// <root>` and connect an MCP client to its stdio. The proxy injects the
-    /// cached bearer; no secret is passed on argv or read here. stderr is
-    /// inherited so proxy diagnostics (e.g. "run `animus mcp auth`") reach the
-    /// user; stdout carries only JSON-RPC frames.
+    /// Spawn `animus-mcp-proxy --server <server> --url <resolved-url>
+    /// [--auth-code] --project-root <root>` and connect an MCP client to its
+    /// stdio. The proxy injects the cached bearer; no secret is passed on argv
+    /// or read here. stderr is inherited so proxy diagnostics (e.g. "run `animus
+    /// mcp auth`") reach the user; stdout carries only JSON-RPC frames.
+    ///
+    /// Passing the already-resolved `--url` (and, for keychain servers,
+    /// `--auth-code`) lets the proxy skip its own `config_source` lookup — so a
+    /// bulk `mcp call` burst does one source resolution per call, here, instead
+    /// of one per proxy spawn.
     async fn connect_via_proxy(
         project_root: &Path,
         server: &str,
-        url_override: Option<&str>,
+        resolution: &crate::config::ServerResolution,
         timeout: Duration,
     ) -> Result<Self> {
         let mut cmd = Command::new(mcp_proxy_command());
         cmd.arg("--server").arg(server).arg("--project-root").arg(project_root);
-        if let Some(url) = url_override.filter(|u| !u.trim().is_empty()) {
-            cmd.arg("--url").arg(url);
+        // Pass the resolved `--url` for BOTH flows so the proxy binds the exact
+        // upstream the parent selected (honoring a `--url` override) instead of
+        // re-resolving the server name to a possibly-different same-named entry.
+        if !resolution.url.trim().is_empty() {
+            cmd.arg("--url").arg(&resolution.url);
+        }
+        // `--auth-code` is added ONLY for keychain (`authorization_code`) servers:
+        // it tells the proxy to trust `--url` and skip its own config_source
+        // lookup entirely (the amplification cut). Broker flows omit it because
+        // they still need their full oauth block from config; the proxy re-resolves
+        // those — and if the source is down it errors cleanly (ConfigSourceUnavailable)
+        // rather than misrouting the broker server to the keychain path.
+        if resolution.is_authorization_code {
+            cmd.arg("--auth-code");
         }
         // `TokioChildProcess::new` pipes stdin/stdout and inherits stderr; the
         // child is killed on drop of the returned transport (held by the

crates/animus-mcp-oauth/src/config.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 use std::sync::Arc;
 
 use orchestrator_config::workflow_config::{
-    load_workflow_config_or_default, load_workflow_config_with_metadata, OauthConfig, OauthFlow,
+    try_load_workflow_config, OauthConfig, OauthFlow, WorkflowConfigAvailability,
 };
 use orchestrator_core::SecretStore;
 use protocol::repository_scope::{repository_scope_for_path, scoped_state_root};
@@ -21,6 +21,13 @@ pub enum ServerResolutionError {
     UnknownServer(String),
     #[error("MCP server `{0}` has no `url`; an HTTP-transport URL is required for OAuth")]
     MissingUrl(String),
+    /// The `config_source` failed to load (spawn / RPC / DB overload / validation),
+    /// so the server's config could NOT be determined. This is distinct from
+    /// [`UnknownServer`], which fires only when the config loaded and the server
+    /// is genuinely absent. Callers should retry / report a transient source
+    /// error rather than "server not configured".
+    #[error("MCP server `{0}` could not be resolved: the workflow config source failed to load ({1}). This is a config-source error, not a missing-server error — retry shortly or check the `config_source` plugin.")]
+    ConfigSourceUnavailable(String, String),
     #[error("failed to load workflow config: {0}")]
     WorkflowConfig(String),
     #[error("failed to load project config: {0}")]
@@ -99,22 +106,30 @@ pub fn resolve_server_url(
 ) -> Result<ServerResolution, ServerResolutionError> {
     // Workflow config mcp_servers first (authoritative for daemon runs).
     //
-    // A malformed `.animus/workflows.yaml` must surface its YAML/validation
-    // error rather than silently falling back to the builtin default (which
-    // would mislead the user into "unknown server"). But an *absent* workflow
-    // config is normal (project-config-only setups), and
-    // `load_workflow_config_with_metadata` returns a "missing" error in that
-    // case. So: only propagate the load error when a workflow YAML source
-    // actually exists; otherwise fall back to the builtin default and continue
-    // to project config.
-    let loaded = match load_workflow_config_with_metadata(project_root, None) {
-        Ok(loaded) => loaded,
-        Err(err) if workflow_yaml_present(project_root) => {
-            return Err(ServerResolutionError::WorkflowConfig(err.to_string()));
+    // Use the NON-SWALLOWING loader so a transient `config_source` failure (DB
+    // overload under bulk `mcp call`) is NOT degraded into an empty config that
+    // then misreports the server as "not defined". `try_load_workflow_config`
+    // distinguishes three cases:
+    //   * Loaded          — use its `mcp_servers`.
+    //   * NoSource        — no config_source configured; benign, fall through to
+    //                       project config (project-config-only setups).
+    //   * SourceUnavailable — the source failed; surface a DISTINCT retryable
+    //                       error. We do this EVEN when the caller supplied a
+    //                       `--url`: a `--url` overrides only the upstream
+    //                       endpoint, not the server's oauth FLOW/block, so
+    //                       synthesizing an `authorization_code` resolution from
+    //                       the URL alone would misroute a broker-flow server
+    //                       (`manual_bearer` / `client_credentials` /
+    //                       `refresh_token`) to the keychain path. Erroring lets
+    //                       the caller retry once the source recovers.
+    let loaded = match try_load_workflow_config(project_root, None) {
+        WorkflowConfigAvailability::Loaded(loaded) => Some(loaded),
+        WorkflowConfigAvailability::NoSource => None,
+        WorkflowConfigAvailability::SourceUnavailable(err) => {
+            return Err(ServerResolutionError::ConfigSourceUnavailable(server.to_string(), err.to_string()));
         }
-        Err(_) => load_workflow_config_or_default(project_root),
     };
-    if let Some(def) = loaded.config.mcp_servers.get(server) {
+    if let Some(def) = loaded.as_ref().and_then(|loaded| loaded.config.mcp_servers.get(server)) {
         return finalize(server, url_override, def.url.clone(), def.oauth.clone());
     }
 
@@ -139,21 +154,6 @@ pub fn resolve_server_url(
     }
 }
 
-/// True when a workflow YAML source exists at `.animus/workflows.yaml` or
-/// any `.animus/workflows/*.yaml`. Used to decide whether a workflow-config
-/// load failure is a real (malformed-config) error worth propagating vs the
-/// benign "no workflow config" case.
-fn workflow_yaml_present(project_root: &Path) -> bool {
-    let animus = project_root.join(".animus");
-    if animus.join("workflows.yaml").exists() {
-        return true;
-    }
-    let dir = animus.join("workflows");
-    std::fs::read_dir(&dir)
-        .map(|entries| entries.flatten().any(|e| e.path().extension().is_some_and(|ext| ext == "yaml" || ext == "yml")))
-        .unwrap_or(false)
-}
-
 fn finalize(
     server: &str,
     url_override: Option<&str>,
@@ -188,3 +188,102 @@ fn finalize(
         }),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use orchestrator_config::workflow_config::builtin_workflow_config;
+    use orchestrator_config::workflow_config::config_source_client::test_seam;
+    use std::sync::{Mutex, MutexGuard, OnceLock};
+
+    // These tests mutate the process-global config_source test seam + HOME env,
+    // so serialize them under one lock held for the whole test body.
+    fn serial() -> &'static Mutex<()> {
+        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+        LOCK.get_or_init(|| Mutex::new(()))
+    }
+
+    /// RAII guard that points `HOME` at a temp dir for the test and RESTORES the
+    /// previous value on drop, so the mutation never leaks to sibling tests.
+    struct HomeGuard {
+        prev: Option<std::ffi::OsString>,
+    }
+
+    impl HomeGuard {
+        fn set(path: &Path) -> Self {
+            let prev = std::env::var_os("HOME");
+            std::env::set_var("HOME", path);
+            Self { prev }
+        }
+    }
+
+    impl Drop for HomeGuard {
+        fn drop(&mut self) {
+            match self.prev.take() {
+                Some(v) => std::env::set_var("HOME", v),
+                None => std::env::remove_var("HOME"),
+            }
+        }
+    }
+
+    /// Common fixture: hold the serial lock, isolate `HOME` to a fresh tempdir
+    /// (restored on drop). Bound in the caller in declaration order so the
+    /// tempdir drops LAST — after `HomeGuard` has restored `HOME`.
+    fn fixture() -> (MutexGuard<'static, ()>, tempfile::TempDir, HomeGuard) {
+        let guard = serial().lock().unwrap_or_else(|p| p.into_inner());
+        let temp = tempfile::tempdir().expect("tempdir");
+        let home = HomeGuard::set(temp.path());
+        (guard, temp, home)
+    }
+
+    /// TASK-326 (fix d): a transient config_source failure must surface as the
+    /// DISTINCT `ConfigSourceUnavailable`, NOT `UnknownServer` — otherwise the
+    /// caller concludes "krisp isn't configured" when the source is merely down.
+    #[test]
+    fn source_failure_yields_config_source_unavailable_not_unknown_server() {
+        let (_guard, temp, _home) = fixture();
+        let root = temp.path();
+
+        let _fail = test_seam::install_failure(root, "config_source RPC timed out");
+        let err = resolve_server_url(root, "krisp", None).expect_err("must fail when source is down");
+        assert!(
+            matches!(err, ServerResolutionError::ConfigSourceUnavailable(ref s, _) if s == "krisp"),
+            "expected ConfigSourceUnavailable, got {err:?}"
+        );
+    }
+
+    /// Even with an explicit `--url`, a source outage yields
+    /// `ConfigSourceUnavailable` (never a synthesized `authorization_code`
+    /// resolution): a `--url` overrides only the upstream endpoint, not the
+    /// server's oauth FLOW, so guessing the flow could misroute a broker server
+    /// to the keychain path. The caller retries once the source recovers.
+    #[test]
+    fn explicit_url_still_errors_when_source_unavailable() {
+        let (_guard, temp, _home) = fixture();
+        let root = temp.path();
+
+        let _fail = test_seam::install_failure(root, "config_source RPC timed out");
+        let err = resolve_server_url(root, "krisp", Some("https://example.test/mcp"))
+            .expect_err("a source outage must error even with --url, to avoid misrouting a broker server");
+        assert!(
+            matches!(err, ServerResolutionError::ConfigSourceUnavailable(ref s, _) if s == "krisp"),
+            "expected ConfigSourceUnavailable, got {err:?}"
+        );
+    }
+
+    /// When the config LOADS but the server is genuinely absent, the error is
+    /// `UnknownServer` (the pre-existing behavior) — never `ConfigSourceUnavailable`.
+    #[test]
+    fn loaded_config_missing_server_yields_unknown_server() {
+        let (_guard, temp, _home) = fixture();
+        let root = temp.path();
+
+        // A base with no mcp_servers loads cleanly; the server is simply absent.
+        let _base = test_seam::install(root, builtin_workflow_config());
+        let err = resolve_server_url(root, "krisp", None).expect_err("absent server must error");
+        assert!(
+            matches!(err, ServerResolutionError::UnknownServer(ref s) if s == "krisp"),
+            "expected UnknownServer, got {err:?}"
+        );
+    }
+}

crates/animus-mcp-oauth/src/flow.rs

@@ -46,7 +46,7 @@ enum Exchange {
 }
 
 use crate::callback::CallbackListener;
-use crate::config::{build_secret_store, resolve_principal_id, resolve_server_url};
+use crate::config::{build_secret_store, resolve_principal_id, resolve_server_url, ServerResolutionError};
 use crate::keychain_store::KeychainCredentialStore;
 use crate::pending::{PendingAuth, PendingStore};
 use crate::state_store::PersistentStateStore;
@@ -745,17 +745,26 @@ pub async fn auth_status(project_root: &Path, server: Option<&str>, url_override
 
     let servers: Vec<String> = match server {
         Some(s) => vec![s.to_string()],
-        None => authorization_code_servers(project_root),
+        // Non-swallowing enumeration: a transient config_source failure surfaces
+        // as an Err here rather than an empty `servers:[]` that would read as
+        // "no OAuth servers configured".
+        None => authorization_code_servers(project_root)?,
     };
 
     let mut out = Vec::with_capacity(servers.len());
     for name in servers {
         // Tokens are keyed by the upstream URL too, so resolve it. A server
         // that can't be resolved (e.g. dropped from config and no --url) is
-        // reported as unauthenticated rather than failing the whole report.
-        let Some(url) = resolve_server_url(project_root, &name, url_override).ok().map(|r| r.url) else {
-            out.push(server_state_from_creds(&name, &principal, None));
-            continue;
+        // reported as unauthenticated rather than failing the whole report —
+        // EXCEPT a transient config_source failure, which must surface (not be
+        // silently reported as an unauthenticated server) so the user retries.
+        let url = match resolve_server_url(project_root, &name, url_override) {
+            Ok(resolution) => resolution.url,
+            Err(err @ ServerResolutionError::ConfigSourceUnavailable(..)) => return Err(err.into()),
+            Err(_) => {
+                out.push(server_state_from_creds(&name, &principal, None));
+                continue;
+            }
         };
         let store = KeychainCredentialStore::new(secrets.clone(), &name, &principal, &url);
         let creds =
@@ -814,14 +823,27 @@ fn server_state_from_creds(server: &str, principal: &str, creds: Option<StoredCr
 
 /// Servers in workflow/project config carrying an `authorization_code`
 /// oauth flow.
-fn authorization_code_servers(project_root: &Path) -> Vec<String> {
-    use orchestrator_config::workflow_config::{load_workflow_config_or_default, OauthFlow};
+///
+/// Uses the NON-SWALLOWING loader: a transient `config_source` failure returns
+/// `Err` so `auth_status` reports a transient error rather than an empty
+/// `servers:[]` (which would read as "no OAuth servers configured"). A genuinely
+/// absent config source (`NoSource`) is benign — enumeration proceeds from
+/// project config only.
+fn authorization_code_servers(project_root: &Path) -> Result<Vec<String>> {
+    use orchestrator_config::workflow_config::{try_load_workflow_config, OauthFlow, WorkflowConfigAvailability};
     let mut names = std::collections::BTreeSet::new();
 
-    let loaded = load_workflow_config_or_default(project_root);
-    for (name, def) in &loaded.config.mcp_servers {
-        if def.oauth.as_ref().is_some_and(|o| o.flow == OauthFlow::AuthorizationCode) {
-            names.insert(name.clone());
+    match try_load_workflow_config(project_root, None) {
+        WorkflowConfigAvailability::Loaded(loaded) => {
+            for (name, def) in &loaded.config.mcp_servers {
+                if def.oauth.as_ref().is_some_and(|o| o.flow == OauthFlow::AuthorizationCode) {
+                    names.insert(name.clone());
+                }
+            }
+        }
+        WorkflowConfigAvailability::NoSource => {}
+        WorkflowConfigAvailability::SourceUnavailable(err) => {
+            return Err(anyhow!("workflow config source unavailable while listing OAuth servers: {err}"));
         }
     }
 
@@ -838,7 +860,7 @@ fn authorization_code_servers(project_root: &Path) -> Vec<String> {
         }
     }
 
-    names.into_iter().collect()
+    Ok(names.into_iter().collect())
 }
 
 /// Extract the `state` query parameter from an authorization URL.