pub fn name(&self) -> &str {
        &self.id
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub async fn evaluate(&self, boundaries_json: String) -> Result<ReactorOutcome, LoaderError> {
        let name = self.name.clone();
        let inv_json =
            serde_json::to_string(&ReactorInvocation { boundaries_json }).map_err(|e| {
                LoaderError::Validation {
                    reason: format!("reactor constructor '{name}': serialize invocation: {e}"),
                }
            })?;

        let handle = self.handle.clone();
        let call_name = name.clone();
        let out_json: String = tokio::task::spawn_blocking(move || {
            handle.call_method::<_, String>(METHOD_EVALUATE, &(inv_json,))
        })
        .await
        .map_err(|e| LoaderError::Validation {
            reason: format!("reactor constructor '{call_name}': evaluate join: {e}"),
        })?
        .map_err(|e| LoaderError::Validation {
            reason: format!("reactor constructor '{call_name}': evaluate FFI call: {e}"),
        })?;

        serde_json::from_str::<ReactorOutcome>(&out_json).map_err(|e| LoaderError::Validation {
            reason: format!("reactor constructor '{name}': parse evaluate outcome: {e}"),
        })
    }

pub fn read_provider_manifest(package_dir: &Path) -> Result<ProviderManifest, LoaderError> {
    let path = package_dir.join(PROVIDER_MANIFEST_FILE);
    let raw = std::fs::read_to_string(&path).map_err(|e| LoaderError::FileSystem {
        path: path.display().to_string(),
        error: e.to_string(),
    })?;
    ProviderManifest::from_json(&raw).map_err(|e| LoaderError::ManifestParse {
        reason: format!("{PROVIDER_MANIFEST_FILE}: {e}"),
    })
}

fn read_member_manifest(
    package_dir: &Path,
    constructor_name: &str,
) -> Result<ConstructorManifest, LoaderError> {
    let provider = read_provider_manifest(package_dir)?;
    provider
        .constructor(constructor_name)
        .cloned()
        .ok_or_else(|| LoaderError::Validation {
            reason: format!(
                "provider '{}' has no constructor '{}'; members: [{}]",
                provider.name,
                constructor_name,
                provider
                    .constructors
                    .iter()
                    .map(|c| c.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
        })
}

fn wrap_member_config<C: Serialize>(
    constructor_name: &str,
    config: &C,
) -> Result<ProviderConfigure, LoaderError> {
    let inner = fidius_core::wire::serialize(config).map_err(|e| LoaderError::Validation {
        reason: format!("serialize config for constructor '{constructor_name}': {e}"),
    })?;
    Ok(ProviderConfigure {
        name: constructor_name.to_string(),
        config: inner,
    })
}

fn resolve_native_provider(
    search_path: &Path,
    package_name: &str,
) -> Result<Option<(std::path::PathBuf, ProviderManifest)>, LoaderError> {
    if !search_path.is_dir() {
        return Ok(None);
    }
    let entries = std::fs::read_dir(search_path).map_err(|e| LoaderError::FileSystem {
        path: search_path.display().to_string(),
        error: e.to_string(),
    })?;
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        // Only a directory that holds a parseable provider.json is a candidate.
        if let Ok(provider) = read_provider_manifest(&path) {
            if provider.name == package_name && provider.runtime == ProviderRuntime::Native {
                return Ok(Some((path, provider)));
            }
        }
    }
    Ok(None)
}

fn load_native_member<C: Serialize>(
    search_path: &Path,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    expected_kind: PrimitiveKind,
    expected_iface_version: u32,
    holder_plugin: &str,
) -> Result<Option<(fidius_host::PluginHandle, ConstructorManifest)>, LoaderError> {
    let Some((dir, provider)) = resolve_native_provider(search_path, package_name)? else {
        return Ok(None);
    };

    let member = provider
        .constructor(constructor_name)
        .cloned()
        .ok_or_else(|| LoaderError::Validation {
            reason: format!(
                "native provider '{}' has no constructor '{}'; members: [{}]",
                provider.name,
                constructor_name,
                provider
                    .constructors
                    .iter()
                    .map(|c| c.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
        })?;

    if member.primitive_kind != expected_kind {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' is {:?}, not {:?}",
                member.name, member.primitive_kind, expected_kind
            ),
        });
    }
    if member.interface_version != expected_iface_version {
        return Err(LoaderError::Validation {
            reason: format!(
                "native constructor '{}' declares interface v{}, loader supports v{}",
                member.name, member.interface_version, expected_iface_version
            ),
        });
    }

    // Same name-in-configure wrapper the WASM path uses (positional bincode config
    // tagged with the member name the suite shell name-dispatches on).
    let configure = wrap_member_config(constructor_name, config)?;
    let lib_path = dir.join(&provider.component);
    let loaded =
        fidius_host::loader::load_library(&lib_path).map_err(|e| LoaderError::LibraryLoad {
            path: lib_path.display().to_string(),
            // Name the host triple: a wrong-arch cdylib (a bundle built on a
            // different arch, CLOACI-T-0908) dies here and should read as an
            // arch mismatch, not corruption.
            error: format!(
                "load_library (native provider, host triple '{}'): {e} — if this is a \
                 wrong-arch cdylib, the per-target compiler has not filled this arch yet",
                crate::fleet::protocol::host_target_triple()
            ),
        })?;
    let plugin = loaded
        .plugins
        .into_iter()
        .find(|p| p.info.name == holder_plugin)
        .ok_or_else(|| LoaderError::Validation {
            reason: format!(
                "native provider '{}' cdylib '{}' has no '{}' plugin (kind {:?})",
                provider.name, provider.component, holder_plugin, expected_kind
            ),
        })?;
    let handle =
        fidius_host::PluginHandle::configure_from_loaded(plugin, &configure).map_err(|e| {
            LoaderError::LibraryLoad {
                path: lib_path.display().to_string(),
                error: format!("configure_from_loaded (native provider): {e}"),
            }
        })?;
    // CLOACI-T-0907: surface the trust tier at load — a native provider runs
    // unsandboxed in-process (grants advisory), unlike sandboxed wasm providers.
    tracing::info!(
        provider = %provider.name,
        constructor = %constructor_name,
        "loaded NATIVE constructor provider (trusted, unsandboxed in-process; grants advisory)"
    );
    Ok(Some((handle, member)))
}

fn exec_err(task_id: &str, message: impl Into<String>) -> TaskError {
    TaskError::ExecutionFailed {
        message: message.into(),
        task_id: task_id.to_string(),
        timestamp: Utc::now(),
    }
}

pub fn read_constructor_manifest(package_dir: &Path) -> Result<ConstructorManifest, LoaderError> {
    let path = package_dir.join(CONSTRUCTOR_MANIFEST_FILE);
    let raw = std::fs::read_to_string(&path).map_err(|e| LoaderError::FileSystem {
        path: path.display().to_string(),
        error: e.to_string(),
    })?;
    ConstructorManifest::from_json(&raw).map_err(|e| LoaderError::ManifestParse {
        reason: format!("{CONSTRUCTOR_MANIFEST_FILE}: {e}"),
    })
}

pub fn load_task_constructor<C: Serialize>(
    search_path: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    grants: &ResolvedGrants,
) -> Result<Arc<dyn Task>, LoaderError> {
    let search_path = search_path.as_ref();

    // NATIVE provider fast-path (CLOACI-T-0902): a `runtime = "native"` provider
    // loads in-process via `configure_from_loaded`; grants are advisory (no
    // sandbox). Falls through to the WASM path when the package isn't native.
    if let Some((handle, member)) = load_native_member(
        search_path,
        package_name,
        constructor_name,
        config,
        PrimitiveKind::Task,
        TASK_CONSTRUCTOR_INTERFACE_VERSION,
        "__ProviderTask",
    )? {
        return Ok(Arc::new(WasmTaskConstructor {
            id: member.name,
            handle: Arc::new(handle),
            dependencies: Vec::new(),
        }));
    }

    let host = PluginHost::builder()
        .search_path(search_path)
        .build()
        .map_err(|e| LoaderError::LibraryLoad {
            path: search_path.display().to_string(),
            error: format!("build plugin host: {e}"),
        })?;

    // Resolve the package directory so we can read its provider manifest + select
    // the requested member constructor.
    let dir = host
        .find_wasm_package(package_name)
        .map_err(|e| LoaderError::Validation {
            reason: format!("locate wasm constructor package '{package_name}': {e}"),
        })?;

    let manifest = read_member_manifest(&dir, constructor_name)?;

    if manifest.primitive_kind != PrimitiveKind::Task {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' is {:?}, not Task; trigger/accumulator/reactor loading is CLOACI-T-0824",
                manifest.name, manifest.primitive_kind
            ),
        });
    }

    if manifest.interface_version != TASK_CONSTRUCTOR_INTERFACE_VERSION {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' declares task-constructor interface v{}, loader supports v{}",
                manifest.name, manifest.interface_version, TASK_CONSTRUCTOR_INTERFACE_VERSION
            ),
        });
    }

    // Config binds ONCE here; the integrity hash is checked inside fidius. The
    // member name travels in the configure payload (name-in-configure) so the suite
    // shell instantiates the right member. The tenant's translated grants override
    // the manifest caps + supply the egress policy (default-closed when deny-all).
    let configure = wrap_member_config(constructor_name, config)?;
    let handle = host
        .load_wasm_configured_with_grants(
            package_name,
            &__fidius_TaskConstructor::TaskConstructor_WASM_DESCRIPTOR,
            &configure,
            grants.capabilities.clone(),
            grants.egress.clone(),
        )
        .map_err(|e| LoaderError::LibraryLoad {
            path: dir.display().to_string(),
            error: format!("load_wasm_configured_with_grants: {e}"),
        })?;

    Ok(Arc::new(WasmTaskConstructor {
        id: manifest.name,
        handle: Arc::new(handle),
        dependencies: Vec::new(),
    }))
}

