udb 0.4.25

//! Per-backend plugin trait — the U2 seam.
//!
//! `Backend` is the single, object-safe trait that a backend module implements
//! to declare its identity, capabilities, and (in later steps) its generation
//! and executor factories. The `all_plugins()` registry is the **one** place
//! where backends are enumerated; everything else iterates that list.
//!
//! Per §9.1 the executor returned by a plugin must be a *stateless leaf I/O*
//! adapter — replica routing, cache, encryption, channels, and circuit breakers
//! stay in `DataBrokerRuntime` orchestration, not in this trait.
//!
//! ## Object safety
//!
//! All trait methods are `&self` and return owned values or sized types, so
//! `&dyn Backend` and `Box<dyn Backend>` both work. The plugin structs in
//! `backend::plugins::*` are zero-sized, so each has a `pub static` instance
//! and the registry is a slice of `'static` references.
//!
//! The trait currently covers identity + capability + DSN scheme methods, with
//! defaults that delegate to `BackendKind`. Generation and executor hooks can
//! be added behind this same seam when those call sites are folded into the
//! plugin inventory.

use crate::backend::{BackendCapability, BackendCapabilityMatrixEntry, BackendKind};
use crate::runtime::config::{BackendInstanceConfig, UdbConfig};
use crate::runtime::core::{DataBrokerRuntime, RuntimeInitReport};
use serde::{Deserialize, Serialize};

/// Runtime implementation state for a known backend token.
///
/// `BackendKind` is intentionally broader than the current runtime. This type
/// keeps operators from mistaking "UDB knows the name" for "this binary can
/// execute it".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BackendSupportState {
    /// The token does not map to any known backend.
    Unknown,
    /// UDB has metadata for this backend, but no runtime implementation yet.
    KnownUnsupported,
    /// UDB has a runtime implementation, but this binary was built without it.
    DisabledByFeature,
    /// This binary includes the runtime plugin for the backend.
    RuntimeSupported,
}

impl BackendSupportState {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Unknown => "unknown",
            Self::KnownUnsupported => "known_unsupported",
            Self::DisabledByFeature => "disabled_by_feature",
            Self::RuntimeSupported => "runtime_supported",
        }
    }

    pub fn is_runtime_supported(self) -> bool {
        matches!(self, Self::RuntimeSupported)
    }

    pub fn diagnostic(self, backend: &str) -> String {
        match self {
            Self::Unknown => format!("backend '{backend}' is not known to UDB"),
            Self::KnownUnsupported => format!(
                "backend '{backend}' is known to UDB metadata but has no runtime executor in this version"
            ),
            Self::DisabledByFeature => format!(
                "backend '{backend}' is not available in this binary; rebuild with the matching Cargo feature"
            ),
            Self::RuntimeSupported => format!("backend '{backend}' is runtime-supported"),
        }
    }
}

/// One stable surface a backend plugin must provide.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendPluginSurface {
    pub name: String,
    pub description: String,
    pub required: bool,
}

impl BackendPluginSurface {
    pub fn required(name: &str, description: &str) -> Self {
        Self {
            name: name.to_string(),
            description: description.to_string(),
            required: true,
        }
    }
}

/// Machine-readable contract for adding or validating a backend plugin.
///
/// This is deliberately higher-level than the Rust trait methods. It describes
/// the public surfaces a plugin owns so external plugin crates can conform
/// without learning `runtime/core` internals.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendPluginContract {
    pub backend: String,
    pub tier: String,
    pub config_schema: BackendPluginSurface,
    pub connection_factory: BackendPluginSurface,
    pub logical_ir_compiler: BackendPluginSurface,
    pub executor: BackendPluginSurface,
    pub migration_resource_applier: BackendPluginSurface,
    pub health_probe: BackendPluginSurface,
    pub metrics_labels: Vec<String>,
    pub capability_matrix: BackendCapabilityMatrixEntry,
}

