soma_application/

provider_registry.rs

//! Provider dispatch registry: the [`ProviderRegistry`] resolves actions to
//! providers, builds catalog snapshots, and indexes resources and prompts.
use std::{
    collections::{BTreeMap, HashMap, VecDeque},
    path::{Path, PathBuf},
    sync::{
        Arc, RwLock,
        atomic::{AtomicBool, Ordering},
    },
};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use soma_domain::provider_validation::{ProviderValidationError, validate_provider_manifest};
use soma_provider_adapters::python::materializer::PreparedPythonEnvironment;
use soma_provider_core::{
    ProviderCall as CoreProviderCall, ProviderCatalog, ProviderRegistry as CoreRegistry,
    ProviderResource, RegistrySnapshot as CoreRegistrySnapshot,
};

use crate::{
    capabilities::CapabilityBroker,
    provider_errors::ProviderError,
    providers::filesystem::{FileProviderSource, PythonProviderEnvironmentSelections},
};

mod dispatch_lease;
mod enforcement;
mod generations;
#[path = "provider_registry_map.rs"]
mod map;
use map::provider_map;
#[path = "provider_registry_openapi.rs"]
mod openapi;
mod python_operations;
mod refresh;
mod reports;
mod resources;
use dispatch_lease::DispatchLease;
pub(super) use enforcement::provider_tool_surface_enabled;
use enforcement::{enforce_capabilities, enforce_pre_input, enforce_response_limit};
use generations::{
    StoredGeneration, publish_generation, settle_transition, settle_transition_async,
};
use openapi::openapi_paths_from_core;
use refresh::{ProviderRefreshEvent, fingerprint_file_source};
use resources::ResourceIndex;
pub use resources::{DynamicResourceTemplate, ResourceReadOutput};
pub use soma_provider_core::{Provider as CoreProvider, ProviderOutput};
/// Product-neutral provider call shape (`soma_provider_core::ProviderCall`),
/// as handed to the shared core registry once auth/scope fields are stripped.
pub type ProviderInvocation = CoreProviderCall;

/// A loaded provider: exposes a catalog and dispatches action calls, with
/// optional MCP resource-read support layered on top.
#[async_trait]
pub trait Provider: Send + Sync {
    /// The provider's catalog — its tools, prompts, resources, and metadata.
    fn catalog(&self) -> ProviderCatalog;

    /// Executes one provider action call and returns its output.
    async fn call(&self, call: ProviderCall) -> Result<ProviderOutput, ProviderError>;

    /// Drains and releases provider-owned runtime resources after a registry
    /// generation swap.
    async fn retire(&self) {}

    /// Parks runtime resources while preserving rollback capability.
    async fn suspend(&self) {}

    /// Returns product-neutral operator status for a stateful runtime.
    fn runtime_status(&self) -> Option<Value> {
        None
    }

    /// Cancels active work without waiting for provider dispatch to complete.
    fn cancel_active(&self) -> bool {
        false
    }

    /// Clears runtime quarantine after explicit operator authorization.
    async fn reset_quarantine(&self) {}

    /// Stops new work while in-flight work drains on a retained generation.
    fn deactivate(&self) {}

    /// Re-enables a retained generation selected by operator rollback.
    fn activate(&self) {}

    /// Reserves an invocation while this provider is still active.
    fn acquire_dispatch(&self) -> bool {
        true
    }

    /// Releases a reservation acquired by [`Provider::acquire_dispatch`].
    fn release_dispatch(&self) {}

    /// Dynamic resource templates this provider serves. Every provider
    /// inherits the empty default — only file-based dynamic resource
    /// readers (`providers/resources/*.ts`) override it.
    fn dynamic_resource_templates(&self) -> Vec<DynamicResourceTemplate> {
        Vec::new()
    }

    /// Whether this provider can actually serve `read_resource` calls.
    /// `catalog().resources` is a schema-legal field on every provider
    /// kind's manifest (OpenAPI, MCP, ai-sdk, WASM, Python, generic JSON),
    /// but only file-based `ResourceFileProvider`s have any mechanism to
    /// read content back — every other kind inherits `read_resource`'s
    /// default error. `ResourceIndex::register` uses this to reject a
    /// manifest that declares resources it can never serve at snapshot-build
    /// time, rather than letting them appear in `resources/list` and always
    /// fail `resources/read` with an opaque error.
    fn supports_resource_reads(&self) -> bool {
        false
    }

    /// Reads resource content for a URI the registry has already matched
    /// against either this provider's `catalog().resources` (exact,
    /// `params` empty) or one of its `dynamic_resource_templates()`
    /// (`params` holds captured path parameters).
    async fn read_resource(
        &self,
        uri: &str,
        params: &BTreeMap<String, String>,
    ) -> Result<ResourceReadOutput, ProviderError> {
        let _ = params;
        Err(ProviderError::validation(
            &self.catalog().provider.name,
            uri,
            "resource_read_not_supported",
            "this provider does not support resource reads",
        ))
    }
}

/// The Soma-facing surface a provider call arrives on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum ProviderSurface {
    /// MCP tool/resource/prompt surface.
    Mcp,
    /// REST HTTP surface.
    Rest,
    /// Command-line surface.
    Cli,
    /// Command palette surface.
    Palette,
}