pub fn unpack_provider_archive(
    archive: impl AsRef<Path>,
    dest: impl AsRef<Path>,
    verify_keys: &[ed25519_dalek::VerifyingKey],
) -> Result<std::path::PathBuf, LoaderError> {
    let archive = archive.as_ref();
    let dest = dest.as_ref();

    let pkg_dir = fidius_core::package::unpack_package(archive, dest).map_err(|e| {
        LoaderError::Validation {
            reason: format!("unpack provider package '{}': {e}", archive.display()),
        }
    })?;

    if !verify_keys.is_empty() {
        fidius_host::package::verify_package(&pkg_dir, verify_keys).map_err(|e| {
            LoaderError::Validation {
                reason: format!("verify provider signature for '{}': {e}", pkg_dir.display()),
            }
        })?;
    }

    Ok(pkg_dir)
}

pub fn load_task_constructor_from_package<C: Serialize>(
    archive: impl AsRef<Path>,
    dest: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    verify_keys: &[ed25519_dalek::VerifyingKey],
    grants: &ResolvedGrants,
) -> Result<Arc<dyn Task>, LoaderError> {
    let dest = dest.as_ref();
    // Unpack (+ verify) first; this is the seam that rejects a tampered or
    // unsigned-but-required archive before any wasm is loaded.
    let _pkg_dir = unpack_provider_archive(archive, dest, verify_keys)?;
    // `dest` is the fidius search path; the package resolves by name out of the
    // freshly-unpacked `<name>-<version>/` directory it contains.
    load_task_constructor(dest, package_name, constructor_name, config, grants)
}