impl BackendPluginContract {
    pub fn default_for(kind: BackendKind) -> Self {
        Self {
            backend: kind.as_str().to_string(),
            tier: kind.tier().as_str().to_string(),
            config_schema: BackendPluginSurface::required(
                "BackendInstanceConfig",
                "Declarative config schema for DSN, labels, pool limits, routing, and feature gates",
            ),
            connection_factory: BackendPluginSurface::required(
                "Backend::register",
                "Startup-time connection/client factory owned by the plugin",
            ),
            logical_ir_compiler: BackendPluginSurface::required(
                "ir::compile",
                "Compiler from neutral logical operations into backend wire requests",
            ),
            executor: BackendPluginSurface::required(
                "runtime::executors::DispatchFactory",
                "Runtime executor for generic query/mutate/search/object/resource operations",
            ),
            migration_resource_applier: BackendPluginSurface::required(
                "Backend::generate_artifacts",
                "Migration/resource artifact generator and applier ownership point",
            ),
            health_probe: BackendPluginSurface::required(
                "probe",
                "Backend-specific health probe used by startup, reload, and admin APIs",
            ),
            metrics_labels: vec![
                "project".to_string(),
                "backend".to_string(),
                "tier".to_string(),
                "instance".to_string(),
                "operation".to_string(),
            ],
            capability_matrix: kind.capability_matrix_entry(),
        }
    }

    pub fn conformance_report(&self) -> BackendConformanceReport {
        let mut failures = Vec::new();
        let required_surfaces = [
            &self.config_schema,
            &self.connection_factory,
            &self.logical_ir_compiler,
            &self.executor,
            &self.migration_resource_applier,
            &self.health_probe,
        ];
        for surface in required_surfaces {
            if surface.required
                && (surface.name.trim().is_empty() || surface.description.trim().is_empty())
            {
                failures.push(format!("required surface '{}' is incomplete", surface.name));
            }
        }
        for label in ["backend", "tier", "instance", "operation"] {
            if !self.metrics_labels.iter().any(|item| item == label) {
                failures.push(format!("missing required metrics label '{label}'"));
            }
        }
        if self.capability_matrix.backend != self.backend {
            failures.push("capability matrix backend does not match contract backend".to_string());
        }
        if self.capability_matrix.tier != self.tier {
            failures.push("capability matrix tier does not match contract tier".to_string());
        }
        if self.capability_matrix.operations.is_empty() {
            failures.push("capability matrix must declare at least one operation".to_string());
        }
        // B.3/#40: capability-evidence summary. Static compiler availability
        // comes from the V2 model, but the conformance report's
        // `compiler_mediated` bit is stricter: it means a runtime path is wired
        // to consume the neutral IR compiler output for this backend. Do not
        // claim compiler mediation for raw generic-dispatch paths.
        let kind = BackendKind::from_token(&self.backend);
        let (
            native_executor,
            compiler_mediated,
            lifecycle,
            system_store,
            canonical_candidate,
            canonical_goal,
        ) = match kind {
            Some(k) => {
                let v2 = k.capabilities_v2();
                (
                    v2.native_executor,
                    compiler_mediated_runtime_path_wired(&k),
                    v2.lifecycle.as_str().to_string(),
                    v2.system_store.as_str().to_string(),
                    v2.canonical_candidate.as_str().to_string(),
                    v2.canonical_goal.to_string(),
                )
            }
            None => {
                failures.push(format!(
                    "contract backend '{}' does not resolve to a BackendKind",
                    self.backend
                ));
                (
                    false,
                    false,
                    "unknown".to_string(),
                    "unknown".to_string(),
                    "unknown".to_string(),
                    String::new(),
                )
            }
        };
        BackendConformanceReport {
            backend: self.backend.clone(),
            passed: failures.is_empty(),
            failures,
            native_executor,
            compiler_mediated,
            lifecycle,
            system_store,
            canonical_candidate,
            canonical_goal,
        }
    }
}