impl ProviderSurface {
    /// The lowercase string identifier for this surface.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Mcp => "mcp",
            Self::Rest => "rest",
            Self::Cli => "cli",
            Self::Palette => "palette",
        }
    }

    fn core(self) -> soma_provider_core::ProviderSurface {
        match self {
            Self::Mcp => soma_provider_core::ProviderSurface::Mcp,
            Self::Rest => soma_provider_core::ProviderSurface::Rest,
            Self::Cli => soma_provider_core::ProviderSurface::Cli,
            Self::Palette => soma_provider_core::ProviderSurface::Palette,
        }
    }
}

/// How the caller authenticated, governing whether scope checks apply.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderAuthMode {
    /// Loopback development mode; scope checks bypassed.
    LoopbackDev,
    /// Non-loopback behind an authz-enforcing gateway; scope checks bypassed.
    TrustedGateway,
    /// Mounted auth middleware; scope checks enforced.
    Mounted,
}

/// The authenticated caller: its subject identity and granted scopes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderPrincipal {
    /// Subject identifier for the caller.
    pub subject: String,
    /// Scopes granted to the caller.
    pub scopes: Vec<String>,
}

impl ProviderPrincipal {
    /// The synthetic principal used in loopback development mode, granted the
    /// read scope.
    pub fn loopback_dev() -> Self {
        Self {
            subject: "loopback-dev".to_owned(),
            scopes: vec![soma_domain::actions::READ_SCOPE.to_owned()],
        }
    }

    /// An anonymous principal with no scopes.
    pub fn anonymous() -> Self {
        Self {
            subject: "anonymous".to_owned(),
            scopes: Vec::new(),
        }
    }
}

/// Byte-size ceilings enforced on a provider call's input and response.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ProviderRequestLimits {
    /// Maximum accepted request input size in bytes.
    pub max_input_bytes: usize,
    /// Maximum permitted response size in bytes.
    pub max_response_bytes: usize,
}

impl Default for ProviderRequestLimits {
    fn default() -> Self {
        Self {
            max_input_bytes: 64 * 1024,
            max_response_bytes: soma_domain::token_limit::MAX_RESPONSE_BYTES,
        }
    }
}

/// A fully-resolved provider action invocation, carrying auth, scope, surface,
/// and limit context alongside the action and its parameters.
#[derive(Debug, Clone)]
pub struct ProviderCall {
    /// Target provider name.
    pub provider: String,
    /// Action to invoke on the provider.
    pub action: String,
    /// Action parameters as JSON.
    pub params: Value,
    /// Authenticated caller.
    pub principal: ProviderPrincipal,
    /// Authentication mode governing scope enforcement.
    pub auth_mode: ProviderAuthMode,
    /// Surface the call arrived on.
    pub surface: ProviderSurface,
    /// Whether the caller confirmed a destructive action.
    pub destructive_confirmed: bool,
    /// Byte-size limits applied to this call.
    pub limits: ProviderRequestLimits,
    /// Registry snapshot id this call is bound to.
    pub snapshot_id: String,
    /// Stable request identity propagated into isolated provider runtimes.
    pub request_id: String,
    /// Optional distributed trace parent propagated after authentication.
    pub traceparent: Option<String>,
    /// Optional distributed trace state propagated after authentication.
    pub tracestate: Option<String>,
    /// Invocation-scoped progress collector shared with provider adapters.
    pub progress: soma_provider_core::ProviderProgressReporter,
}

impl ProviderCall {
    /// Strips this call down to the product-neutral core invocation shape.
    pub fn provider_invocation(&self) -> ProviderInvocation {
        self.provider_invocation_with_progress(self.progress.clone())
    }

    /// Builds the core invocation with a host-owned progress collector.
    pub fn provider_invocation_with_progress(
        &self,
        progress: soma_provider_core::ProviderProgressReporter,
    ) -> ProviderInvocation {
        ProviderInvocation {
            provider: self.provider.clone(),
            action: self.action.clone(),
            params: self.params.clone(),
            surface: self.surface.core(),
            snapshot_id: self.snapshot_id.clone(),
            context: soma_provider_core::ProviderInvocationContext {
                request_id: self.request_id.clone(),
                actor_id: (!self.principal.subject.is_empty())
                    .then(|| self.principal.subject.clone()),
                actor_scopes: self.principal.scopes.clone(),
                traceparent: self.traceparent.clone(),
                tracestate: self.tracestate.clone(),
                progress,
            },
        }
    }

    /// Builds the serializable execution envelope describing this call.
    pub fn execution_envelope(&self) -> ProviderExecutionEnvelope {
        ProviderExecutionEnvelope {
            schema_version: 1,
            provider: self.provider.clone(),
            action: self.action.clone(),
            params: self.params.clone(),
            surface: self.surface,
            snapshot_id: self.snapshot_id.clone(),
        }
    }

    /// Serializes the execution envelope to JSON bytes.
    pub fn execution_payload(&self) -> Result<Vec<u8>, serde_json::Error> {
        serde_json::to_vec(&self.execution_envelope())
    }
}