pub fn load_trigger_constructor<C: Serialize>(
    search_path: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    binding: TriggerBinding,
    grants: &ResolvedGrants,
) -> Result<Arc<dyn Trigger>, LoaderError> {
    let search_path = search_path.as_ref();

    // NATIVE provider fast-path (CLOACI-T-0902): grants advisory; falls through
    // to the WASM path when the package isn't a `runtime = "native"` provider.
    if let Some((handle, member)) = load_native_member(
        search_path,
        package_name,
        constructor_name,
        config,
        PrimitiveKind::Trigger,
        TRIGGER_CONSTRUCTOR_INTERFACE_VERSION,
        "__ProviderTrigger",
    )? {
        return Ok(Arc::new(WasmTriggerConstructor {
            name: member.name,
            handle: Arc::new(handle),
            binding,
        }));
    }

    let host = PluginHost::builder()
        .search_path(search_path)
        .build()
        .map_err(|e| LoaderError::LibraryLoad {
            path: search_path.display().to_string(),
            error: format!("build plugin host: {e}"),
        })?;

    let dir = host
        .find_wasm_package(package_name)
        .map_err(|e| LoaderError::Validation {
            reason: format!("locate wasm constructor package '{package_name}': {e}"),
        })?;

    let manifest = read_member_manifest(&dir, constructor_name)?;

    if manifest.primitive_kind != PrimitiveKind::Trigger {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' is {:?}, not Trigger",
                manifest.name, manifest.primitive_kind
            ),
        });
    }

    if manifest.interface_version != TRIGGER_CONSTRUCTOR_INTERFACE_VERSION {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' declares trigger-constructor interface v{}, loader supports v{}",
                manifest.name, manifest.interface_version, TRIGGER_CONSTRUCTOR_INTERFACE_VERSION
            ),
        });
    }

    let configure = wrap_member_config(constructor_name, config)?;
    let handle = host
        .load_wasm_configured_with_grants(
            package_name,
            &__fidius_TriggerConstructor::TriggerConstructor_WASM_DESCRIPTOR,
            &configure,
            grants.capabilities.clone(),
            grants.egress.clone(),
        )
        .map_err(|e| LoaderError::LibraryLoad {
            path: dir.display().to_string(),
            error: format!("load_wasm_configured_with_grants: {e}"),
        })?;

    Ok(Arc::new(WasmTriggerConstructor {
        name: manifest.name,
        handle: Arc::new(handle),
        binding,
    }))
}

