lenso-plugin-bundle 0.4.0

//! Immutable Plugin Release manifests, source materialization, and Bundle verification.

mod model;
mod selection;

use std::{
    collections::{BTreeMap, BTreeSet},
    fmt, fs,
    io::Read as _,
    path::{Component, Path, PathBuf},
};

use lenso_app_plan::{
    CapabilityEndpointPlan, CapabilityOperationKind, CapabilityRequirementPlan, ExecutionClassId,
    authoring::{PluginContract, PluginDescriptor, PluginImplementation},
};
pub use model::*;
pub use selection::*;
use serde::{Deserialize, de::DeserializeOwned};
use serde_json::Value;
use sha2::{Digest, Sha256};

/// The only manifest filename accepted in a materialized Plugin Bundle.
pub const MANIFEST_FILE: &str = "lenso-plugin.json";

/// Custom section carrying source-derived Plugin descriptor bytes.
pub const PLUGIN_DESCRIPTOR_SECTION: &str = "lenso.plugin-descriptor.v1";

/// Maximum accepted source-derived descriptor size.
pub const MAX_PLUGIN_DESCRIPTOR_BYTES: usize = 64 * 1024;

/// Host-owned resource bounds for verifying an untrusted materialized Bundle.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BundleVerificationLimits {
    pub max_manifest_bytes: u64,
    pub max_file_bytes: u64,
    pub max_total_bytes: u64,
    pub max_file_count: usize,
    pub max_entry_count: usize,
    pub max_directory_depth: usize,
}

impl Default for BundleVerificationLimits {
    fn default() -> Self {
        Self {
            max_manifest_bytes: 1024 * 1024,
            max_file_bytes: 256 * 1024 * 1024,
            max_total_bytes: 512 * 1024 * 1024,
            max_file_count: 128,
            max_entry_count: 256,
            max_directory_depth: 32,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct BundleFileSummary {
    size: u64,
    digest: String,
}

/// Source-only input for one generated V2 Plugin Bundle.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourcePluginBuild {
    pub package_manifest: PathBuf,
    pub wasm_module: PathBuf,
    pub output: PathBuf,
}

/// Source-only input for one precompiled Process Plugin Bundle.
///
/// The descriptor is generated by the language SDK. Bundle construction never
/// executes the process; the Process Adapter repeats the descriptor handshake
/// before readiness.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourceProcessPluginBuild {
    pub package_manifest: PathBuf,
    pub executable: PathBuf,
    pub runtime_descriptor: PathBuf,
    pub authoring_version: u32,
    pub runtime_profile: String,
    pub target: String,
    pub output: PathBuf,
}

/// Source-only input for one V4 Plugin Release containing multiple implementations.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourcePluginReleaseBuild {
    pub contract: PluginContract,
    pub implementations: Vec<SourcePluginImplementation>,
    pub output: PathBuf,
}

/// One already-built implementation Artifact admitted to a V4 Plugin Release.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourcePluginImplementation {
    pub id: String,
    pub host_targets: Vec<String>,
    pub artifact: PathBuf,
    pub bundle_path: String,
    pub media_type: String,
    pub target: String,
    pub entrypoint: String,
    pub execution_class: ExecutionClassId,
    pub runtime_profile: String,
}

#[derive(Clone, Debug)]
struct SourceManifestDocument {
    value: PluginManifestV2,
    bytes: Vec<u8>,
    digest: String,
}

#[derive(Clone, Debug)]
struct ManifestDocument {
    value: PluginManifest,
    digest: String,
}

impl ManifestDocument {
    fn parse(input: &[u8]) -> Result<Self, BundleError> {
        let value = strict_json::<Value>(input)?;
        let schema_version = value
            .get("schema_version")
            .and_then(Value::as_u64)
            .ok_or_else(|| BundleError::InvalidManifest("schema_version is required".to_owned()))?;
        let value = match schema_version {
            2 => PluginManifest::V2(
                serde_json::from_value(value)
                    .map_err(|error| BundleError::InvalidManifest(error.to_string()))?,
            ),
            3 => {
                validate_profile_wire_shape(&value, false)?;
                PluginManifest::V3(
                    serde_json::from_value(value)
                        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?,
                )
            }
            4 => {
                validate_profile_wire_shape(&value, true)?;
                PluginManifest::V4(
                    serde_json::from_value(value)
                        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?,
                )
            }
            _ => return invalid_manifest("unsupported schema version"),
        };
        validate_manifest(&value)?;
        let canonical = canonical_manifest_bytes(&value)?;
        Ok(Self {
            value,
            digest: sha256_digest(&canonical),
        })
    }
}

fn canonical_manifest_bytes(manifest: &PluginManifest) -> Result<Vec<u8>, BundleError> {
    let mut value = match manifest {
        PluginManifest::V2(value) => serde_json::to_value(value),
        PluginManifest::V3(value) => serde_json::to_value(value),
        PluginManifest::V4(value) => serde_json::to_value(value),
    }
    .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    if matches!(manifest, PluginManifest::V3(_)) {
        let object = value
            .as_object_mut()
            .ok_or_else(|| BundleError::InvalidManifest("Manifest must be an object".to_owned()))?;
        object
            .get_mut("contract")
            .and_then(Value::as_object_mut)
            .and_then(|contract| contract.remove("authoring_version"));
        if let Some(implementations) = object
            .get_mut("implementations")
            .and_then(Value::as_array_mut)
        {
            for implementation in implementations {
                implementation
                    .get_mut("runtime")
                    .and_then(Value::as_object_mut)
                    .and_then(|runtime| runtime.remove("runtime_profile"));
            }
        }
    }
    serde_json::to_vec(&value).map_err(|error| BundleError::InvalidManifest(error.to_string()))
}

fn validate_profile_wire_shape(value: &Value, require_profiles: bool) -> Result<(), BundleError> {
    let contract = value
        .get("contract")
        .and_then(Value::as_object)
        .ok_or_else(|| BundleError::InvalidManifest("contract is required".to_owned()))?;
    let authoring = contract.get("authoring_version");
    if require_profiles != authoring.is_some() {
        return invalid_manifest(if require_profiles {
            "V4 contract requires authoring_version"
        } else {
            "V3 contract cannot contain authoring_version"
        });
    }
    let implementations = value
        .get("implementations")
        .and_then(Value::as_array)
        .ok_or_else(|| BundleError::InvalidManifest("implementations are required".to_owned()))?;
    for implementation in implementations {
        let runtime = implementation
            .get("runtime")
            .and_then(Value::as_object)
            .ok_or_else(|| BundleError::InvalidManifest("runtime is required".to_owned()))?;
        let profile = runtime.get("runtime_profile");
        if require_profiles {
            if !matches!(profile.and_then(Value::as_str), Some(value) if !value.trim().is_empty()) {
                return invalid_manifest("V4 implementation requires a non-empty runtime_profile");
            }
        } else if profile.is_some() {
            return invalid_manifest("V3 implementation cannot contain runtime_profile");
        }
    }
    Ok(())
}

impl SourceManifestDocument {
    #[cfg(test)]
    fn parse(input: &[u8]) -> Result<Self, BundleError> {
        let value = strict_json::<PluginManifestV2>(input)?;
        Self::from_value(value)
    }

    fn from_value(value: PluginManifestV2) -> Result<Self, BundleError> {
        validate_source_manifest(&value)?;
        let json = serde_json::to_value(&value)
            .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
        validate_json_value(&json)?;
        let bytes = serde_json::to_vec(&json)
            .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
        let digest = sha256_digest(&bytes);
        Ok(Self {
            value,
            bytes,
            digest,
        })
    }
}

#[derive(Debug, Deserialize)]
struct CargoManifest {
    package: CargoPackage,
}

#[derive(Debug, Deserialize)]
struct CargoPackage {
    version: String,
    metadata: CargoMetadata,
}

#[derive(Debug, Deserialize)]
struct CargoMetadata {
    lenso: CargoLensoMetadata,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "kebab-case")]