/// The serializable record of a provider action invocation, emitted for
/// downstream execution/audit consumers.
#[derive(Debug, Clone, Serialize)]
pub struct ProviderExecutionEnvelope {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Target provider name.
    pub provider: String,
    /// Action invoked.
    pub action: String,
    /// Action parameters as JSON.
    pub params: Value,
    /// Surface the call arrived on.
    pub surface: ProviderSurface,
    /// Registry snapshot id the call is bound to.
    pub snapshot_id: String,
}

/// An immutable, fingerprinted view of the loaded providers: their catalogs,
/// routing/resource indexes, and cached derived artifacts.
#[derive(Clone)]
pub struct RegistrySnapshot {
    /// Snapshot identifier (equal to the fingerprint).
    pub id: String,
    /// Content fingerprint of the loaded provider set.
    pub fingerprint: String,
    /// Catalogs for every loaded provider.
    pub catalogs: Vec<ProviderCatalog>,
    core: Arc<CoreRegistrySnapshot>,
    exact_resources: HashMap<String, (String, ProviderResource)>,
    dynamic_resources: Vec<(String, DynamicResourceTemplate)>,
    /// Number of JSON schema validators compiled for this snapshot.
    pub compiled_validator_count: usize,
    /// Cached, pre-serialized OpenAPI document bytes.
    pub cached_openapi_bytes: Arc<Vec<u8>>,
    /// Cached catalog summary JSON.
    pub cached_catalog_summary: Arc<Value>,
}

impl RegistrySnapshot {
    /// The names of every action in this snapshot.
    pub fn action_names(&self) -> Vec<&str> {
        self.core.action_names().collect()
    }

    /// Resolves a REST method+path to its action name, if routed.
    pub fn route_action(&self, method: &str, path: &str) -> Option<&str> {
        self.core.route_action(method, path)
    }

    /// Resolves a CLI command to its action name, if routed.
    pub fn cli_action(&self, command: &str) -> Option<&str> {
        self.core.cli_action(command)
    }

    /// The primitive kind (tool/prompt/resource) for a named primitive.
    pub fn primitive_kind(&self, name: &str) -> Option<&str> {
        self.core.primitive_kind(name)
    }

    /// Whether the action is marked destructive and requires confirmation.
    pub fn action_requires_confirmation(&self, action: &str) -> bool {
        self.core
            .tool(action)
            .map(|entry| entry.spec().destructive)
            .unwrap_or(false)
    }

    /// The provider that owns the given action.
    pub fn provider_for_action(&self, action: &str) -> Option<&str> {
        self.core
            .tool(action)
            .map(|entry| entry.provider_id().as_str())
    }

    /// Borrows the underlying product-neutral core snapshot.
    pub fn core_snapshot(&self) -> &CoreRegistrySnapshot {
        &self.core
    }

    pub(crate) fn rest_routes(&self) -> impl Iterator<Item = (&str, &str, &str)> {
        self.core.rest_routes()
    }
}

/// Thread-safe registry of loaded providers backing every Soma surface,
/// holding the active snapshot and supporting hot-reload from a file source.
#[derive(Clone)]
pub struct ProviderRegistry {
    state: Arc<RwLock<RegistryState>>,
    refresh_gate: Arc<tokio::sync::Mutex<()>>,
    refresh_dirty: Arc<AtomicBool>,
    capabilities: CapabilityBroker,
    base_providers: Arc<Vec<Arc<dyn Provider>>>,
    file_source: Option<FileProviderSource>,
}

struct RegistryState {
    providers: BTreeMap<String, Arc<dyn Provider>>,
    core_registry: CoreRegistry,
    snapshot: Arc<RegistrySnapshot>,
    file_fingerprint: Option<String>,
    python_environments: PythonProviderEnvironmentSelections,
    generation_id: u64,
    history: VecDeque<StoredGeneration>,
}

const REFRESH_DEBOUNCE: std::time::Duration = std::time::Duration::from_millis(75);

impl ProviderRegistry {
    /// Returns the configured dynamic-provider root, when file providers are enabled.
    #[must_use]
    pub fn file_provider_root(&self) -> Option<&std::path::Path> {
        self.file_source.as_ref().map(FileProviderSource::root)
    }

    /// Builds a registry from the given providers with a default-deny
    /// capability broker.
    pub fn new(providers: Vec<Arc<dyn Provider>>) -> Result<Self, ProviderValidationError> {
        Self::with_capabilities(providers, CapabilityBroker::default_deny())
    }

    /// Builds a registry from the given providers and capability broker.
    pub fn with_capabilities(
        providers: Vec<Arc<dyn Provider>>,
        capabilities: CapabilityBroker,
    ) -> Result<Self, ProviderValidationError> {
        let (providers, core_registry, snapshot) = build_registry(providers)?;
        Ok(Self {
            state: Arc::new(RwLock::new(RegistryState {
                providers,
                core_registry,
                snapshot,
                file_fingerprint: None,
                python_environments: PythonProviderEnvironmentSelections::new(),
                generation_id: 1,
                history: VecDeque::new(),
            })),
            refresh_gate: Arc::new(tokio::sync::Mutex::new(())),
            refresh_dirty: Arc::new(AtomicBool::new(false)),
            capabilities,
            base_providers: Arc::new(Vec::new()),
            file_source: None,
        })
    }