pub fn load_constructor<C: Serialize>(
    runtime: &Runtime,
    search_path: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    binding: ConstructorBinding,
    grants: &ResolvedGrants,
) -> Result<(), LoaderError> {
    let search_path = search_path.as_ref();

    // Peek the manifest so a mismatched binding fails before we pay for a load.
    let host = PluginHost::builder()
        .search_path(search_path)
        .build()
        .map_err(|e| LoaderError::LibraryLoad {
            path: search_path.display().to_string(),
            error: format!("build plugin host: {e}"),
        })?;
    let dir = host
        .find_wasm_package(package_name)
        .map_err(|e| LoaderError::Validation {
            reason: format!("locate wasm constructor package '{package_name}': {e}"),
        })?;
    let manifest = read_member_manifest(&dir, constructor_name)?;

    match (&binding, manifest.primitive_kind) {
        (ConstructorBinding::Task { namespace }, PrimitiveKind::Task) => {
            let task =
                load_task_constructor(search_path, package_name, constructor_name, config, grants)?;
            let namespace = namespace.clone();
            runtime.register_task(namespace, move || task.clone());
            Ok(())
        }
        (ConstructorBinding::Trigger(tb), PrimitiveKind::Trigger) => {
            let name = manifest.name.clone();
            let trigger = load_trigger_constructor(
                search_path,
                package_name,
                constructor_name,
                config,
                tb.clone(),
                grants,
            )?;
            runtime.register_trigger(name, move || trigger.clone());
            Ok(())
        }
        (_, kind) => Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' is {:?}, which does not match the supplied binding \
                 (accumulator/reactor registration is a CLOACI-T-0824 continuation)",
                manifest.name, kind
            ),
        }),
    }
}

pub fn load_accumulator_constructor<C: Serialize>(
    search_path: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
    grants: &ResolvedGrants,
) -> Result<WasmAccumulatorConstructor, LoaderError> {
    let search_path = search_path.as_ref();

    // NATIVE provider fast-path (CLOACI-T-0902; also T-0904's stream-accumulator
    // prereq): grants advisory; falls through to WASM when not a native provider.
    if let Some((handle, member)) = load_native_member(
        search_path,
        package_name,
        constructor_name,
        config,
        PrimitiveKind::Accumulator,
        ACCUMULATOR_CONSTRUCTOR_INTERFACE_VERSION,
        "__ProviderAccumulator",
    )? {
        return Ok(WasmAccumulatorConstructor {
            name: member.name,
            handle: Arc::new(handle),
        });
    }

    let host = PluginHost::builder()
        .search_path(search_path)
        .build()
        .map_err(|e| LoaderError::LibraryLoad {
            path: search_path.display().to_string(),
            error: format!("build plugin host: {e}"),
        })?;

    let dir = host
        .find_wasm_package(package_name)
        .map_err(|e| LoaderError::Validation {
            reason: format!("locate wasm constructor package '{package_name}': {e}"),
        })?;

    let manifest = read_member_manifest(&dir, constructor_name)?;

    if manifest.primitive_kind != PrimitiveKind::Accumulator {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' is {:?}, not Accumulator",
                manifest.name, manifest.primitive_kind
            ),
        });
    }

    if manifest.interface_version != ACCUMULATOR_CONSTRUCTOR_INTERFACE_VERSION {
        return Err(LoaderError::Validation {
            reason: format!(
                "constructor '{}' declares accumulator-constructor interface v{}, loader supports v{}",
                manifest.name, manifest.interface_version, ACCUMULATOR_CONSTRUCTOR_INTERFACE_VERSION
            ),
        });
    }

    let configure = wrap_member_config(constructor_name, config)?;
    let handle = host
        .load_wasm_configured_with_grants(
            package_name,
            &__fidius_AccumulatorConstructor::AccumulatorConstructor_WASM_DESCRIPTOR,
            &configure,
            grants.capabilities.clone(),
            grants.egress.clone(),
        )
        .map_err(|e| LoaderError::LibraryLoad {
            path: dir.display().to_string(),
            error: format!("load_wasm_configured_with_grants: {e}"),
        })?;

    Ok(WasmAccumulatorConstructor {
        name: manifest.name,
        handle: Arc::new(handle),
    })
}