struct CargoLensoMetadata {
    plugin_id: String,
    root_slot: String,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct GuestRuntimeDescriptor {
    abi: String,
    capabilities: Vec<GuestCapability>,
    #[serde(default)]
    required_capabilities: Vec<GuestRequirement>,
    #[serde(default)]
    configuration_schema: Option<Value>,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct GuestCapability {
    capability_id: String,
    descriptor_version: String,
    request_operations: Vec<String>,
    #[serde(default)]
    stream_operations: Vec<String>,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct GuestRequirement {
    #[serde(default)]
    requirement_id: Option<String>,
    capability_id: String,
    descriptor_version: String,
    cardinality: String,
}

/// Verified closure of one immutable Plugin Release.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct VerifiedBundle {
    pub plugin_id: String,
    pub release_version: String,
    pub manifest_digest: String,
    pub artifact_digests: Vec<String>,
    pub product_metadata_digests: Vec<String>,
}

/// A Plugin authoring or immutable Bundle invariant failed closed.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum BundleError {
    InvalidManifest(String),
    InvalidBundle(String),
    DigestMismatch(String),
    Io(String),
    Wasm(String),
}

impl fmt::Display for BundleError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidManifest(detail) => write!(formatter, "invalid Plugin Manifest: {detail}"),
            Self::InvalidBundle(detail) => write!(formatter, "invalid Plugin Bundle: {detail}"),
            Self::DigestMismatch(subject) => write!(formatter, "digest mismatch for {subject}"),
            Self::Io(detail) => formatter.write_str(detail),
            Self::Wasm(detail) => write!(
                formatter,
                "failed to encode WebAssembly Component: {detail}"
            ),
        }
    }
}

impl std::error::Error for BundleError {}

/// Builds a one-entry V2 Plugin Bundle entirely from package and source evidence.
pub fn build_source_plugin_bundle(
    build: &SourcePluginBuild,
) -> Result<VerifiedBundle, BundleError> {
    if build.output.exists() {
        return invalid_bundle(format!(
            "output `{}` already exists",
            build.output.display()
        ));
    }
    let package_bytes = read_regular_file(&build.package_manifest, "Cargo manifest")?;
    let package = toml::from_slice::<CargoManifest>(&package_bytes)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    let module = read_regular_file(&build.wasm_module, "Plugin Wasm module")?;
    let component = wit_component::ComponentEncoder::default()
        .module(&module)
        .map_err(|error| BundleError::Wasm(error.to_string()))?
        .validate(true)
        .encode()
        .map_err(|error| BundleError::Wasm(error.to_string()))?;
    let runtime_descriptor = extract_plugin_descriptor(&component)?;
    let artifact = PluginArtifactV2 {
        path: "plugin.wasm".to_owned(),
        digest: sha256_digest(&component),
        size: u64::try_from(component.len())
            .map_err(|_| BundleError::InvalidBundle("Artifact size exceeds u64".to_owned()))?,
        media_type: "application/wasm".to_owned(),
        target: "wasm32-unknown-unknown".to_owned(),
    };
    let descriptor = portable_plugin_descriptor(
        &package.package.metadata.lenso.plugin_id,
        &package.package.version,
        &package.package.metadata.lenso.root_slot,
        &artifact.digest,
        &runtime_descriptor,
        PortableRuntime {
            execution_class: "lenso.wasm-component@1",
            authoring_version: 1,
            runtime_profile: "lenso.wasm-component@1",
        },
    )?;
    let document = SourceManifestDocument::from_value(PluginManifestV2 {
        schema_version: 2,
        plugin_id: package.package.metadata.lenso.plugin_id,
        release_version: package.package.version,
        artifact,
        entry: PluginEntryV2 { descriptor },
    })?;

    let output_parent = build.output.parent().unwrap_or_else(|| Path::new("."));
    fs::create_dir_all(output_parent).map_err(io_error)?;
    let staging = tempfile::Builder::new()
        .prefix(".lenso-plugin-")
        .tempdir_in(output_parent)
        .map_err(io_error)?;
    write_bundle_file(staging.path(), &document.value.artifact.path, &component)?;
    fs::write(staging.path().join(MANIFEST_FILE), &document.bytes).map_err(io_error)?;
    fs::rename(staging.path(), &build.output).map_err(io_error)?;
    verify_bundle_directory(&build.output)
}

/// Builds a one-entry V2 Process Plugin Bundle from generated source evidence.
pub fn build_source_process_plugin_bundle(
    build: &SourceProcessPluginBuild,
) -> Result<VerifiedBundle, BundleError> {
    if build.output.exists() {
        return invalid_bundle(format!(
            "output `{}` already exists",
            build.output.display()
        ));
    }
    if build.target.trim().is_empty() {
        return invalid_manifest("Process target is empty");
    }
    let package_bytes = read_regular_file(&build.package_manifest, "Cargo manifest")?;
    let package = toml::from_slice::<CargoManifest>(&package_bytes)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    let executable = read_regular_file(&build.executable, "Process executable")?;
    let encoded_descriptor = read_regular_file(&build.runtime_descriptor, "runtime descriptor")?;
    let artifact = PluginArtifactV2 {
        path: if cfg!(windows) {
            "plugin.exe".to_owned()
        } else {
            "plugin".to_owned()
        },
        digest: sha256_digest(&executable),
        size: u64::try_from(executable.len())
            .map_err(|_| BundleError::InvalidBundle("Artifact size exceeds u64".to_owned()))?,
        media_type: "application/vnd.lenso.process".to_owned(),
        target: build.target.clone(),
    };
    let descriptor = portable_plugin_descriptor(
        &package.package.metadata.lenso.plugin_id,
        &package.package.version,
        &package.package.metadata.lenso.root_slot,
        &artifact.digest,
        &encoded_descriptor,
        PortableRuntime {
            execution_class: "lenso.process@1",
            authoring_version: build.authoring_version,
            runtime_profile: &build.runtime_profile,
        },
    )?;
    let document = SourceManifestDocument::from_value(PluginManifestV2 {
        schema_version: 2,
        plugin_id: package.package.metadata.lenso.plugin_id,
        release_version: package.package.version,
        artifact,
        entry: PluginEntryV2 { descriptor },
    })?;

    let output_parent = build.output.parent().unwrap_or_else(|| Path::new("."));
    fs::create_dir_all(output_parent).map_err(io_error)?;
    let staging = tempfile::Builder::new()
        .prefix(".lenso-plugin-")
        .tempdir_in(output_parent)
        .map_err(io_error)?;
    write_bundle_file(staging.path(), &document.value.artifact.path, &executable)?;
    preserve_executable_permissions(
        &build.executable,
        &staging.path().join(&document.value.artifact.path),
    )?;
    fs::write(staging.path().join(MANIFEST_FILE), &document.bytes).map_err(io_error)?;
    fs::rename(staging.path(), &build.output).map_err(io_error)?;
    verify_bundle_directory(&build.output)
}

/// Materializes a V4 Plugin Bundle from one contract and built implementation Artifacts.
pub fn build_source_plugin_release_bundle(
    build: &SourcePluginReleaseBuild,
) -> Result<VerifiedBundle, BundleError> {
    if build.output.exists() {
        return invalid_bundle(format!(
            "output `{}` already exists",
            build.output.display()
        ));
    }
    let mut files = Vec::with_capacity(build.implementations.len());
    let mut implementations = Vec::with_capacity(build.implementations.len());
    for source in &build.implementations {
        let bytes = read_regular_file(&source.artifact, "Plugin implementation Artifact")?;
        let digest = sha256_digest(&bytes);
        let artifact = PluginArtifactV2 {
            path: source.bundle_path.clone(),
            digest: digest.clone(),
            size: u64::try_from(bytes.len())
                .map_err(|_| BundleError::InvalidBundle("Artifact size exceeds u64".to_owned()))?,
            media_type: source.media_type.clone(),
            target: source.target.clone(),
        };
        implementations.push(PluginImplementationV4 {
            id: source.id.clone(),
            host_targets: source.host_targets.clone(),
            artifact,
            runtime: PluginImplementation::new(
                build.contract.plugin_id(),
                digest,
                &source.entrypoint,
                source.execution_class.clone(),
            )
            .with_runtime_profile(&source.runtime_profile),
        });
        files.push((source, bytes));
    }
    implementations.sort_by(|left, right| left.id.cmp(&right.id));
    let manifest = PluginManifestV4 {
        schema_version: 4,
        contract: build.contract.clone(),
        implementations,
    };
    validate_v4_manifest(&manifest)?;
    let bytes = serde_json::to_vec(&manifest)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;

    let output_parent = build.output.parent().unwrap_or_else(|| Path::new("."));
    fs::create_dir_all(output_parent).map_err(io_error)?;
    let staging = tempfile::Builder::new()
        .prefix(".lenso-plugin-")
        .tempdir_in(output_parent)
        .map_err(io_error)?;
    for (source, artifact) in files {
        write_bundle_file(staging.path(), &source.bundle_path, &artifact)?;
        if source.media_type == "application/vnd.lenso.process" {
            preserve_executable_permissions(
                &source.artifact,
                &staging.path().join(&source.bundle_path),
            )?;
        }
    }
    fs::write(staging.path().join(MANIFEST_FILE), bytes).map_err(io_error)?;
    fs::rename(staging.path(), &build.output).map_err(io_error)?;
    verify_bundle_directory(&build.output)
}