    /// Builds a registry from the given base providers plus dynamic providers
    /// loaded from a file source, enabling later hot-reloads.
    pub fn with_file_source(
        providers: Vec<Arc<dyn Provider>>,
        capabilities: CapabilityBroker,
        file_source: FileProviderSource,
    ) -> Result<Self, ProviderValidationError> {
        let file_fingerprint = file_source.fingerprint().map_err(|error| {
            ProviderValidationError::new("provider_file_load_failed", error.to_string())
        })?;
        let dynamic_providers = file_source.load().map_err(|error| {
            ProviderValidationError::new("provider_file_load_failed", error.to_string())
        })?;
        let base_providers = Arc::new(providers);
        let mut all_providers = base_providers.iter().cloned().collect::<Vec<_>>();
        all_providers.extend(dynamic_providers);
        let (providers, core_registry, snapshot) = build_registry(all_providers)?;
        Ok(Self {
            state: Arc::new(RwLock::new(RegistryState {
                providers,
                core_registry,
                snapshot,
                file_fingerprint: Some(file_fingerprint),
                python_environments: PythonProviderEnvironmentSelections::new(),
                generation_id: 1,
                history: VecDeque::new(),
            })),
            refresh_gate: Arc::new(tokio::sync::Mutex::new(())),
            refresh_dirty: Arc::new(AtomicBool::new(false)),
            capabilities,
            base_providers,
            file_source: Some(file_source),
        })
    }

    /// Async constructor that preflights persistent Python workers before the
    /// first registry snapshot becomes visible.
    pub async fn with_file_source_async(
        providers: Vec<Arc<dyn Provider>>,
        capabilities: CapabilityBroker,
        file_source: FileProviderSource,
    ) -> Result<Self, ProviderValidationError> {
        let selections = PythonProviderEnvironmentSelections::new();
        let file_fingerprint = file_source
            .fingerprint_with_python_environments(&selections)
            .map_err(|error| {
                ProviderValidationError::new("provider_file_load_failed", error.to_string())
            })?;
        let dynamic_providers = file_source
            .load_with_python_environments_async(&selections)
            .await
            .map_err(|error| {
                ProviderValidationError::new("provider_file_load_failed", error.to_string())
            })?;
        let base_providers = Arc::new(providers);
        let mut all_providers = base_providers.iter().cloned().collect::<Vec<_>>();
        all_providers.extend(dynamic_providers);
        let (providers, core_registry, snapshot) = build_registry(all_providers)?;
        Ok(Self {
            state: Arc::new(RwLock::new(RegistryState {
                providers,
                core_registry,
                snapshot,
                file_fingerprint: Some(file_fingerprint),
                python_environments: selections,
                generation_id: 1,
                history: VecDeque::new(),
            })),
            refresh_gate: Arc::new(tokio::sync::Mutex::new(())),
            refresh_dirty: Arc::new(AtomicBool::new(false)),
            capabilities,
            base_providers,
            file_source: Some(file_source),
        })
    }

    /// Returns the currently active registry snapshot.
    pub fn snapshot(&self) -> Arc<RegistrySnapshot> {
        self.state
            .read()
            .expect("provider registry lock should not be poisoned")
            .snapshot
            .clone()
    }

    /// Authorize a not-yet-active candidate against the same deployment
    /// grants used by normal provider dispatch.
    pub(crate) fn authorize_candidate_capabilities(
        &self,
        catalog: &ProviderCatalog,
        action: &str,
    ) -> Result<(), ProviderError> {
        self.capabilities
            .authorize(&catalog.provider.name, action, &catalog.capabilities)
    }

    /// Refreshes providers from the file source, if any. Per the drop-in
    /// provider layout contract ("If a resource disappears or becomes
    /// invalid, a reload must leave the last valid snapshot active until a
    /// valid replacement snapshot is available"), a failure anywhere in this
    /// pipeline (an unreadable directory, a newly-invalid or colliding
    /// provider file) is logged and this returns the previous snapshot
    /// rather than propagating the error — one bad drop-in file must not
    /// take down `list_tools`/`list_prompts`/`read_resource`/etc. for every
    /// other, unrelated, already-loaded provider.
    pub fn refresh_file_providers(&self) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let Some(file_source) = &self.file_source else {
            return Ok(self.snapshot());
        };
        let (baseline_fingerprint, active_environments) = {
            let state = self
                .state
                .read()
                .expect("provider registry lock should not be poisoned");
            (
                state.file_fingerprint.clone(),
                state.python_environments.clone(),
            )
        };
        let file_fingerprint =
            match file_source.fingerprint_with_python_environments(&active_environments) {
                Ok(fingerprint) => fingerprint,
                Err(error) => {
                    return Ok(self.snapshot_after_refresh_failure(
                        file_source,
                        "provider_file_fingerprint_failed",
                        &error.to_string(),
                    ));
                }
            };
        if baseline_fingerprint.as_deref() == Some(file_fingerprint.as_str()) {
            return Ok(self.snapshot());
        }