pub async fn load_stream_accumulator_source<C: Serialize>(
    search_path: impl AsRef<Path>,
    package_name: &str,
    constructor_name: &str,
    config: &C,
) -> Result<ProviderStreamSource, LoaderError> {
    let search_path = search_path.as_ref();

    let (handle, member) = load_native_member(
        search_path,
        package_name,
        constructor_name,
        config,
        PrimitiveKind::Accumulator,
        STREAM_ACCUMULATOR_CONSTRUCTOR_INTERFACE_VERSION,
        "__ProviderStreamAccumulator",
    )?
    .ok_or_else(|| LoaderError::Validation {
        reason: format!(
            "provider '{package_name}' member '{constructor_name}' is not a native stream \
             accumulator (needs runtime = \"native\" + a stream-accumulator member; wasm \
             streaming parity is CLOACI-T-0906/0907)"
        ),
    })?;

    let handle = Arc::new(handle);
    // Start the server-streaming `source` (METHOD_SOURCE). The `seed` arg is unused
    // by the shell — a single-arg fidius method takes a `(T,)` tuple. Items decode
    // as boundary-JSON `String`s.
    let stream = handle
        .call_streaming::<(u64,), String>(METHOD_SOURCE, &(0u64,))
        .await
        .map_err(|e| LoaderError::LibraryLoad {
            path: package_name.to_string(),
            error: format!("call_streaming(source) on native stream accumulator: {e}"),
        })?;

    Ok(ProviderStreamSource {
        name: member.name,
        _handle: handle,
        stream,
    })
}

pub async fn load_stream_accumulator_source_from_config(
    provider_name: &str,
    constructor_name: &str,
    config: &std::collections::HashMap<String, String>,
) -> Result<ProviderStreamSource, LoaderError> {
    let search_path = provider_search_path();

    // Member manifest first — bind_config_by_name needs its config_fields.
    let (_dir, provider) =
        resolve_native_provider(&search_path, provider_name)?.ok_or_else(|| {
            LoaderError::Validation {
                reason: format!(
                "stream accumulator declares provider '{provider_name}' but no NATIVE provider \
                 by that name exists under '{}' (bundle it with the package, or check the \
                 provider/runtime — wasm stream providers are CLOACI-T-0906/0907 follow-up)",
                search_path.display()
            ),
            }
        })?;
    let member = provider
        .constructor(constructor_name)
        .ok_or_else(|| LoaderError::Validation {
            reason: format!(
                "provider '{provider_name}' has no constructor '{constructor_name}'; members: [{}]",
                provider
                    .constructors
                    .iter()
                    .map(|c| c.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
        })?
        .clone();

    // Map → JSON values (strings that parse as the declared numeric/bool type
    // coerce — package.toml accumulator config is stringly-typed).
    let author: Vec<(String, serde_json::Value)> = config
        .iter()
        .map(|(k, v)| {
            let field_ty = member
                .config_fields
                .iter()
                .find(|f| f.name == *k)
                .map(|f| f.ty.as_str());
            let value = match field_ty {
                Some("String") | Some("str") | None => serde_json::Value::String(v.clone()),
                Some("bool") => v
                    .parse::<bool>()
                    .map(serde_json::Value::Bool)
                    .unwrap_or_else(|_| serde_json::Value::String(v.clone())),
                // Numeric field: parse the string; a non-parsing value falls
                // through as a string so bind_config_by_name names the field in
                // its type error.
                Some(_) => serde_json::from_str::<serde_json::Value>(v)
                    .ok()
                    .filter(serde_json::Value::is_number)
                    .unwrap_or_else(|| serde_json::Value::String(v.clone())),
            };
            (k.clone(), value)
        })
        .collect();

    let ordered = bind_config_by_name(
        // Context strings for the fail-closed error messages.
        &format!("accumulator '{constructor_name}'"),
        provider_name,
        constructor_name,
        &member,
        author,
    )?;

    load_stream_accumulator_source(&search_path, provider_name, constructor_name, &ordered).await
}