/// Verifies an already materialized directory as an exact immutable Bundle closure.
pub fn verify_bundle_directory(root: &Path) -> Result<VerifiedBundle, BundleError> {
    verify_bundle_directory_with_limits(root, &BundleVerificationLimits::default())
}

/// Verifies one Bundle with explicit Host-owned resource bounds.
pub fn verify_bundle_directory_with_limits(
    root: &Path,
    limits: &BundleVerificationLimits,
) -> Result<VerifiedBundle, BundleError> {
    verify_bundle_document_with_limits(root, limits).map(|(verified, _)| verified)
}

fn verify_bundle_document_with_limits(
    root: &Path,
    limits: &BundleVerificationLimits,
) -> Result<(VerifiedBundle, ManifestDocument), BundleError> {
    verify_bundle_document_with_limits_after_manifest_read(root, limits, || {})
}

fn verify_bundle_document_with_limits_after_manifest_read(
    root: &Path,
    limits: &BundleVerificationLimits,
    after_manifest_read: impl FnOnce(),
) -> Result<(VerifiedBundle, ManifestDocument), BundleError> {
    validate_verification_limits(limits)?;
    let manifest_path = root.join(MANIFEST_FILE);
    let manifest_bytes =
        read_regular_file_bounded(&manifest_path, "Plugin Manifest", limits.max_manifest_bytes)?;
    after_manifest_read();
    let mut files = BTreeMap::new();
    let mut total_size = 0_u64;
    let mut entry_count = 0_usize;
    collect_bundle_files(
        root,
        root,
        0,
        limits,
        &mut entry_count,
        &mut total_size,
        &mut files,
    )?;
    let manifest_summary = files
        .remove(MANIFEST_FILE)
        .ok_or_else(|| BundleError::InvalidBundle("Bundle is missing its Manifest".to_owned()))?;
    if manifest_summary.size != u64::try_from(manifest_bytes.len()).unwrap_or(u64::MAX)
        || manifest_summary.digest != sha256_digest(&manifest_bytes)
    {
        return invalid_bundle("Plugin Manifest changed during Bundle verification");
    }
    let manifest = ManifestDocument::parse(&manifest_bytes)?;
    let verified = verify_manifest_bundle_files(root, &manifest, &files, limits)?;
    Ok((verified, manifest))
}

/// Strictly reads either supported Plugin Manifest version from a verified Bundle.
pub fn read_bundle_manifest(root: &Path) -> Result<PluginManifest, BundleError> {
    let (_, manifest) =
        verify_bundle_document_with_limits(root, &BundleVerificationLimits::default())?;
    Ok(manifest.value)
}

fn verify_manifest_bundle_files(
    root: &Path,
    manifest: &ManifestDocument,
    files: &BTreeMap<String, BundleFileSummary>,
    limits: &BundleVerificationLimits,
) -> Result<VerifiedBundle, BundleError> {
    match &manifest.value {
        PluginManifest::V2(value) => verify_source_bundle_files(
            &SourceManifestDocument {
                value: value.clone(),
                bytes: Vec::new(),
                digest: manifest.digest.clone(),
            },
            root,
            files,
            limits,
        ),
        PluginManifest::V3(value) => {
            verify_v3_bundle_files(root, value, &manifest.digest, files, limits)
        }
        PluginManifest::V4(value) => {
            verify_v4_bundle_files(root, value, &manifest.digest, files, limits)
        }
    }
}

fn verify_v3_bundle_files(
    root: &Path,
    manifest: &PluginManifestV3,
    manifest_digest: &str,
    files: &BTreeMap<String, BundleFileSummary>,
    limits: &BundleVerificationLimits,
) -> Result<VerifiedBundle, BundleError> {
    verify_profiled_bundle_files(
        root,
        &manifest.contract,
        manifest
            .implementations
            .iter()
            .map(|implementation| (&implementation.artifact, &implementation.runtime)),
        manifest.implementations.len(),
        manifest_digest,
        files,
        limits,
        "V3",
    )
}

#[allow(clippy::too_many_arguments)]
fn verify_profiled_bundle_files<'a>(
    root: &Path,
    contract: &PluginContract,
    implementations: impl Iterator<Item = (&'a PluginArtifactV2, &'a PluginImplementation)>,
    implementation_count: usize,
    manifest_digest: &str,
    files: &BTreeMap<String, BundleFileSummary>,
    limits: &BundleVerificationLimits,
    schema: &str,
) -> Result<VerifiedBundle, BundleError> {
    if files.len() != implementation_count {
        return invalid_bundle(format!(
            "{schema} Bundle closure does not equal its implementation Artifacts"
        ));
    }
    let mut artifact_digests = Vec::with_capacity(implementation_count);
    for (artifact, runtime) in implementations {
        let Some(summary) = files.get(&artifact.path) else {
            return invalid_bundle(format!("{schema} Bundle is missing `{}`", artifact.path));
        };
        if artifact.size != summary.size || artifact.digest != summary.digest {
            return Err(BundleError::DigestMismatch(artifact.path.clone()));
        }
        if runtime.runtime_package_revision() != artifact.digest {
            return invalid_manifest("implementation revision must equal its Artifact digest");
        }
        if artifact.media_type == "application/wasm" {
            let bytes = read_verified_bundle_artifact(root, artifact, limits)?;
            let encoded = extract_plugin_descriptor(&bytes)?;
            let derived = portable_plugin_descriptor(
                contract.plugin_id(),
                contract.release_version(),
                contract.root_slot(),
                &artifact.digest,
                &encoded,
                PortableRuntime {
                    execution_class: runtime.execution_class().as_str(),
                    authoring_version: contract.authoring_version(),
                    runtime_profile: runtime.runtime_profile(),
                },
            )?;
            let derived = serde_json::from_value::<PluginDescriptor>(derived)
                .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
            if derived.contract() != *contract || derived.implementation() != *runtime {
                return invalid_bundle(format!(
                    "Wasm source descriptor does not match its {schema} Contract and implementation"
                ));
            }
        }
        artifact_digests.push(artifact.digest.clone());
    }
    Ok(VerifiedBundle {
        plugin_id: contract.plugin_id().to_owned(),
        release_version: contract.release_version().to_owned(),
        manifest_digest: manifest_digest.to_owned(),
        artifact_digests,
        product_metadata_digests: Vec::new(),
    })
}

fn verify_v4_bundle_files(
    root: &Path,
    manifest: &PluginManifestV4,
    manifest_digest: &str,
    files: &BTreeMap<String, BundleFileSummary>,
    limits: &BundleVerificationLimits,
) -> Result<VerifiedBundle, BundleError> {
    verify_profiled_bundle_files(
        root,
        &manifest.contract,
        manifest
            .implementations
            .iter()
            .map(|implementation| (&implementation.artifact, &implementation.runtime)),
        manifest.implementations.len(),
        manifest_digest,
        files,
        limits,
        "V4",
    )
}