pub(crate) fn compiler_mediated_runtime_path_wired(kind: &BackendKind) -> bool {
    // GenericDispatch now accepts a neutral `ir` envelope, lowers it through
    // `ir::compile::compile_for_backend`, converts the `CompiledRendering` into
    // the existing executor request shape, and executes it on the same
    // channel/breaker/context path as raw dispatch. Keep this stricter than V2:
    // KV/object compilers exist for planner use, but V2 intentionally does not
    // classify those backends as compiler-mediated data-plane targets.
    if !kind.capabilities_v2().compiler_mediated {
        return false;
    }
    // Whether a compiler is actually compiled into THIS build is owned by
    // `ir::compile` (the same `#[cfg(..)]` arms that drive `compile_for_backend`).
    // Defer to it instead of re-listing the feature gates here, so the two can
    // never drift. The V2 gate above is the policy filter (KV/object stores
    // intentionally are not classified as compiler-mediated data-plane targets
    // even though planner-only compilers exist for them).
    crate::ir::compile::is_mediated_backend(kind)
}

/// Result of validating one plugin against the stable contract.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendConformanceReport {
    pub backend: String,
    pub passed: bool,
    pub failures: Vec<String>,
    // ── B.3 capability-evidence fields ─────────────────────────────────────────
    /// V2 dimension 1: backend ships a compiled-in runtime executor.
    #[serde(default)]
    pub native_executor: bool,
    /// True only when runtime dispatch is mediated by the neutral IR compiler.
    /// Static compiler availability lives in `BackendCapabilityV2`; this report
    /// is intentionally stricter so raw backend-shaped dispatch is not
    /// over-reported as compiler-mediated.
    #[serde(default)]
    pub compiler_mediated: bool,
    /// V2 dimension 4: resource-lifecycle kind
    /// (`none`/`compiler_mediated`/`native`/`catalog_migration`).
    #[serde(default)]
    pub lifecycle: String,
    /// V2 dimension 5: canonical system-store support (`none`/`full`).
    #[serde(default)]
    pub system_store: String,
    /// B.12-B.15: canonical promotion roadmap bucket.
    #[serde(default)]
    pub canonical_candidate: String,
    /// B.12-B.15: concrete promotion/conformance goal.
    #[serde(default)]
    pub canonical_goal: String,
}

/// Backends that have a runtime implementation in this source tree.
///
/// Feature gates decide whether each implementation is compiled into a given
/// binary; unsupported-but-known metadata-only backends stay out of this set.
pub fn has_runtime_implementation(kind: &BackendKind) -> bool {
    matches!(
        kind,
        BackendKind::Postgres
            // P2P: MySQL + SQLite are now first-class canonical stores
            // (via sqlx-mysql and sqlx-sqlite). See
            // `runtime/canonical_store/{mysql,sqlite}.rs`.
            | BackendKind::Mysql
            | BackendKind::Sqlite
            | BackendKind::Redis
            | BackendKind::Qdrant
            | BackendKind::Minio
            | BackendKind::S3
            | BackendKind::Mongodb
            | BackendKind::Neo4j
            | BackendKind::Clickhouse
            // These all have runtime executors under `runtime/executors/`; they
            // were missing here, so feature-disabled builds wrongly reported
            // KnownUnsupported (a hard lint error) instead of DisabledByFeature.
            | BackendKind::Mssql
            | BackendKind::Memcached
            | BackendKind::Elasticsearch
            | BackendKind::Weaviate
            | BackendKind::Pinecone
            | BackendKind::Cassandra
            | BackendKind::AzureBlob
            | BackendKind::Gcs
    )
}

/// Implementation state for a parsed backend kind.
pub fn support_state_for_kind(kind: &BackendKind) -> BackendSupportState {
    if all_plugins()
        .into_iter()
        .any(|plugin| plugin.kind() == *kind)
    {
        BackendSupportState::RuntimeSupported
    } else if has_runtime_implementation(kind) {
        BackendSupportState::DisabledByFeature
    } else {
        BackendSupportState::KnownUnsupported
    }
}

/// Implementation state for a backend token or alias.
pub fn support_state_for_token(token: &str) -> BackendSupportState {
    let Some(kind) =
        BackendKind::from_store_kind("", token).or_else(|| BackendKind::from_token(token))
    else {
        return BackendSupportState::Unknown;
    };
    support_state_for_kind(&kind)
}