        let rebuilt: Result<_, ProviderValidationError> = (|| {
            let dynamic_providers = file_source
                .load_with_python_environments(&active_environments)
                .map_err(|error| {
                    ProviderValidationError::new("provider_file_load_failed", error.to_string())
                })?;
            let mut providers = self.base_providers.iter().cloned().collect::<Vec<_>>();
            providers.extend(dynamic_providers);
            build_registry(providers)
        })();
        let (providers, core_registry, snapshot) = match rebuilt {
            Ok(parts) => parts,
            Err(error) => {
                return Ok(self.snapshot_after_refresh_failure(
                    file_source,
                    error.code(),
                    error.message(),
                ));
            }
        };

        let mut state = self
            .state
            .write()
            .expect("provider registry lock should not be poisoned");
        if state.file_fingerprint != baseline_fingerprint
            || state.python_environments != active_environments
        {
            return Ok(state.snapshot.clone());
        }
        let previous = state.snapshot.clone();
        let public_catalog_changed = previous.fingerprint != snapshot.fingerprint;
        let transition = publish_generation(
            &mut state,
            providers,
            core_registry,
            snapshot.clone(),
            Some(file_fingerprint),
            active_environments,
        );
        if public_catalog_changed {
            ProviderRefreshEvent::new(&previous, &snapshot).log(file_source.root());
        }
        let active = state.snapshot.clone();
        drop(state);
        settle_transition(transition);
        Ok(active)
    }

    /// Refresh variant that preflights persistent Python candidates outside
    /// the registry write lock before publishing the candidate snapshot.
    pub async fn refresh_file_providers_async(
        &self,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let Some(file_source) = &self.file_source else {
            return Ok(self.snapshot());
        };
        // One caller owns candidate preparation. Other request-path refresh
        // checks keep serving the last healthy generation instead of queueing
        // behind debounce, catalog import, and old-generation drain.
        let Ok(_refresh_guard) = self.refresh_gate.try_lock() else {
            self.refresh_dirty.store(true, Ordering::Release);
            return Ok(self.snapshot());
        };
        let mut active = self.snapshot();
        for pass in 0..2 {
            self.refresh_dirty.store(false, Ordering::Release);
            active = self
                .refresh_file_providers_owned(file_source, false)
                .await?;
            if !self.refresh_dirty.swap(false, Ordering::AcqRel) {
                break;
            }
            if pass == 1 {
                // Bound request-path work under sustained contention. The
                // retained dirty bit makes the next caller perform another
                // refresh instead of forcing this owner to loop forever.
                self.refresh_dirty.store(true, Ordering::Release);
            }
        }
        Ok(active)
    }

    /// Acquires the exclusive provider publication lane for a multi-step
    /// filesystem mutation, strict refresh, verification, and commit.
    pub(crate) async fn lock_refresh_lane(&self) -> tokio::sync::OwnedMutexGuard<()> {
        self.refresh_gate.clone().lock_owned().await
    }

    /// Strictly refreshes while the caller holds [`Self::lock_refresh_lane`].
    pub(crate) async fn refresh_file_providers_strict_in_lane(
        &self,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let Some(file_source) = &self.file_source else {
            return Ok(self.snapshot());
        };
        self.refresh_file_providers_owned(file_source, true).await
    }

    async fn refresh_file_providers_owned(
        &self,
        file_source: &FileProviderSource,
        strict: bool,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let (baseline_fingerprint, active_environments) = {
            let state = self
                .state
                .read()
                .expect("provider registry lock should not be poisoned");
            (
                state.file_fingerprint.clone(),
                state.python_environments.clone(),
            )
        };
        let observed_fingerprint =
            match fingerprint_file_source(file_source.clone(), active_environments.clone()).await {
                Ok(fingerprint) => fingerprint,
                Err(error) => {
                    if strict {
                        return Err(ProviderValidationError::new(
                            "provider_file_fingerprint_failed",
                            error.to_string(),
                        ));
                    }
                    return Ok(self.snapshot_after_refresh_failure(
                        file_source,
                        "provider_file_fingerprint_failed",
                        &error.to_string(),
                    ));
                }
            };
        if baseline_fingerprint.as_deref() == Some(observed_fingerprint.as_str()) {
            return Ok(self.snapshot());
        }
        // Editors commonly emit write/rename/chmod bursts. Hold the single
        // refresh lane briefly, then fingerprint the settled directory once.
        tokio::time::sleep(REFRESH_DEBOUNCE).await;
        let file_fingerprint =
            match fingerprint_file_source(file_source.clone(), active_environments.clone()).await {
                Ok(fingerprint) => fingerprint,
                Err(error) => {
                    if strict {
                        return Err(ProviderValidationError::new(
                            "provider_file_fingerprint_failed",
                            error.to_string(),
                        ));
                    }
                    return Ok(self.snapshot_after_refresh_failure(
                        file_source,
                        "provider_file_fingerprint_failed",
                        &error.to_string(),
                    ));
                }
            };
        if baseline_fingerprint.as_deref() == Some(file_fingerprint.as_str()) {
            return Ok(self.snapshot());
        }
        let dynamic_providers = match file_source
            .load_with_python_environments_async(&active_environments)
            .await
        {
            Ok(providers) => providers,
            Err(error) => {
                if strict {
                    return Err(ProviderValidationError::new(
                        "provider_file_load_failed",
                        error.to_string(),
                    ));
                }
                return Ok(self.snapshot_after_refresh_failure(
                    file_source,
                    "provider_file_load_failed",
                    &error.to_string(),
                ));
            }
        };
        if strict
            && let Some((provider, status)) = dynamic_providers.iter().find_map(|provider| {
                provider
                    .runtime_status()
                    .filter(|status| status.get("ready").and_then(Value::as_bool) == Some(false))
                    .map(|status| (provider.catalog().provider.name, status))
            })
        {
            return Err(ProviderValidationError::new(
                "provider_runtime_not_ready",
                format!("provider `{provider}` failed runtime preparation: {status}"),
            ));
        }
        let mut all_providers = self.base_providers.iter().cloned().collect::<Vec<_>>();
        all_providers.extend(dynamic_providers);
        let (providers, core_registry, snapshot) = match build_registry(all_providers) {
            Ok(parts) => parts,
            Err(error) => {
                if strict {
                    return Err(error);
                }
                return Ok(self.snapshot_after_refresh_failure(
                    file_source,
                    error.code(),
                    error.message(),
                ));
            }
        };
        let (active, transition) = {
            let mut state = self
                .state
                .write()
                .expect("provider registry lock should not be poisoned");
            if state.file_fingerprint != baseline_fingerprint
                || state.python_environments != active_environments
            {
                if strict {
                    return Err(ProviderValidationError::new(
                        "provider_refresh_raced",
                        "provider registry changed during strict refresh",
                    ));
                }
                return Ok(state.snapshot.clone());
            }
            let previous = state.snapshot.clone();
            let public_catalog_changed = previous.fingerprint != snapshot.fingerprint;
            let transition = publish_generation(
                &mut state,
                providers,
                core_registry,
                snapshot.clone(),
                Some(file_fingerprint),
                active_environments,
            );
            if public_catalog_changed {
                ProviderRefreshEvent::new(&previous, &snapshot).log(file_source.root());
            }
            (state.snapshot.clone(), transition)
        };
        settle_transition_async(transition).await;
        Ok(active)
    }

    /// Validates and atomically activates one immutable Python environment candidate.
    pub async fn activate_python_candidate(
        &self,
        provider_path: &Path,
        candidate: PreparedPythonEnvironment,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        // Explicit updates share the same preparation lane as filesystem
        // refreshes, preventing duplicate candidate worker construction.
        let _refresh_guard = self.refresh_gate.lock().await;
        let file_source = self.file_source.as_ref().ok_or_else(|| {
            ProviderValidationError::new(
                "provider_file_source_unavailable",
                "Python candidates require a file-backed provider registry",
            )
        })?;
        let provider_path = file_source
            .resolve_python_provider_path(provider_path)
            .map_err(|error| {
                ProviderValidationError::new("provider_candidate_path_invalid", error.to_string())
            })?;
        let (baseline_fingerprint, baseline_environments) = {
            let state = self
                .state
                .read()
                .expect("provider registry lock should not be poisoned");
            (
                state.file_fingerprint.clone(),
                state.python_environments.clone(),
            )
        };
        let mut next_environments = baseline_environments.clone();
        next_environments.insert(provider_path, candidate);

        let dynamic_providers = file_source
            .load_with_python_environments_async(&next_environments)
            .await
            .map_err(|error| {
                ProviderValidationError::new(
                    "provider_candidate_validation_failed",
                    error.to_string(),
                )
            })?;
        let file_fingerprint = file_source
            .fingerprint_with_python_environments(&next_environments)
            .map_err(|error| {
                ProviderValidationError::new("provider_file_fingerprint_failed", error.to_string())
            })?;
        let mut providers = self.base_providers.iter().cloned().collect::<Vec<_>>();
        providers.extend(dynamic_providers);
        let (providers, core_registry, snapshot) = build_registry(providers)?;

        let (active, transition) = {
            let mut state = self
                .state
                .write()
                .expect("provider registry lock should not be poisoned");
            if state.file_fingerprint != baseline_fingerprint
                || state.python_environments != baseline_environments
            {
                return Err(ProviderValidationError::new(
                    "provider_candidate_activation_raced",
                    "provider registry changed while the candidate was being validated",
                ));
            }
            let previous = state.snapshot.clone();
            let public_catalog_changed = previous.fingerprint != snapshot.fingerprint;
            let transition = publish_generation(
                &mut state,
                providers,
                core_registry,
                snapshot.clone(),
                Some(file_fingerprint),
                next_environments,
            );
            if public_catalog_changed {
                ProviderRefreshEvent::new(&previous, &snapshot).log(file_source.root());
            }
            (state.snapshot.clone(), transition)
        };
        settle_transition_async(transition).await;
        Ok(active)
    }

    fn snapshot_after_refresh_failure(
        &self,
        file_source: &FileProviderSource,
        code: &str,
        message: &str,
    ) -> Arc<RegistrySnapshot> {
        tracing::warn!(
            root = %file_source.root().display(),
            code,
            message,
            "provider directory refresh failed; keeping the last valid snapshot active"
        );
        self.snapshot()
    }

    /// Builds and validates a snapshot from the given providers without
    /// installing it, for pre-flight checking a candidate reload.
    pub fn validate_reload(
        &self,
        providers: Vec<Arc<dyn Provider>>,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let (_, _, snapshot) = build_registry(providers)?;
        Ok(snapshot)
    }

    /// Atomically replaces the loaded providers and active snapshot,
    /// clearing any file fingerprint.
    pub fn reload(
        &self,
        providers: Vec<Arc<dyn Provider>>,
    ) -> Result<Arc<RegistrySnapshot>, ProviderValidationError> {
        let (providers, core_registry, snapshot) = build_registry(providers)?;
        let mut state = self
            .state
            .write()
            .expect("provider registry lock should not be poisoned");
        let transition = publish_generation(
            &mut state,
            providers,
            core_registry,
            snapshot,
            None,
            PythonProviderEnvironmentSelections::new(),
        );
        let active = state.snapshot.clone();
        drop(state);
        settle_transition(transition);
        Ok(active)
    }

    /// Routes a provider call to its owning provider, enforcing pre-input
    /// checks, declared capabilities, and the response size limit around the
    /// provider's own `call`.
    pub async fn dispatch(&self, mut call: ProviderCall) -> Result<ProviderOutput, ProviderError> {
        let (snapshot, core_registry, lease, tool, capabilities, provider_kind) = {
            let state = self
                .state
                .read()
                .expect("provider registry lock should not be poisoned");
            let snapshot = state.snapshot.clone();
            let entry = snapshot.core_snapshot().tool(&call.action).ok_or_else(|| {
                ProviderError::validation(
                    "registry",
                    call.action.clone(),
                    "unknown_action",
                    format!("unknown provider action `{}`", call.action),
                )
            })?;
            let provider_name = entry.provider_id().as_str();
            let tool = entry.spec().clone();
            let provider = state.providers.get(provider_name).cloned().ok_or_else(|| {
                ProviderError::new(
                    "provider_not_loaded",
                    provider_name,
                    Some(entry.spec().name.clone()),
                    "provider is not loaded in the active registry",
                    "Reload providers and retry.",
                )
            })?;
            let lease = DispatchLease::acquire(provider).ok_or_else(|| {
                ProviderError::new(
                    "provider_generation_draining",
                    provider_name,
                    Some(entry.spec().name.clone()),
                    "provider generation is draining",
                    "Retry against the active provider generation.",
                )
            })?;
            let (capabilities, provider_kind) = snapshot
                .catalogs
                .iter()
                .find(|catalog| catalog.provider.name == provider_name)
                .map(|catalog| (catalog.capabilities.clone(), catalog.provider.kind))
                .expect("core provider index must reference an active catalog");
            (
                snapshot,
                state.core_registry.clone(),
                lease,
                tool,
                capabilities,
                provider_kind,
            )
        };

        call.provider = lease.provider.catalog().provider.name;
        call.snapshot_id = snapshot.id.clone();
        let pre_input_call = call.clone();
        let invocation_call = call.clone();
        let pre_input_tool = tool.clone();
        let capability_broker = self.capabilities.clone();
        let progress = soma_provider_core::ProviderProgressReporter::default();
        let mut output = core_registry
            .dispatch_with_pre_input(
                call.provider_invocation_with_progress(progress.clone()),
                move |invocation| {
                    let mut call = pre_input_call;
                    call.provider.clone_from(&invocation.provider);
                    call.snapshot_id.clone_from(&invocation.snapshot_id);
                    enforce_pre_input(&pre_input_tool, &call, provider_kind)
                },
                move |_, invocation| {
                    let mut call = invocation_call;
                    call.provider = invocation.provider;
                    call.snapshot_id = invocation.snapshot_id;
                    call.progress = invocation.context.progress;
                    async move {
                        enforce_capabilities(&capabilities, &call, &capability_broker)?;
                        let output = lease.provider.call(call.clone()).await?;
                        enforce_response_limit(&tool, &call, &output)?;
                        Ok(output)
                    }
                },
            )
            .await
            .inspect_err(|error| {
                let (provider, action, code) = error.log_code();
                tracing::warn!(provider, action, code, "provider call failed");
            })?;
        output.progress.extend(progress.events());
        Ok(output)
    }
}