fn verify_source_bundle_files(
    manifest: &SourceManifestDocument,
    root: &Path,
    files: &BTreeMap<String, BundleFileSummary>,
    limits: &BundleVerificationLimits,
) -> Result<VerifiedBundle, BundleError> {
    let artifact = &manifest.value.artifact;
    if files.len() != 1 {
        return invalid_bundle("V2 Bundle must contain exactly one Artifact");
    }
    let Some(summary) = files.get(&artifact.path) else {
        return invalid_bundle("V2 Bundle does not contain its declared Artifact");
    };
    if artifact.size != summary.size || artifact.digest != summary.digest {
        return Err(BundleError::DigestMismatch(artifact.path.clone()));
    }
    if artifact.media_type == "application/wasm" {
        let bytes = read_verified_bundle_artifact(root, artifact, limits)?;
        let runtime_descriptor = extract_plugin_descriptor(&bytes)?;
        let descriptor = portable_plugin_descriptor(
            &manifest.value.plugin_id,
            &manifest.value.release_version,
            manifest
                .value
                .entry
                .descriptor
                .get("root_slot")
                .and_then(Value::as_str)
                .ok_or_else(|| BundleError::InvalidManifest("root_slot is required".to_owned()))?,
            &artifact.digest,
            &runtime_descriptor,
            PortableRuntime {
                execution_class: "lenso.wasm-component@1",
                authoring_version: manifest
                    .value
                    .entry
                    .descriptor
                    .get("authoring_version")
                    .and_then(Value::as_u64)
                    .and_then(|value| u32::try_from(value).ok())
                    .unwrap_or(1),
                runtime_profile: manifest
                    .value
                    .entry
                    .descriptor
                    .get("runtime_profile")
                    .and_then(Value::as_str)
                    .unwrap_or("lenso.wasm-component@1"),
            },
        )?;
        let packaged = serde_json::to_vec(&manifest.value.entry.descriptor)
            .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
        let derived = serde_json::to_vec(&descriptor)
            .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
        if derived != packaged {
            return invalid_bundle("source descriptor does not match the V2 Plugin entry");
        }
    } else {
        validate_process_descriptor(manifest)?;
    }
    Ok(VerifiedBundle {
        plugin_id: manifest.value.plugin_id.clone(),
        release_version: manifest.value.release_version.clone(),
        manifest_digest: manifest.digest.clone(),
        artifact_digests: vec![artifact.digest.clone()],
        product_metadata_digests: Vec::new(),
    })
}

#[derive(Clone, Copy)]
struct PortableRuntime<'a> {
    execution_class: &'a str,
    authoring_version: u32,
    runtime_profile: &'a str,
}

fn portable_plugin_descriptor(
    plugin_id: &str,
    release_version: &str,
    root_slot: &str,
    artifact_digest: &str,
    encoded: &[u8],
    authoring: PortableRuntime<'_>,
) -> Result<Value, BundleError> {
    let runtime = strict_json::<GuestRuntimeDescriptor>(encoded)?;
    if ![
        "lenso.json-request@1",
        "lenso.json-interactions@1",
        "lenso.json-host-imports@1",
        "lenso.json-host-imports@2",
    ]
    .contains(&runtime.abi.as_str())
    {
        return invalid_manifest("unsupported guest Plugin ABI");
    }
    let mut descriptor = PluginDescriptor::new(plugin_id, release_version, root_slot)
        .with_authoring(authoring.authoring_version, authoring.runtime_profile)
        .with_runtime_package(plugin_id, artifact_digest)
        .with_entrypoint("plugin")
        .with_execution_class(ExecutionClassId::new(authoring.execution_class));
    if let Some(configuration_schema) = runtime.configuration_schema {
        descriptor = descriptor.with_configuration_schema(configuration_schema);
    }
    for capability in runtime.capabilities {
        let mut endpoint = CapabilityEndpointPlan::new(
            capability.capability_id,
            capability.descriptor_version,
            capability
                .request_operations
                .iter()
                .chain(&capability.stream_operations)
                .cloned(),
        );
        for operation in capability.stream_operations {
            endpoint = endpoint.with_operation_kind(operation, CapabilityOperationKind::Stream);
        }
        descriptor = descriptor.with_capability(endpoint);
    }
    for requirement in runtime.required_capabilities {
        if requirement.cardinality != "one" {
            return invalid_manifest("unsupported guest Capability cardinality");
        }
        let requirement_id = match requirement.requirement_id {
            Some(requirement_id) if !requirement_id.trim().is_empty() => requirement_id,
            Some(_) => return invalid_manifest("guest requirement identity must not be empty"),
            None if runtime.abi == "lenso.json-host-imports@1" => requirement.capability_id.clone(),
            None => return invalid_manifest("guest requirement identity is missing"),
        };
        descriptor = descriptor.with_requirement(
            CapabilityRequirementPlan::one(
                requirement.capability_id,
                requirement.descriptor_version,
            )
            .with_requirement_id(requirement_id),
        );
    }
    serde_json::to_value(descriptor)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))
}

fn validate_process_descriptor(manifest: &SourceManifestDocument) -> Result<(), BundleError> {
    if manifest.value.artifact.media_type != "application/vnd.lenso.process" {
        return invalid_manifest("non-Wasm V2 Artifact must be a Process executable");
    }
    let descriptor =
        serde_json::from_value::<PluginDescriptor>(manifest.value.entry.descriptor.clone())
            .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    if descriptor.plugin_id() != manifest.value.plugin_id
        || descriptor.release_version() != manifest.value.release_version
        || descriptor.root_slot().is_empty()
        || descriptor.runtime_package_id() != manifest.value.plugin_id
        || descriptor.runtime_package_revision() != manifest.value.artifact.digest
        || descriptor.entrypoint() != "plugin"
        || descriptor.execution_class().as_str() != "lenso.process@1"
        || descriptor.provided_capabilities().is_empty()
    {
        return invalid_manifest("Process descriptor does not close exact Bundle authority");
    }
    Ok(())
}

/// Extracts one canonical source-derived Plugin descriptor without executing it.
pub fn extract_plugin_descriptor(component: &[u8]) -> Result<Vec<u8>, BundleError> {
    let mut descriptors = Vec::new();
    collect_plugin_descriptors(component, &mut descriptors)?;
    let [descriptor] = descriptors.as_slice() else {
        return invalid_bundle(if descriptors.is_empty() {
            "Plugin Component does not contain a source-derived descriptor"
        } else {
            "Plugin Component contains duplicate source-derived descriptors"
        });
    };
    let value = strict_json::<Value>(descriptor)?;
    let canonical = serde_json::to_vec(&value)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    if canonical != *descriptor {
        return invalid_bundle("Plugin descriptor is not canonical JSON");
    }
    Ok(descriptor.clone())
}

fn collect_plugin_descriptors(
    bytes: &[u8],
    descriptors: &mut Vec<Vec<u8>>,
) -> Result<(), BundleError> {
    for payload in wasmparser::Parser::new(0).parse_all(bytes) {
        match payload.map_err(|error| BundleError::Wasm(error.to_string()))? {
            wasmparser::Payload::CustomSection(section)
                if section.name() == PLUGIN_DESCRIPTOR_SECTION =>
            {
                if section.data().len() > MAX_PLUGIN_DESCRIPTOR_BYTES {
                    return invalid_bundle("Plugin descriptor exceeds the size limit");
                }
                descriptors.push(section.data().to_vec());
            }
            _ => {}
        }
    }
    Ok(())
}

fn validate_source_manifest(manifest: &PluginManifestV2) -> Result<(), BundleError> {
    if manifest.schema_version != 2 {
        return invalid_manifest("unsupported schema version");
    }
    if manifest.plugin_id.is_empty() || semver::Version::parse(&manifest.release_version).is_err() {
        return invalid_manifest("Plugin identity or Release version is invalid");
    }
    validate_relative_path(&manifest.artifact.path)?;
    digest_component(&manifest.artifact.digest)?;
    if manifest.artifact.size == 0 {
        return invalid_manifest("V2 Artifact size must be non-zero");
    }
    match manifest.artifact.media_type.as_str() {
        "application/wasm" if manifest.artifact.target == "wasm32-unknown-unknown" => {}
        "application/vnd.lenso.process" if !manifest.artifact.target.trim().is_empty() => {}
        _ => return invalid_manifest("V2 Artifact media type and target are not supported"),
    }
    if !manifest.entry.descriptor.is_object() {
        return invalid_manifest("V2 Plugin entry descriptor must be an object");
    }
    Ok(())
}

fn validate_manifest(manifest: &PluginManifest) -> Result<(), BundleError> {
    match manifest {
        PluginManifest::V2(value) => validate_source_manifest(value),
        PluginManifest::V3(value) => validate_v3_manifest(value),
        PluginManifest::V4(value) => validate_v4_manifest(value),
    }
}

fn validate_v4_manifest(manifest: &PluginManifestV4) -> Result<(), BundleError> {
    if manifest.schema_version != 4 || manifest.contract.authoring_version() != 2 {
        return invalid_manifest("V4 requires authoring_version 2");
    }
    validate_profiled_manifest(
        &manifest.contract,
        manifest.implementations.iter().map(|implementation| {
            (
                &implementation.id,
                &implementation.host_targets,
                &implementation.artifact,
                &implementation.runtime,
            )
        }),
        "V4",
    )
}

