byteor_runtime/

compatibility.rs

use std::collections::BTreeMap;
use std::path::Path;

use byteor_app_kit::approval::{
    approval_spec_hash_from_canonical_spec_kv, required_approval_capabilities_for_spec_kv,
};
use byteor_app_kit::capabilities::{
    ExecutionTarget, PolicyClassId, ReplaySupport, RuntimeCapabilityDocument, RuntimeEnvironment,
    StageCatalogEntry, StageKeyKind,
};
use byteor_core::config::Environment;
use serde::Serialize;

use crate::{canonical_encode_spec_kv_v1, default_runtime_capability_document, read_spec_kv_v1};

/// Schema version for runtime compatibility reports.
pub(crate) const COMPATIBILITY_REPORT_VERSION: u32 = 1;

/// How a stage requirement was satisfied in the report.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StageRequirementSource {
    /// Supported as an intrinsic built-in stage op.
    BuiltIn,
    /// Matched a stable catalog key exactly.
    CatalogStable,
    /// Matched a parameterized catalog family.
    CatalogPrefix,
    /// No supporting catalog entry was found.
    Unknown,
}

/// One stage-related compatibility requirement derived from a spec.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StageCompatibilityRequirement {
    /// Stable stage identifier or family-expanded stage key.
    pub key: String,
    /// How the requirement was satisfied.
    pub source: StageRequirementSource,
    /// Whether the runtime can execute this stage requirement.
    pub supported: bool,
    /// Policy class, when known.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub policy_class: Option<PolicyClassId>,
    /// Replay posture for this stage requirement.
    pub replay_support: ReplaySupport,
    /// Approval capability ids implied by this stage requirement.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub approval_capabilities: Vec<String>,
}

/// Machine-readable compatibility report for a authored spec against a runtime surface.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompatibilityReport {
    /// Report schema version.
    pub schema_version: u32,
    /// Runtime product identity that produced the report.
    pub product: String,
    /// Requested environment posture.
    pub environment: RuntimeEnvironment,
    /// Canonical spec hash.
    pub spec_hash: String,
    /// Lowered execution target.
    pub target: ExecutionTarget,
    /// Whether the spec is executable on this runtime surface.
    pub compatible: bool,
    /// Whether deploy-time approval is required in the requested environment.
    pub requires_approval: bool,
    /// Approval capabilities required by the authored spec.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub required_approval_capabilities: Vec<String>,
    /// Effective replay posture after considering all stage requirements.
    pub replay_support: ReplaySupport,
    /// Stage-level requirements derived from the spec.
    pub stage_requirements: Vec<StageCompatibilityRequirement>,
    /// Stage keys the runtime does not recognize.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub unknown_stage_keys: Vec<String>,
    /// Model-specific run modes available for the target.
    pub supported_run_modes: Vec<String>,
    /// Human-readable reasons for incompatibility or deployment blockers.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub reasons: Vec<String>,
}

/// Build a compatibility report for a spec against the default runtime capability surface.
pub fn compatibility_report_for_spec(
    spec_path: &Path,
    env: Environment,
) -> Result<CompatibilityReport, String> {
    let spec = read_spec_kv_v1(spec_path)?;
    let canonical_spec_kv = canonical_encode_spec_kv_v1(spec_path)?;
    Ok(compatibility_report_for_document(
        &default_runtime_capability_document("byteor-runtime"),
        &spec,
        &canonical_spec_kv,
        env,
    ))
}

/// Build a compatibility report for a spec against a supplied capability document.
pub fn compatibility_report_for_document(
    document: &RuntimeCapabilityDocument,
    spec: &byteor_pipeline_spec::PipelineSpecV1,
    canonical_spec_kv: &str,
    env: Environment,
) -> CompatibilityReport {
    let target = match spec {
        byteor_pipeline_spec::PipelineSpecV1::SingleRing(_) => ExecutionTarget::SingleRing,
        byteor_pipeline_spec::PipelineSpecV1::LaneGraph(_) => ExecutionTarget::LaneGraph,
    };
    let environment = runtime_environment(env);
    let required_approval_capabilities = required_approval_capabilities_for_spec_kv(canonical_spec_kv);
    let requires_approval = !required_approval_capabilities.is_empty()
        && matches!(environment, RuntimeEnvironment::Staging | RuntimeEnvironment::Prod);

    let mut requirements = collect_stage_requirements(document, spec);
    requirements.sort_by(|left, right| left.key.cmp(&right.key));

    let unknown_stage_keys: Vec<String> = requirements
        .iter()
        .filter(|requirement| !requirement.supported)
        .map(|requirement| requirement.key.clone())
        .collect();

    let mut reasons = Vec::new();
    if !document.supported_targets.contains(&target) {
        reasons.push(format!(
            "target {:?} is not supported by {}",
            target, document.product
        ));
    }
    if !unknown_stage_keys.is_empty() {
        reasons.push(format!(
            "unknown stage keys: {}",
            unknown_stage_keys.join(", ")
        ));
    }
    if requires_approval {
        reasons.push(format!(
            "approval required in {} for capabilities: {}",
            environment_label(environment),
            required_approval_capabilities.join(", ")
        ));
    }

    let replay_support = requirements.iter().fold(ReplaySupport::Allowed, |current, requirement| {
        max_replay_support(current, requirement.replay_support)
    });

    CompatibilityReport {
        schema_version: COMPATIBILITY_REPORT_VERSION,
        product: document.product.clone(),
        environment,
        spec_hash: approval_spec_hash_from_canonical_spec_kv(canonical_spec_kv),
        target,
        compatible: document.supported_targets.contains(&target) && unknown_stage_keys.is_empty(),
        requires_approval,
        required_approval_capabilities,
        replay_support,
        stage_requirements: requirements,
        unknown_stage_keys,
        supported_run_modes: supported_run_modes(target),
        reasons,
    }
}