/// Wraps a product-neutral `soma_provider_core::Provider` (as implemented by
/// every adapter in `soma-provider-adapters`) so it satisfies this crate's
/// own `Provider` trait, which carries additional auth/scope fields
/// (`principal`, `auth_mode`, `destructive_confirmed`, `limits`) that no
/// shared adapter reads — see `ProviderCall::provider_invocation()`, which
/// this reuses to build the generic call. This is the mirror image of
/// `CoreProviderAdapter` below, which wraps the other direction.
#[derive(Clone)]
pub struct SharedAdapter(Arc<dyn soma_provider_core::Provider>);

impl SharedAdapter {
    /// Wraps a shared core provider so it satisfies this crate's `Provider`
    /// trait.
    pub fn wrap(inner: Arc<dyn soma_provider_core::Provider>) -> Arc<dyn Provider> {
        Arc::new(Self(inner))
    }
}

#[async_trait]
impl Provider for SharedAdapter {
    fn catalog(&self) -> ProviderCatalog {
        self.0.catalog()
    }

    async fn call(&self, call: ProviderCall) -> Result<ProviderOutput, ProviderError> {
        self.0.call(call.provider_invocation()).await
    }

    async fn retire(&self) {
        self.0.retire().await;
    }

    async fn suspend(&self) {
        self.0.suspend().await;
    }