fn validate_profiled_manifest<'a>(
    contract: &PluginContract,
    implementations: impl Iterator<
        Item = (
            &'a String,
            &'a Vec<String>,
            &'a PluginArtifactV2,
            &'a PluginImplementation,
        ),
    >,
    schema: &str,
) -> Result<(), BundleError> {
    if contract.plugin_id().is_empty()
        || semver::Version::parse(contract.release_version()).is_err()
        || contract.root_slot().is_empty()
    {
        return invalid_manifest(format!("{schema} Contract is invalid"));
    }
    let mut ids = BTreeSet::new();
    let mut paths = BTreeSet::new();
    let mut count = 0_usize;
    for (id, host_targets, artifact, runtime) in implementations {
        count += 1;
        if id.trim().is_empty() || !ids.insert(id) {
            return invalid_manifest(format!(
                "{schema} implementation ids must be non-empty and unique"
            ));
        }
        if host_targets.is_empty() || host_targets.iter().any(|target| target.trim().is_empty()) {
            return invalid_manifest(format!(
                "{schema} implementation host targets must be non-empty"
            ));
        }
        validate_artifact(artifact)?;
        if !paths.insert(&artifact.path) {
            return invalid_manifest(format!(
                "{schema} implementation Artifact paths must be unique"
            ));
        }
        if runtime.runtime_package_id() != contract.plugin_id()
            || runtime.runtime_package_revision() != artifact.digest
            || runtime.entrypoint().is_empty()
            || runtime.runtime_profile().trim().is_empty()
        {
            return invalid_manifest(format!(
                "{schema} implementation does not close Plugin authority"
            ));
        }
    }
    if count == 0 {
        return invalid_manifest(format!("{schema} implementation set is empty"));
    }
    Ok(())
}

fn validate_v3_manifest(manifest: &PluginManifestV3) -> Result<(), BundleError> {
    if manifest.schema_version != 3 {
        return invalid_manifest("unsupported schema version");
    }
    if manifest.contract.plugin_id().is_empty()
        || semver::Version::parse(manifest.contract.release_version()).is_err()
        || manifest.contract.root_slot().is_empty()
        || manifest.implementations.is_empty()
    {
        return invalid_manifest("V3 Contract or implementation set is invalid");
    }
    let mut ids = BTreeSet::new();
    let mut paths = BTreeSet::new();
    for implementation in &manifest.implementations {
        if implementation.id.trim().is_empty() || !ids.insert(&implementation.id) {
            return invalid_manifest("V3 implementation ids must be non-empty and unique");
        }
        if implementation.host_targets.is_empty()
            || implementation
                .host_targets
                .iter()
                .any(|target| target.trim().is_empty())
        {
            return invalid_manifest("V3 implementation host targets must be non-empty");
        }
        validate_artifact(&implementation.artifact)?;
        if !paths.insert(&implementation.artifact.path) {
            return invalid_manifest("V3 implementation Artifact paths must be unique");
        }
        if implementation.runtime.runtime_package_id() != manifest.contract.plugin_id()
            || implementation.runtime.runtime_package_revision() != implementation.artifact.digest
            || implementation.runtime.entrypoint().is_empty()
        {
            return invalid_manifest("V3 implementation does not close Plugin authority");
        }
    }
    Ok(())
}

fn validate_artifact(artifact: &PluginArtifactV2) -> Result<(), BundleError> {
    validate_relative_path(&artifact.path)?;
    digest_component(&artifact.digest)?;
    if artifact.size == 0 {
        return invalid_manifest("Artifact size must be non-zero");
    }
    match artifact.media_type.as_str() {
        "application/wasm" if artifact.target == "wasm32-unknown-unknown" => Ok(()),
        "application/vnd.lenso.process" | "application/javascript"
            if !artifact.target.trim().is_empty() =>
        {
            Ok(())
        }
        _ => invalid_manifest("Artifact media type and target are not supported"),
    }
}

/// Validates publisher-owned Manifest semantics independently of Host policy.
#[allow(clippy::too_many_lines)]
/// Computes the canonical digest syntax used by Plugin Release documents and files.
pub fn sha256_digest(bytes: &[u8]) -> String {
    format!("sha256:{}", hex::encode(Sha256::digest(bytes)))
}

fn strict_json<T: DeserializeOwned>(input: &[u8]) -> Result<T, BundleError> {
    let mut deserializer = serde_json::Deserializer::from_slice(input);
    let strict = StrictValue::deserialize(&mut deserializer)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    deserializer
        .end()
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))?;
    validate_json_value(&strict.0)?;
    serde_json::from_value(strict.0)
        .map_err(|error| BundleError::InvalidManifest(error.to_string()))
}

#[derive(Clone, Debug)]
struct StrictValue(Value);

impl<'de> Deserialize<'de> for StrictValue {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        deserializer.deserialize_any(StrictVisitor)
    }
}

struct StrictVisitor;

impl<'de> serde::de::Visitor<'de> for StrictVisitor {
    type Value = StrictValue;

    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("strict Plugin Manifest JSON")
    }

    fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::Bool(value)))
    }

    fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::Number(value.into())))
    }

    fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        u64::try_from(value)
            .map_err(|_| E::custom("negative integers are forbidden"))
            .and_then(|value| self.visit_u64(value))
    }

    fn visit_f64<E>(self, _: f64) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        Err(E::custom("floating-point values are forbidden"))
    }

    fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::String(value.to_owned())))
    }

    fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::String(value)))
    }

    fn visit_none<E>(self) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::Null))
    }

    fn visit_unit<E>(self) -> Result<Self::Value, E> {
        Ok(StrictValue(Value::Null))
    }

    fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
    where
        A: serde::de::SeqAccess<'de>,
    {
        let mut values = Vec::new();
        while let Some(value) = sequence.next_element::<StrictValue>()? {
            values.push(value.0);
        }
        Ok(StrictValue(Value::Array(values)))
    }

    fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
    where
        A: serde::de::MapAccess<'de>,
    {
        let mut keys = BTreeSet::new();
        let mut values = serde_json::Map::new();
        while let Some(key) = map.next_key::<String>()? {
            if !keys.insert(key.clone()) {
                return Err(serde::de::Error::custom(format!("duplicate field `{key}`")));
            }
            values.insert(key, map.next_value::<StrictValue>()?.0);
        }
        Ok(StrictValue(Value::Object(values)))
    }
}

fn validate_json_value(value: &Value) -> Result<(), BundleError> {
    match value {
        Value::Number(number) if !number.is_u64() => {
            invalid_manifest("numbers must be non-negative integers")
        }
        Value::Array(values) => values.iter().try_for_each(validate_json_value),
        Value::Object(values) => values.values().try_for_each(validate_json_value),
        _ => Ok(()),
    }
}

fn validate_relative_path(path: &str) -> Result<(), BundleError> {
    if path.is_empty() || path.contains('\\') {
        return invalid_manifest("Bundle path is empty or platform-ambiguous");
    }
    let path = Path::new(path);
    if path.is_absolute()
        || path
            .components()
            .any(|part| !matches!(part, Component::Normal(_)))
    {
        return invalid_manifest("Bundle path must contain only normalized relative segments");
    }
    Ok(())
}

fn digest_component(digest: &str) -> Result<&str, BundleError> {
    let Some(value) = digest.strip_prefix("sha256:") else {
        return invalid_manifest("digest does not use sha256 prefix");
    };
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return invalid_manifest("digest is not 64 lowercase hexadecimal characters");
    }
    Ok(value)
}

fn read_regular_file(path: &Path, kind: &str) -> Result<Vec<u8>, BundleError> {
    let metadata = fs::symlink_metadata(path)
        .map_err(|error| BundleError::Io(format!("failed to inspect {kind}: {error}")))?;
    if !metadata.is_file() || metadata.file_type().is_symlink() {
        return invalid_bundle(format!("{kind} is not a regular file"));
    }
    fs::read(path).map_err(io_error)
}

fn validate_verification_limits(limits: &BundleVerificationLimits) -> Result<(), BundleError> {
    if limits.max_manifest_bytes == 0
        || limits.max_file_bytes == 0
        || limits.max_total_bytes == 0
        || limits.max_file_count == 0
        || limits.max_entry_count == 0
        || limits.max_directory_depth == 0
        || limits.max_file_count > limits.max_entry_count
        || limits.max_manifest_bytes > limits.max_file_bytes
        || limits.max_file_bytes > limits.max_total_bytes
    {
        return invalid_bundle("Bundle verification limits are invalid");
    }
    Ok(())
}