/// Mutable context passed to [`Backend::register`].
///
/// Holds borrows of the partially-built runtime/report being assembled by
/// `DataBrokerRuntime::from_config` so each plugin can wire its own pools,
/// instance maps, and report flags without that function knowing the backend
/// list.
pub struct RegisterCtx<'a> {
    /// User-supplied configuration (read-only).
    pub config: &'a UdbConfig,
    /// Derived per-instance config (read-only) — labels, DSNs, etc.
    pub instance_config: &'a BackendInstanceConfig,
    /// Effective application name used in DSN labels.
    pub app_name: &'a str,
    /// The runtime being built. Plugins write their executor/client into this.
    pub runtime: &'a mut DataBrokerRuntime,
    /// The init report being built. Plugins set their `*_configured` flag and
    /// push any warnings encountered during setup.
    pub report: &'a mut RuntimeInitReport,
}

/// A plugin module describing one backend.
///
/// Implementors are typically zero-sized structs with one `pub static`
/// instance per backend (see `backend::plugins`).
#[async_trait::async_trait]
pub trait Backend: Send + Sync {
    /// The canonical identity of this backend.
    ///
    /// All other defaulted methods derive their value from this; an override
    /// is only needed if a plugin wants to deviate from the enum's metadata.
    fn kind(&self) -> BackendKind;

    /// Whether this backend is compiled into the current build.
    ///
    /// Defaulted to `true`. Plugins for feature-gated backends (s3, redis,
    /// kafka, …) override this by sitting behind `#[cfg(feature = "…")]`
    /// themselves; the slim build simply doesn't include them in
    /// [`all_plugins`].
    fn enabled(&self) -> bool {
        true
    }

    /// Capability flags (delegates to `BackendKind`).
    fn capabilities(&self) -> BackendCapability {
        self.kind().capabilities()
    }

    /// `udb+<tier>+<backend>` URI scheme prefix (delegates to `BackendKind`).
    fn dsn_scheme(&self) -> String {
        self.kind().dsn_scheme()
    }

    /// Default environment variable holding the connection DSN
    /// (delegates to `BackendKind`).
    fn default_dsn_env(&self) -> &'static str {
        self.kind().default_env_key()
    }

    /// Stable plugin contract advertised by this backend.
    fn contract(&self) -> BackendPluginContract {
        BackendPluginContract::default_for(self.kind())
    }

    /// Validate this plugin's contract.
    fn conformance_report(&self) -> BackendConformanceReport {
        self.contract().conformance_report()
    }

    /// Wire this backend into the runtime under construction.
    ///
    /// Default: no-op. Plugins override this to move their per-backend setup
    /// block out of `DataBrokerRuntime::from_config` (pool creation, HTTP
    /// client, instance map, `*_configured` flag, any warnings). Per §9.1 this
    /// is the *connection* layer only — replica routing, channels, breakers,
    /// cache, and encryption stay in orchestration.
    async fn register(&self, _ctx: &mut RegisterCtx<'_>) {}

    /// Generate the migration artifacts this backend ships for the given AST.
    ///
    /// Default: empty. Plugins for backends with a generator
    /// (`generation::<backend>::generate_<backend>_artifacts`) override this to
    /// delegate. `BackendKind::Postgres` keeps the default — its bootstrap SQL
    /// is produced via the separate `generate_bootstrap_sql` path because it
    /// also takes a `CatalogManifest`.
    fn generate_artifacts(
        &self,
        _manifest: &crate::generation::CatalogManifest,
        _sql_config: &crate::generation::sql::SqlGenerationConfig,
    ) -> Result<Vec<crate::generation::GeneratedArtifact>, String> {
        Ok(Vec::new())
    }

    /// Subdirectory name under `db_ops/` where this backend's artifacts are
    /// written by `sync_all_backends`. Defaults to the canonical token; only
    /// override when the directory layout deviates (none today).
    fn sync_subdir(&self) -> &'static str {
        self.kind().as_str()
    }
}

/// All compiled-in backend plugins, in canonical order.
///
/// Cargo features that drop a backend (`s3`, `redis`, `kafka`, …) also drop
/// its plugin via `#[cfg]` in `backend::plugins::mod`. This is the **single**
/// enumeration point: registry build, generation dispatch, and resolver lookup
/// all iterate this list rather than matching on `BackendKind` themselves.
pub fn all_plugins() -> Vec<&'static dyn Backend> {
    crate::backend::plugins::all()
}