    fn runtime_status(&self) -> Option<Value> {
        self.0.runtime_status()
    }

    fn cancel_active(&self) -> bool {
        self.0.cancel_active()
    }

    async fn reset_quarantine(&self) {
        self.0.reset_quarantine().await;
    }

    fn deactivate(&self) {
        self.0.deactivate();
    }

    fn activate(&self) {
        self.0.activate();
    }

    fn acquire_dispatch(&self) -> bool {
        self.0.acquire_dispatch()
    }

    fn release_dispatch(&self) {
        self.0.release_dispatch();
    }
}

#[derive(Clone)]
struct CoreProviderAdapter(Arc<dyn Provider>);

type BuiltRegistry = (
    BTreeMap<String, Arc<dyn Provider>>,
    CoreRegistry,
    Arc<RegistrySnapshot>,
);

#[async_trait]
impl CoreProvider for CoreProviderAdapter {
    fn catalog(&self) -> ProviderCatalog {
        self.0.catalog()
    }

    async fn call(&self, call: CoreProviderCall) -> Result<ProviderOutput, ProviderError> {
        let surface = match call.surface {
            soma_provider_core::ProviderSurface::Mcp => ProviderSurface::Mcp,
            soma_provider_core::ProviderSurface::Rest => ProviderSurface::Rest,
            soma_provider_core::ProviderSurface::Cli => ProviderSurface::Cli,
            soma_provider_core::ProviderSurface::Palette => ProviderSurface::Palette,
            soma_provider_core::ProviderSurface::Internal
            | soma_provider_core::ProviderSurface::Ui => {
                return Err(ProviderError::validation(
                    call.provider,
                    call.action,
                    "unsupported_product_surface",
                    "Soma providers expose only MCP, REST, CLI, and Palette surfaces",
                ));
            }
        };
        self.0
            .call(ProviderCall {
                provider: call.provider,
                action: call.action,
                params: call.params,
                principal: ProviderPrincipal::anonymous(),
                auth_mode: ProviderAuthMode::TrustedGateway,
                surface,
                destructive_confirmed: false,
                limits: ProviderRequestLimits::default(),
                snapshot_id: call.snapshot_id,
                request_id: call.context.request_id,
                traceparent: call.context.traceparent,
                tracestate: call.context.tracestate,
                progress: call.context.progress,
            })
            .await
    }