fn read_regular_file_bounded(path: &Path, kind: &str, limit: u64) -> Result<Vec<u8>, BundleError> {
    read_regular_file_bounded_after_inspection(path, kind, limit, || {})
}

fn read_regular_file_bounded_after_inspection(
    path: &Path,
    kind: &str,
    limit: u64,
    after_inspection: impl FnOnce(),
) -> Result<Vec<u8>, BundleError> {
    let metadata = fs::symlink_metadata(path)
        .map_err(|error| BundleError::Io(format!("failed to inspect {kind}: {error}")))?;
    if !metadata.is_file() || metadata.file_type().is_symlink() {
        return invalid_bundle(format!("{kind} is not a regular file"));
    }
    if metadata.len() > limit {
        return invalid_bundle(format!("{kind} exceeds the configured size limit"));
    }
    after_inspection();
    let file = fs::File::open(path).map_err(io_error)?;
    let opened = file.metadata().map_err(io_error)?;
    if !opened.is_file() || !same_file_identity(&metadata, &opened) {
        return invalid_bundle(format!("{kind} changed during bounded read"));
    }
    let mut bytes = Vec::with_capacity(usize::try_from(metadata.len()).unwrap_or(0));
    file.take(limit.saturating_add(1))
        .read_to_end(&mut bytes)
        .map_err(io_error)?;
    if u64::try_from(bytes.len()).unwrap_or(u64::MAX) > limit {
        return invalid_bundle(format!("{kind} exceeds the configured size limit"));
    }
    Ok(bytes)
}

fn read_verified_bundle_artifact(
    root: &Path,
    artifact: &PluginArtifactV2,
    limits: &BundleVerificationLimits,
) -> Result<Vec<u8>, BundleError> {
    let bytes = read_regular_file_bounded(
        &root.join(&artifact.path),
        "Plugin Artifact",
        limits.max_file_bytes,
    )?;
    if u64::try_from(bytes.len()).unwrap_or(u64::MAX) != artifact.size
        || sha256_digest(&bytes) != artifact.digest
    {
        return Err(BundleError::DigestMismatch(artifact.path.clone()));
    }
    Ok(bytes)
}

fn write_bundle_file(root: &Path, relative: &str, bytes: &[u8]) -> Result<(), BundleError> {
    let path = root.join(relative);
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(io_error)?;
    }
    fs::write(path, bytes).map_err(io_error)
}

#[cfg(unix)]
fn preserve_executable_permissions(source: &Path, destination: &Path) -> Result<(), BundleError> {
    use std::os::unix::fs::PermissionsExt as _;

    let source_permissions = fs::metadata(source).map_err(io_error)?.permissions();
    let mode = source_permissions.mode();
    if mode & 0o111 == 0 {
        return invalid_bundle("Process executable has no executable permission bit");
    }
    fs::set_permissions(destination, fs::Permissions::from_mode(mode)).map_err(io_error)
}

#[cfg(not(unix))]
fn preserve_executable_permissions(_: &Path, _: &Path) -> Result<(), BundleError> {
    Ok(())
}

fn collect_bundle_files(
    root: &Path,
    directory: &Path,
    depth: usize,
    limits: &BundleVerificationLimits,
    entry_count: &mut usize,
    total_size: &mut u64,
    files: &mut BTreeMap<String, BundleFileSummary>,
) -> Result<(), BundleError> {
    if depth > limits.max_directory_depth {
        return invalid_bundle("Bundle directory depth exceeds the configured limit");
    }
    let metadata = fs::symlink_metadata(directory).map_err(io_error)?;
    if !metadata.is_dir() || metadata.file_type().is_symlink() {
        return invalid_bundle("Bundle root contains a non-regular directory");
    }
    for entry in fs::read_dir(directory).map_err(io_error)? {
        let entry = entry.map_err(io_error)?;
        *entry_count = entry_count
            .checked_add(1)
            .ok_or_else(|| BundleError::InvalidBundle("Bundle entry count overflow".to_owned()))?;
        if *entry_count > limits.max_entry_count {
            return invalid_bundle("Bundle entry count exceeds the configured limit");
        }
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(io_error)?;
        if metadata.file_type().is_symlink() {
            return invalid_bundle("Bundle contains a symbolic link");
        }
        if metadata.is_dir() {
            collect_bundle_files(
                root,
                &path,
                depth + 1,
                limits,
                entry_count,
                total_size,
                files,
            )?;
            continue;
        }
        if !metadata.is_file() {
            return invalid_bundle("Bundle contains a non-regular file");
        }
        let relative = path
            .strip_prefix(root)
            .map_err(|_| BundleError::InvalidBundle("Bundle path escaped root".to_owned()))?
            .to_str()
            .ok_or_else(|| BundleError::InvalidBundle("Bundle path is not UTF-8".to_owned()))?
            .replace(std::path::MAIN_SEPARATOR, "/");
        validate_relative_path(&relative)?;
        if files.len() >= limits.max_file_count {
            return invalid_bundle("Bundle file count exceeds the configured limit");
        }
        let summary = summarize_bundle_file(&path, &metadata, limits.max_file_bytes)?;
        *total_size = total_size
            .checked_add(summary.size)
            .ok_or_else(|| BundleError::InvalidBundle("Bundle total size overflow".to_owned()))?;
        if *total_size > limits.max_total_bytes {
            return invalid_bundle("Bundle total size exceeds the configured limit");
        }
        files.insert(relative, summary);
    }
    Ok(())
}

fn summarize_bundle_file(
    path: &Path,
    metadata: &fs::Metadata,
    max_file_bytes: u64,
) -> Result<BundleFileSummary, BundleError> {
    if metadata.len() > max_file_bytes {
        return invalid_bundle("Bundle file exceeds the configured size limit");
    }
    let mut file = fs::File::open(path).map_err(io_error)?;
    let opened = file.metadata().map_err(io_error)?;
    if !opened.is_file() || !same_file_identity(metadata, &opened) || opened.len() > max_file_bytes
    {
        return invalid_bundle("Bundle file changed during verification");
    }
    let mut hasher = Sha256::new();
    let mut size = 0_u64;
    let mut buffer = vec![0_u8; 64 * 1024];
    loop {
        let read = file.read(&mut buffer).map_err(io_error)?;
        if read == 0 {
            break;
        }
        size = size
            .checked_add(u64::try_from(read).expect("buffer length fits u64"))
            .ok_or_else(|| BundleError::InvalidBundle("Bundle file size overflow".to_owned()))?;
        if size > max_file_bytes {
            return invalid_bundle("Bundle file exceeds the configured size limit");
        }
        hasher.update(&buffer[..read]);
    }
    if size != opened.len() {
        return invalid_bundle("Bundle file changed during verification");
    }
    Ok(BundleFileSummary {
        size,
        digest: format!("sha256:{}", hex::encode(hasher.finalize())),
    })
}

#[cfg(unix)]
fn same_file_identity(inspected: &fs::Metadata, opened: &fs::Metadata) -> bool {
    use std::os::unix::fs::MetadataExt as _;

    inspected.dev() == opened.dev() && inspected.ino() == opened.ino()
}

#[cfg(not(unix))]
fn same_file_identity(_: &fs::Metadata, _: &fs::Metadata) -> bool {
    true
}

fn invalid_manifest<T>(detail: impl Into<String>) -> Result<T, BundleError> {
    Err(BundleError::InvalidManifest(detail.into()))
}

fn invalid_bundle<T>(detail: impl Into<String>) -> Result<T, BundleError> {
    Err(BundleError::InvalidBundle(detail.into()))
}

fn io_error(error: impl fmt::Display) -> BundleError {
    BundleError::Io(error.to_string())
}

#[cfg(test)]
mod tests {
    use std::borrow::Cow;

    use super::*;

    #[cfg(unix)]
    #[test]
    fn bounded_reader_rejects_a_symlink_swap_between_inspection_and_open() {
        use std::os::unix::fs::symlink;

        let directory = tempfile::tempdir().unwrap();
        let selected = directory.path().join("selected");
        let replacement = directory.path().join("replacement");
        fs::write(&selected, b"selected").unwrap();
        fs::write(&replacement, b"selected").unwrap();

        let result = read_regular_file_bounded_after_inspection(&selected, "test file", 64, || {
            fs::remove_file(&selected).unwrap();
            symlink(&replacement, &selected).unwrap();
        });

        assert!(matches!(
            result,
            Err(BundleError::InvalidBundle(detail)) if detail.contains("changed during bounded read")
        ));
    }