fn collect_stage_requirements(
    document: &RuntimeCapabilityDocument,
    spec: &byteor_pipeline_spec::PipelineSpecV1,
) -> Vec<StageCompatibilityRequirement> {
    let mut requirements = BTreeMap::<String, StageCompatibilityRequirement>::new();

    match spec {
        byteor_pipeline_spec::PipelineSpecV1::SingleRing(single_ring) => {
            for stage in &single_ring.stages {
                match &stage.op {
                    byteor_pipeline_spec::StageOpV1::Identity => {
                        requirements.entry("builtin:identity".to_string()).or_insert_with(|| {
                            builtin_requirement("builtin:identity")
                        });
                    }
                    byteor_pipeline_spec::StageOpV1::AddU8 { .. } => {
                        requirements.entry("builtin:add_u8".to_string()).or_insert_with(|| {
                            builtin_requirement("builtin:add_u8")
                        });
                    }
                    byteor_pipeline_spec::StageOpV1::ResolverKey { stage } => {
                        requirements
                            .entry(stage.clone())
                            .or_insert_with(|| catalog_requirement(document, stage));
                    }
                }
            }
        }
        byteor_pipeline_spec::PipelineSpecV1::LaneGraph(lane_graph) => {
            for role in &lane_graph.roles {
                if let byteor_pipeline_spec::RoleCfgV1::Stage(stage_role) = role {
                    let stage = &stage_role.stage;
                    requirements
                        .entry(stage.clone())
                        .or_insert_with(|| catalog_requirement(document, stage));
                }
            }
        }
    }

    requirements.into_values().collect()
}

fn builtin_requirement(key: &str) -> StageCompatibilityRequirement {
    StageCompatibilityRequirement {
        key: key.to_string(),
        source: StageRequirementSource::BuiltIn,
        supported: true,
        policy_class: Some(PolicyClassId::PureTransform),
        replay_support: ReplaySupport::Allowed,
        approval_capabilities: Vec::new(),
    }
}

fn catalog_requirement(
    document: &RuntimeCapabilityDocument,
    stage_key: &str,
) -> StageCompatibilityRequirement {
    match match_stage_catalog_entry(document, stage_key) {
        Some((entry, source)) => StageCompatibilityRequirement {
            key: stage_key.to_string(),
            source,
            supported: true,
            policy_class: Some(entry.policy_class),
            replay_support: entry.replay_support,
            approval_capabilities: entry.approval_capabilities.clone(),
        },
        None => StageCompatibilityRequirement {
            key: stage_key.to_string(),
            source: StageRequirementSource::Unknown,
            supported: false,
            policy_class: None,
            replay_support: ReplaySupport::Unsupported,
            approval_capabilities: Vec::new(),
        },
    }
}

fn match_stage_catalog_entry<'a>(
    document: &'a RuntimeCapabilityDocument,
    stage_key: &str,
) -> Option<(&'a StageCatalogEntry, StageRequirementSource)> {
    document.stage_catalog.iter().find_map(|entry| match entry.key_kind {
        StageKeyKind::Stable if entry.key == stage_key => {
            Some((entry, StageRequirementSource::CatalogStable))
        }
        StageKeyKind::Prefix => {
            let prefix = entry.key.trim_end_matches('*');
            if stage_key.starts_with(prefix) {
                Some((entry, StageRequirementSource::CatalogPrefix))
            } else {
                None
            }
        }
        _ => None,
    })
}

fn runtime_environment(env: Environment) -> RuntimeEnvironment {
    match env {
        Environment::Dev => RuntimeEnvironment::Dev,
        Environment::Staging => RuntimeEnvironment::Staging,
        Environment::Prod => RuntimeEnvironment::Prod,
    }
}

fn environment_label(env: RuntimeEnvironment) -> &'static str {
    match env {
        RuntimeEnvironment::Dev => "dev",
        RuntimeEnvironment::Staging => "staging",
        RuntimeEnvironment::Prod => "prod",
    }
}

fn supported_run_modes(target: ExecutionTarget) -> Vec<String> {
    match target {
        ExecutionTarget::SingleRing => vec![
            "shm_supervised".to_string(),
            "one_shot".to_string(),
            "adapter_loop".to_string(),
        ],
        ExecutionTarget::LaneGraph => vec![
            "shm_supervised".to_string(),
            "multi_process".to_string(),
        ],
    }
}

fn max_replay_support(left: ReplaySupport, right: ReplaySupport) -> ReplaySupport {
    match (left, right) {
        (ReplaySupport::Unsupported, _) | (_, ReplaySupport::Unsupported) => ReplaySupport::Unsupported,
        (ReplaySupport::DryRunOnly, _) | (_, ReplaySupport::DryRunOnly) => ReplaySupport::DryRunOnly,
        _ => ReplaySupport::Allowed,
    }
}