    fn acquire_dispatch(&self) -> bool {
        self.0.acquire_dispatch()
    }

    fn release_dispatch(&self) {
        self.0.release_dispatch();
    }
}

fn build_registry(
    providers: Vec<Arc<dyn Provider>>,
) -> Result<BuiltRegistry, ProviderValidationError> {
    let providers = provider_map(providers)?;
    let mut builder = CoreRegistry::builder();
    for provider in providers.values() {
        validate_provider_manifest(&provider.catalog())?;
        builder = builder.register_arc(Arc::new(CoreProviderAdapter(provider.clone())))?;
    }
    let core_registry = builder.build()?;
    let snapshot = Arc::new(build_snapshot(&providers, &core_registry)?);
    Ok((providers, core_registry, snapshot))
}

fn build_snapshot(
    providers: &BTreeMap<String, Arc<dyn Provider>>,
    core_registry: &CoreRegistry,
) -> Result<RegistrySnapshot, ProviderValidationError> {
    let core = core_registry.snapshot();
    let mut resources = ResourceIndex::new();
    for provider in providers.values() {
        let catalog = provider.catalog();
        resources.register(&**provider, &catalog.provider.name, &catalog.resources)?;
    }
    let catalogs = core.catalogs().to_vec();
    let fingerprint = core.fingerprint().to_string();
    let id = fingerprint.clone();
    let action_names = core.action_names().map(str::to_owned).collect::<Vec<_>>();
    let openapi_paths = openapi_paths_from_core(&core);
    let cached_catalog_summary = Arc::new(json!({
        "schema_version": 1,
        "provider_fingerprint": fingerprint,
        "actions": action_names,
    }));
    let cached_openapi_bytes = Arc::new(
        serde_json::to_vec_pretty(&json!({
            "openapi": "3.1.0",
            "info": {"title": "soma provider API", "version": env!("CARGO_PKG_VERSION")},
            "x-soma": {
                "preferred_rest_style": "direct_routes",
                "provider_fingerprint": fingerprint
            },
            "paths": openapi_paths
        }))
        .expect("static OpenAPI summary serializes"),
    );

    Ok(RegistrySnapshot {
        id,
        fingerprint,
        catalogs,
        core: core.clone(),
        exact_resources: resources.exact,
        dynamic_resources: resources.dynamic,
        compiled_validator_count: core.compiled_validator_count(),
        cached_openapi_bytes,
        cached_catalog_summary,
    })
}