/// Find a plugin by canonical token (alias of `BackendKind::from_token`).
///
/// Returns `None` for unknown tokens or for backends whose feature is disabled.
pub fn plugin_for(token: &str) -> Option<&'static dyn Backend> {
    let kind = BackendKind::from_token(token)?;
    all_plugins().into_iter().find(|p| p.kind() == kind)
}

/// Find a compiled runtime plugin by parsed backend kind.
pub fn plugin_for_kind(kind: &BackendKind) -> Option<&'static dyn Backend> {
    all_plugins().into_iter().find(|p| p.kind() == *kind)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn expected_runtime_plugin_kinds() -> Vec<BackendKind> {
        vec![
            BackendKind::Postgres,
            // NW3-1 + NW3-2: MySQL + SQLite sit at the SQL tier
            // next to Postgres.
            #[cfg(feature = "mysql")]
            BackendKind::Mysql,
            #[cfg(feature = "sqlite")]
            BackendKind::Sqlite,
            #[cfg(feature = "redis")]
            BackendKind::Redis,
            #[cfg(feature = "qdrant")]
            BackendKind::Qdrant,
            #[cfg(feature = "s3")]
            BackendKind::Minio,
            #[cfg(feature = "s3")]
            BackendKind::S3,
            #[cfg(feature = "mongodb")]
            BackendKind::Mongodb,
            #[cfg(feature = "neo4j")]
            BackendKind::Neo4j,
            #[cfg(feature = "clickhouse")]
            BackendKind::Clickhouse,
            // C9: Elasticsearch promoted from metadata-only.
            #[cfg(feature = "elasticsearch")]
            BackendKind::Elasticsearch,
            // C9: Memcached promoted from metadata-only via the
            // `memcache` crate.
            #[cfg(feature = "memcached")]
            BackendKind::Memcached,
            // C9: SQL Server promoted via the tiberius driver.
            #[cfg(feature = "mssql")]
            BackendKind::Mssql,
            // C9: Weaviate (REST + GraphQL via reqwest).
            #[cfg(feature = "weaviate")]
            BackendKind::Weaviate,
            // C9: Pinecone (REST via reqwest).
            #[cfg(feature = "pinecone")]
            BackendKind::Pinecone,
            // C9: Cassandra / ScyllaDB via the `scylla` driver.
            #[cfg(feature = "cassandra")]
            BackendKind::Cassandra,
            // C9: Azure Blob Storage via the Azure SDK.
            #[cfg(feature = "azureblob")]
            BackendKind::AzureBlob,
            // C9: Google Cloud Storage via the google-cloud-storage crate.
            #[cfg(feature = "gcs")]
            BackendKind::Gcs,
        ]
    }

    #[test]
    fn wired_classification_agrees_with_single_source_of_truth() {
        // Anti-drift guard for master-plan item 2.3: `compiler_mediated_runtime_path_wired`
        // no longer hand-lists per-backend `cfg!()` results. It must equal the V2 policy
        // filter ANDed with `ir::compile::is_mediated_backend` (the one source of truth for
        // "this build has a compiler for `kind`"). If the two ever disagree, the hand-list
        // has crept back in.
        for kind in BackendKind::all_known() {
            let expected = kind.capabilities_v2().compiler_mediated
                && crate::ir::compile::is_mediated_backend(kind);
            assert_eq!(
                compiler_mediated_runtime_path_wired(kind),
                expected,
                "{kind:?}: wired classification diverged from V2 filter + is_mediated_backend"
            );
        }
    }

    #[test]
    fn registry_is_non_empty_and_contains_postgres() {
        let plugins = all_plugins();
        assert!(!plugins.is_empty(), "registry must not be empty");
        assert!(
            plugins.iter().any(|p| p.kind() == BackendKind::Postgres),
            "postgres plugin must always be registered"
        );
    }

    #[test]
    fn registry_matches_compiled_runtime_backends() {
        let actual: Vec<_> = all_plugins().into_iter().map(|p| p.kind()).collect();
        assert_eq!(actual, expected_runtime_plugin_kinds());
    }

    #[test]
    fn plugin_metadata_matches_backend_kind() {
        // Defaulted methods must agree with the underlying BackendKind for
        // every registered plugin — anything else would split the source of
        // truth in two.
        for plugin in all_plugins() {
            let kind = plugin.kind();
            assert_eq!(plugin.dsn_scheme(), kind.dsn_scheme(), "{kind:?}");
            assert_eq!(plugin.default_dsn_env(), kind.default_env_key(), "{kind:?}");
            assert_eq!(plugin.capabilities(), kind.capabilities(), "{kind:?}");
            assert_eq!(plugin.contract().backend, kind.as_str(), "{kind:?}");
        }
    }

    #[test]
    fn registered_plugins_satisfy_stable_contract() {
        for plugin in all_plugins() {
            let report = plugin.conformance_report();
            assert!(
                report.passed,
                "{} plugin contract failed: {:?}",
                report.backend, report.failures
            );
        }
    }

    struct ToyBackend;

    #[async_trait::async_trait]
    impl Backend for ToyBackend {
        fn kind(&self) -> BackendKind {
            BackendKind::Sqlite
        }
    }

    #[test]
    fn toy_backend_can_conform_without_runtime_core_edits() {
        let plugin = ToyBackend;
        let contract = plugin.contract();
        assert_eq!(contract.backend, "sqlite");
        assert!(plugin.conformance_report().passed);
        assert!(
            contract
                .capability_matrix
                .operations
                .contains(&"ping".to_string())
        );
    }

    #[test]
    fn plugin_for_round_trips_through_canonical_token() {
        for plugin in all_plugins() {
            let token = plugin.kind().as_str();
            let resolved = plugin_for(token).unwrap_or_else(|| {
                panic!("plugin_for({token}) returned None for registered plugin")
            });
            assert_eq!(resolved.kind(), plugin.kind());
        }
        assert!(plugin_for("not_a_backend").is_none());
    }

    #[test]
    fn support_state_distinguishes_metadata_from_runtime() {
        assert_eq!(
            support_state_for_kind(&BackendKind::Postgres),
            BackendSupportState::RuntimeSupported
        );
        // C9: MSSQL, Memcached, Elasticsearch are now RuntimeSupported
        // (plugin entries shipped). The KnownUnsupported anchor is
        // one of the remaining metadata-only backends.
        #[cfg(feature = "mysql")]
        assert_eq!(
            support_state_for_kind(&BackendKind::Mysql),
            BackendSupportState::RuntimeSupported
        );
        #[cfg(feature = "sqlite")]
        assert_eq!(
            support_state_for_kind(&BackendKind::Sqlite),
            BackendSupportState::RuntimeSupported
        );
        #[cfg(feature = "mssql")]
        assert_eq!(
            support_state_for_kind(&BackendKind::Mssql),
            BackendSupportState::RuntimeSupported
        );
        // C9 complete: every BackendKind is wired. There's no
        // KnownUnsupported in-tree anchor anymore — verify via
        // token lookup that unknown strings still map correctly.
        #[cfg(feature = "cassandra")]
        assert_eq!(
            support_state_for_kind(&BackendKind::Cassandra),
            BackendSupportState::RuntimeSupported
        );
        assert_eq!(
            support_state_for_token("not_a_backend"),
            BackendSupportState::Unknown
        );
        assert!(support_state_for_token("postgres").is_runtime_supported());
    }

    // ── B.3: plugin conformance against matrix + dispatch registry ─────────────

    #[test]
    fn every_plugin_kind_matches_its_contract() {
        // The plugin's BackendKind must agree with its advertised contract
        // (backend token + tier), so the inventory and the contract cannot drift.
        for plugin in all_plugins() {
            let kind = plugin.kind();
            let contract = plugin.contract();
            assert_eq!(
                contract.backend,
                kind.as_str(),
                "{kind:?}: contract backend token mismatch"
            );
            assert_eq!(
                contract.tier,
                kind.tier().as_str(),
                "{kind:?}: contract tier mismatch"
            );
            assert_eq!(
                contract.capability_matrix.backend,
                kind.as_str(),
                "{kind:?}: contract capability_matrix backend mismatch"
            );
        }
    }

    #[test]
    fn runtime_plugins_advertising_dispatch_ops_have_a_dispatch_factory() {
        // B.3: a runtime-supported plugin that advertises generic-dispatch
        // operations beyond the universal ping/probe MUST resolve to a
        // DispatchFactory (cross-checked against handle.rs, read-only). Otherwise
        // generic dispatch would admit an operation it cannot build an executor
        // for.
        use crate::runtime::executors::handle::dispatch_factory_for;
        for plugin in all_plugins() {
            let kind = plugin.kind();
            let ops = kind.supported_operations();
            let advertises_real_ops = ops.iter().any(|op| *op != "ping" && *op != "probe");
            if advertises_real_ops {
                assert!(
                    dispatch_factory_for(&kind).is_some(),
                    "{kind:?} advertises dispatch ops {ops:?} but has no DispatchFactory"
                );
            }
        }
    }

    #[test]
    fn conformance_report_carries_capability_evidence() {
        // B.3: every plugin's conformance report must carry honest capability
        // evidence. Runtime executor evidence mirrors V2; compiler mediation is
        // stricter than V2 and must only claim a wired runtime compiler path.
        for plugin in all_plugins() {
            let kind = plugin.kind();
            let report = plugin.conformance_report();
            assert!(
                report.passed,
                "{kind:?} conformance failed: {:?}",
                report.failures
            );
            let v2 = kind.capabilities_v2();
            assert_eq!(report.native_executor, v2.native_executor, "{kind:?}");
            assert_eq!(
                report.compiler_mediated,
                compiler_mediated_runtime_path_wired(&kind),
                "{kind:?}"
            );
            assert!(
                !report.compiler_mediated || v2.compiler_mediated,
                "{kind:?} cannot report compiler mediation without a static compiler"
            );
            assert_eq!(report.lifecycle, v2.lifecycle.as_str(), "{kind:?}");
            assert_eq!(report.system_store, v2.system_store.as_str(), "{kind:?}");
            assert_eq!(
                report.canonical_candidate,
                v2.canonical_candidate.as_str(),
                "{kind:?}"
            );
            assert_eq!(report.canonical_goal, v2.canonical_goal, "{kind:?}");
            // Registered plugins are runtime-supported, so native_executor must
            // be true (the evidence is not hollow).
            assert!(
                report.native_executor,
                "{kind:?} is a registered plugin but reports no native executor"
            );
        }
    }

    #[test]
    fn conformance_report_marks_wired_neutral_ir_dispatch_as_compiler_mediated() {
        for backend in [BackendKind::Mongodb, BackendKind::Clickhouse] {
            if let Some(plugin) = plugin_for_kind(&backend) {
                let report = plugin.conformance_report();
                assert!(
                    report.compiler_mediated,
                    "{backend:?} generic dispatch accepts neutral IR, compiles it, and executes the compiled rendering"
                );
            }
        }
    }

    #[test]
    fn known_but_compiled_out_backends_classify_as_disabled_by_feature() {
        // B.3: a backend that has a runtime implementation but whose plugin is
        // not in the current `all_plugins()` set (feature disabled) must classify
        // as DisabledByFeature, never KnownUnsupported. Since every known kind has
        // a runtime implementation in-tree, any kind absent from the live registry
        // is by definition feature-disabled.
        let live: Vec<BackendKind> = all_plugins().into_iter().map(|p| p.kind()).collect();
        for kind in BackendKind::all_known() {
            if live.contains(kind) {
                assert_eq!(
                    support_state_for_kind(kind),
                    BackendSupportState::RuntimeSupported,
                    "{kind:?} is in the live registry"
                );
            } else {
                assert_eq!(
                    support_state_for_kind(kind),
                    BackendSupportState::DisabledByFeature,
                    "{kind:?} is compiled out and must report DisabledByFeature, not \
                     KnownUnsupported"
                );
                assert!(
                    has_runtime_implementation(kind),
                    "{kind:?} must have a runtime implementation to be DisabledByFeature"
                );
            }
        }
    }
}