    fn wasm_with_descriptors(descriptors: &[&[u8]]) -> Vec<u8> {
        let mut module = wasm_encoder::Module::new();
        for descriptor in descriptors {
            module.section(&wasm_encoder::CustomSection {
                name: Cow::Borrowed(PLUGIN_DESCRIPTOR_SECTION),
                data: Cow::Borrowed(descriptor),
            });
        }
        module.finish()
    }

    #[test]
    fn source_metadata_rejects_old_multi_entry_fields() {
        let error = toml::from_str::<CargoManifest>(
            r#"
                [package]
                version = "1.0.0"

                [package.metadata.lenso]
                plugin-id = "example.echo"
                root-slot = "tools"
                module-contributions = []
            "#,
        )
        .unwrap_err();

        assert!(error.to_string().contains("module-contributions"));
    }

    #[test]
    fn descriptor_extraction_requires_one_canonical_descriptor() {
        assert!(extract_plugin_descriptor(&wasm_with_descriptors(&[])).is_err());
        let descriptor = br#"{"profile":"one"}"#;
        assert!(
            extract_plugin_descriptor(&wasm_with_descriptors(&[
                descriptor.as_slice(),
                descriptor.as_slice(),
            ]))
            .is_err()
        );
        assert!(extract_plugin_descriptor(&wasm_with_descriptors(&[b"{"])).is_err());
        assert!(
            extract_plugin_descriptor(&wasm_with_descriptors(&[br#"{ "profile": "one" }"#]))
                .is_err()
        );
    }

    #[test]
    fn descriptor_extraction_rejects_oversized_evidence() {
        let descriptor = vec![b' '; MAX_PLUGIN_DESCRIPTOR_BYTES + 1];
        assert!(extract_plugin_descriptor(&wasm_with_descriptors(&[&descriptor])).is_err());
    }

    #[test]
    fn host_imports_v2_preserves_named_requirements_during_bundle_lowering() {
        let encoded = br#"{"abi":"lenso.json-host-imports@2","capabilities":[],"required_capabilities":[{"requirement_id":"source","capability_id":"example.store@1","descriptor_version":"1.0.0","cardinality":"one"}],"configuration_schema":{"type":"object","required":["prefix"]}}"#;
        let value = portable_plugin_descriptor(
            "example.copy",
            "1.0.0",
            "tools",
            "sha256:artifact",
            encoded,
            PortableRuntime {
                execution_class: "lenso.wasm-component@1",
                authoring_version: 2,
                runtime_profile: "lenso.wasm-component@2",
            },
        )
        .unwrap();
        let descriptor: PluginDescriptor = serde_json::from_value(value).unwrap();

        assert_eq!(descriptor.required_capabilities().len(), 1);
        assert_eq!(
            descriptor.required_capabilities()[0].requirement_id(),
            "source"
        );
        assert_eq!(
            descriptor.configuration_schema().unwrap()["required"],
            serde_json::json!(["prefix"])
        );
    }

    #[test]
    fn strict_v2_manifest_rejects_duplicate_fields_and_path_escape() {
        assert!(
            SourceManifestDocument::parse(br#"{"schema_version":2,"schema_version":2}"#).is_err()
        );
        let manifest = PluginManifestV2 {
            schema_version: 2,
            plugin_id: "example.echo".to_owned(),
            release_version: "1.0.0".to_owned(),
            artifact: PluginArtifactV2 {
                path: "../plugin.wasm".to_owned(),
                digest: sha256_digest(b"plugin"),
                size: 6,
                media_type: "application/wasm".to_owned(),
                target: "wasm32-unknown-unknown".to_owned(),
            },
            entry: PluginEntryV2 {
                descriptor: serde_json::json!({"plugin_id":"example.echo"}),
            },
        };
        assert!(SourceManifestDocument::from_value(manifest).is_err());
    }

    #[test]
    fn process_bundle_is_built_without_executing_the_artifact() {
        let root = tempfile::tempdir().unwrap();
        let manifest = root.path().join("Cargo.toml");
        fs::write(
            &manifest,
            r#"[package]
name = "example-process"
version = "1.0.0"

[package.metadata.lenso]
plugin-id = "example.process"
root-slot = "tools"
"#,
        )
        .unwrap();
        let descriptor = root.path().join("descriptor.json");
        fs::write(
            &descriptor,
            br#"{"abi":"lenso.json-request@1","capabilities":[{"capability_id":"example.echo@1","descriptor_version":"1.0.0","request_operations":["echo"]}]}"#,
        )
        .unwrap();
        let output = root.path().join("example.process.lenso-plugin");
        let verified = build_source_process_plugin_bundle(&SourceProcessPluginBuild {
            package_manifest: manifest,
            executable: std::env::current_exe().unwrap(),
            runtime_descriptor: descriptor,
            authoring_version: 2,
            runtime_profile: "lenso.process-stdio@2".to_owned(),
            target: "test-host".to_owned(),
            output: output.clone(),
        })
        .unwrap();

        assert_eq!(verified.plugin_id, "example.process");
        assert_eq!(verified, verify_bundle_directory(&output).unwrap());
        let document =
            SourceManifestDocument::parse(&fs::read(output.join(MANIFEST_FILE)).unwrap()).unwrap();
        assert_eq!(
            document.value.artifact.media_type,
            "application/vnd.lenso.process"
        );
        assert_eq!(
            document.value.entry.descriptor["execution_class"],
            "lenso.process@1"
        );
        assert_eq!(document.value.entry.descriptor["authoring_version"], 2);
        assert_eq!(
            document.value.entry.descriptor["runtime_profile"],
            "lenso.process-stdio@2"
        );

        let bounded = BundleVerificationLimits {
            max_manifest_bytes: 16 * 1024,
            max_file_bytes: 16 * 1024,
            max_total_bytes: 32 * 1024,
            ..BundleVerificationLimits::default()
        };
        assert!(matches!(
            verify_bundle_directory_with_limits(&output, &bounded),
            Err(BundleError::InvalidBundle(detail)) if detail.contains("size limit")
        ));

        let file_count_bounded = BundleVerificationLimits {
            max_file_count: 1,
            ..BundleVerificationLimits::default()
        };
        assert!(matches!(
            verify_bundle_directory_with_limits(&output, &file_count_bounded),
            Err(BundleError::InvalidBundle(detail)) if detail.contains("file count")
        ));

        for index in 0..64 {
            fs::create_dir(output.join(format!("empty-directory-{index}"))).unwrap();
        }
        let entry_count_bounded = BundleVerificationLimits {
            max_file_count: 2,
            max_entry_count: 4,
            ..BundleVerificationLimits::default()
        };
        assert!(matches!(
            verify_bundle_directory_with_limits(&output, &entry_count_bounded),
            Err(BundleError::InvalidBundle(detail)) if detail.contains("entry count")
        ));

        let manifest_path = output.join(MANIFEST_FILE);
        let drift = verify_bundle_document_with_limits_after_manifest_read(
            &output,
            &BundleVerificationLimits::default(),
            || fs::write(&manifest_path, br#"{"schema_version":2}"#).unwrap(),
        );
        assert!(matches!(
            drift,
            Err(BundleError::InvalidBundle(detail)) if detail.contains("Manifest changed")
        ));
    }

    #[test]
    fn v3_release_selects_one_implementation_by_host_policy() {
        let root = tempfile::tempdir().unwrap();
        let process = std::env::current_exe().unwrap();
        let script = root.path().join("plugin.js");
        fs::write(
            &script,
            b"export function invoke(request) { return request; }",
        )
        .unwrap();
        let output = root.path().join("example.multi.lenso-plugin");
        let contract = PluginContract::new("example.multi", "1.0.0", "tools")
            .with_authoring_version(2)
            .with_capability(CapabilityEndpointPlan::new(
                "example.echo@1",
                "1.0.0",
                ["echo"],
            ));
        build_source_plugin_release_bundle(&SourcePluginReleaseBuild {
            contract,
            implementations: vec![
                SourcePluginImplementation {
                    id: "bun".to_owned(),
                    host_targets: vec!["test-host".to_owned()],
                    artifact: process,
                    bundle_path: "implementations/bun/plugin".to_owned(),
                    media_type: "application/vnd.lenso.process".to_owned(),
                    target: "test-host".to_owned(),
                    entrypoint: "plugin".to_owned(),
                    execution_class: ExecutionClassId::new("lenso.process@1"),
                    runtime_profile: "lenso.process-authoring@2".to_owned(),
                },
                SourcePluginImplementation {
                    id: "quickjs".to_owned(),
                    host_targets: vec!["*".to_owned()],
                    artifact: script,
                    bundle_path: "implementations/quickjs/plugin.js".to_owned(),
                    media_type: "application/javascript".to_owned(),
                    target: "javascript-es2023".to_owned(),
                    entrypoint: "plugin.js".to_owned(),
                    execution_class: ExecutionClassId::new("lenso.quickjs@1"),
                    runtime_profile: "lenso.quickjs-authoring@2".to_owned(),
                },
            ],
            output: output.clone(),
        })
        .unwrap();

        let manifest = read_bundle_manifest(&output).unwrap();
        let selected = resolve_implementation(
            &manifest,
            &ImplementationPolicy {
                host_target: "test-host".to_owned(),
                runtimes: vec![
                    RuntimeAdmission {
                        execution_class: ExecutionClassId::new("lenso.quickjs@1"),
                        runtime_profile: "lenso.quickjs-authoring@2".to_owned(),
                    },
                    RuntimeAdmission {
                        execution_class: ExecutionClassId::new("lenso.process@1"),
                        runtime_profile: "lenso.process-authoring@2".to_owned(),
                    },
                ],
            },
        )
        .unwrap();
        assert_eq!(selected.implementation_id, "quickjs");
        assert_eq!(
            selected.descriptor.execution_class().as_str(),
            "lenso.quickjs@1"
        );
        assert_eq!(selected.descriptor.authoring_version(), 2);
        assert_eq!(
            selected.descriptor.runtime_profile(),
            "lenso.quickjs-authoring@2"
        );
        assert_eq!(
            selected.descriptor.contract(),
            match manifest {
                PluginManifest::V4(value) => value.contract,
                PluginManifest::V2(_) | PluginManifest::V3(_) => {
                    panic!("expected V4 manifest")
                }
            }
        );

        let manifest_bytes = fs::read(output.join(MANIFEST_FILE)).unwrap();
        let manifest_json: Value = serde_json::from_slice(&manifest_bytes).unwrap();
        assert_eq!(manifest_json["schema_version"], 4);
        assert_eq!(manifest_json["contract"]["authoring_version"], 2);
        assert_eq!(
            manifest_json["implementations"][0]["runtime"]["runtime_profile"],
            "lenso.process-authoring@2"
        );
    }

    #[test]
    fn v3_wire_shape_and_digest_remain_stable_after_core_upgrade() {
        let artifact = PluginArtifactV2 {
            path: "plugin.js".to_owned(),
            digest: sha256_digest(b"plugin"),
            size: 6,
            media_type: "application/javascript".to_owned(),
            target: "javascript-es2023".to_owned(),
        };
        let manifest = PluginManifest::V3(PluginManifestV3 {
            schema_version: 3,
            contract: PluginContract::new("example.v3", "1.0.0", "tools").with_capability(
                CapabilityEndpointPlan::new("example.echo@1", "1.0.0", ["echo"]),
            ),
            implementations: vec![PluginImplementationV3 {
                id: "quickjs".to_owned(),
                host_targets: vec!["*".to_owned()],
                artifact: artifact.clone(),
                runtime: PluginImplementation::new(
                    "example.v3",
                    &artifact.digest,
                    "plugin.js",
                    ExecutionClassId::new("lenso.quickjs@1"),
                ),
            }],
        });
        let old_wire = canonical_manifest_bytes(&manifest).unwrap();
        let parsed = ManifestDocument::parse(&old_wire).unwrap();

        assert_eq!(parsed.digest, sha256_digest(&old_wire));
        assert!(
            !String::from_utf8(old_wire.clone())
                .unwrap()
                .contains("authoring_version")
        );
        assert!(
            !String::from_utf8(old_wire.clone())
                .unwrap()
                .contains("runtime_profile")
        );

        let mut extended: Value = serde_json::from_slice(&old_wire).unwrap();
        extended["contract"]["authoring_version"] = Value::from(1);
        assert!(matches!(
            ManifestDocument::parse(&serde_json::to_vec(&extended).unwrap()),
            Err(BundleError::InvalidManifest(detail)) if detail.contains("V3 contract")
        ));
    }

    #[test]
    fn v4_requires_explicit_versions_and_exact_host_admission() {
        let artifact = PluginArtifactV2 {
            path: "plugin.js".to_owned(),
            digest: sha256_digest(b"plugin"),
            size: 6,
            media_type: "application/javascript".to_owned(),
            target: "javascript-es2023".to_owned(),
        };
        let manifest = PluginManifest::V4(PluginManifestV4 {
            schema_version: 4,
            contract: PluginContract::new("example.v4", "1.0.0", "tools")
                .with_authoring_version(2)
                .with_capability(CapabilityEndpointPlan::new(
                    "example.echo@1",
                    "1.0.0",
                    ["echo"],
                )),
            implementations: vec![PluginImplementationV4 {
                id: "quickjs".to_owned(),
                host_targets: vec!["*".to_owned()],
                artifact,
                runtime: PluginImplementation::new(
                    "example.v4",
                    sha256_digest(b"plugin"),
                    "plugin.js",
                    ExecutionClassId::new("lenso.quickjs@1"),
                )
                .with_runtime_profile("lenso.quickjs-authoring@2"),
            }],
        });
        let wire = canonical_manifest_bytes(&manifest).unwrap();
        ManifestDocument::parse(&wire).unwrap();

        let unsupported = resolve_implementation(
            &manifest,
            &ImplementationPolicy {
                host_target: "test-host".to_owned(),
                runtimes: vec![RuntimeAdmission {
                    execution_class: ExecutionClassId::new("lenso.quickjs@1"),
                    runtime_profile: "lenso.quickjs-authoring@1".to_owned(),
                }],
            },
        );
        assert!(matches!(
            unsupported,
            Err(BundleError::InvalidBundle(detail)) if detail.contains("no implementation admitted")
        ));

        let mut missing_profile: Value = serde_json::from_slice(&wire).unwrap();
        missing_profile["implementations"][0]["runtime"]
            .as_object_mut()
            .unwrap()
            .remove("runtime_profile");
        assert!(matches!(
            ManifestDocument::parse(&serde_json::to_vec(&missing_profile).unwrap()),
            Err(BundleError::InvalidManifest(detail)) if detail.contains("runtime_profile")
        ));
    }

    #[test]
    fn v4_release_accepts_a_providerless_lifecycle_implementation() {
        let root = tempfile::tempdir().unwrap();
        let script = root.path().join("plugin.js");
        fs::write(&script, b"export default {};\n").unwrap();
        let output = root.path().join("example.lifecycle.lenso-plugin");
        let contract = PluginContract::new("example.lifecycle", "1.0.0", "workflows")
            .with_authoring_version(2)
            .with_requirement(
                CapabilityRequirementPlan::one("example.store@1", "1.0.0")
                    .with_requirement_id("store"),
            );

        let verified = build_source_plugin_release_bundle(&SourcePluginReleaseBuild {
            contract,
            implementations: vec![SourcePluginImplementation {
                id: "bun".to_owned(),
                host_targets: vec!["*".to_owned()],
                artifact: script,
                bundle_path: "implementations/bun/plugin.js".to_owned(),
                media_type: "application/javascript".to_owned(),
                target: "javascript-bun".to_owned(),
                entrypoint: "plugin.js".to_owned(),
                execution_class: ExecutionClassId::new("lenso.bun-process@1"),
                runtime_profile: "lenso.bun-authoring@2".to_owned(),
            }],
            output: output.clone(),
        })
        .unwrap();

        assert_eq!(verified, verify_bundle_directory(&output).unwrap());
        let manifest = read_bundle_manifest(&output).unwrap();
        let selected = resolve_implementation(
            &manifest,
            &ImplementationPolicy {
                host_target: "test-host".to_owned(),
                runtimes: vec![RuntimeAdmission {
                    execution_class: ExecutionClassId::new("lenso.bun-process@1"),
                    runtime_profile: "lenso.bun-authoring@2".to_owned(),
                }],
            },
        )
        .unwrap();
        assert!(selected.descriptor.provided_capabilities().is_empty());
        assert_eq!(
            selected.descriptor.required_capabilities()[0].requirement_id(),
            "store"
        );
    }
}