//! Compiler-owned workflow primitive loading, validation, and materialization.

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
    fs, io,
    path::{Path, PathBuf},
};

use serde::de::DeserializeOwned;
use serde_json::Value;
use sha2::{Digest, Sha256};

use super::{
    assets::{CompilerAssetError, CompilerAssetResult},
    contracts::{
        CompiledWorkflowPrimitiveBundle, FrozenGraphPlanePlan, MaterializedGraphNodePlan,
        ResolvedAssetRef,
    },
};
use crate::contracts::{
    ArtifactContractDefinition, LaneConflictPolicyDefinition, LifecycleMutationPlanDefinition,
    Plane, PlaneQueueClaimPolicyDefinition, RequestContextProfileDefinition,
    RequestContextProviderDefinition, RequestContextRenderPlan, RuntimeEffectHandlerDefinition,
    RuntimeEffectOperationDefinition, RuntimeEffectOperationRunnerDefinition,
    RuntimeEffectRuleDefinition, RuntimeEffectStoreDefinition, RuntimeEffectValidatorDefinition,
    RuntimeFailurePolicyDefinition, RuntimeJsonContract, TerminalActionDefinition,
    WorkItemDocumentAdapterDefinition, WorkItemFamilyDefinition, WorkflowLaneDefinition,
    WorkflowPlaneSchedulerPolicyDefinition, WorkflowRecoveryPolicyDefinition,
    WorkspaceSchemaEpochDefinition,
};
use crate::workspace::WorkspacePaths;

/// Current workspace-schema epoch expected by the Rust v0.20.0 compiler authority surface.
pub const WORKSPACE_SCHEMA_EPOCH_ID: &str = "v0.20";

const RUNTIME_EFFECT_HANDLER_IMPLEMENTATIONS: &[&str] = &[
    "planner_disposition",
    "manager_blueprint_manifest_to_blueprint_drafts",
    "contractor_blueprint_candidate_persist",
    "evaluator_blueprint_approved_to_task",
    "evaluator_blueprint_rejected_to_draft_revision",
    "mechanic_blueprint_repair_apply",
];

const SUPPORTED_RUNTIME_EFFECT_PRIMITIVE_IDS: &[&str] = &[
    "artifact_presence",
    "artifact_model_parse",
    "active_work_item_lookup",
    "blueprint_critique_packet_validation",
    "blueprint_evaluation_packet_validation",
    "blueprint_generated_task_validation",
    "blueprint_manifest_sequence_validation",
    "blueprint_packet_draft_validation",
    "copy_artifact",
    "enqueue_work_items",
    "legacy_python_handler",
    "move_record",
    "mutation_journal_append",
    "persist_record",
    "source_lifecycle",
    "store_equivalence_check",
    "work_item_patch",
];

const ARTIFACT_CONTRACTS_PATH: &str = "registry/artifact_contracts/default_artifact_contracts.json";
const BUILTIN_DOCUMENT_ADAPTER_PATH: &str = "registry/document_adapters/builtin_markdown_v1.json";
const BLUEPRINT_DOCUMENT_ADAPTER_PATH: &str =
    "registry/document_adapters/blueprint_draft_markdown_v1.json";
const LIFECYCLE_MUTATION_PLANS_PATH: &str =
    "registry/lifecycle_mutation_plans/default_lifecycle_mutations.json";
const QUEUE_CLAIM_POLICIES_PATH: &str =
    "registry/queue_claim_policies/default_queue_claim_policies.json";
const REQUEST_CONTEXT_PROFILES_PATH: &str =
    "registry/request_context_profiles/default_request_context_profiles.json";
const REQUEST_CONTEXT_PROVIDERS_PATH: &str =
    "registry/request_context_providers/default_request_context_providers.json";
const REQUEST_CONTEXT_RENDER_PLANS_PATH: &str =
    "registry/request_context_render_plans/default_request_context_render_plans.json";
const RUNTIME_EFFECT_HANDLERS_PATH: &str =
    "registry/runtime_effect_handlers/default_effect_handlers.json";
const RUNTIME_EFFECT_RUNNERS_PATH: &str =
    "registry/runtime_effect_runners/default_effect_runners.json";
const BLUEPRINT_EFFECT_RULES_PATH: &str =
    "registry/runtime_effect_rules/blueprint_effect_rules.json";
const PLANNER_EFFECT_RULES_PATH: &str = "registry/runtime_effect_rules/planner_effect_rules.json";
const EFFECT_STORES_PATH: &str = "registry/effect_stores/default_effect_stores.json";
const EFFECT_VALIDATORS_PATH: &str = "registry/effect_validators/default_effect_validators.json";
const RUNTIME_EFFECT_OPERATIONS_PATH: &str =
    "registry/runtime_effect_operations/default_runtime_effect_operations.json";
const RUNTIME_FAILURE_POLICIES_PATH: &str =
    "registry/runtime_failure_policies/default_runtime_failure_policies.json";
const TERMINAL_ACTIONS_PATH: &str = "registry/terminal_actions/default_terminal_actions.json";
const WORKSPACE_SCHEMA_EPOCH_PATH: &str = "registry/workspace_schema_epochs/current.json";
const RECOVERY_POLICIES_PATH: &str = "registry/recovery_policies/default_recovery_policies.json";

const WORK_ITEM_FAMILY_PATHS: &[&str] = &[
    "registry/work_item_families/blueprint_draft.json",
    "registry/work_item_families/incident.json",
    "registry/work_item_families/learning_request.json",
    "registry/work_item_families/probe.json",
    "registry/work_item_families/spec.json",
    "registry/work_item_families/task.json",
];

/// Resolved workflow primitive authority for one compile selection.
#[derive(Debug, Clone, PartialEq)]
pub struct CompiledWorkflowPrimitiveAuthority {
    pub workflow_primitives: CompiledWorkflowPrimitiveBundle,
    pub workflow_primitive_fingerprints: BTreeMap<String, String>,
    pub lane_policy: WorkflowPlaneSchedulerPolicyDefinition,
    pub resolved_assets: Vec<ResolvedAssetRef>,
}

/// Load, select, and validate workflow primitive authority for selected graph loops.
pub fn load_workflow_primitive_authority(
    paths: &WorkspacePaths,
    graph_loops: &[super::assets::ResolvedGraphLoopAsset],
) -> CompilerAssetResult<CompiledWorkflowPrimitiveAuthority> {
    let loaded = load_all_workflow_primitives(paths)?;
    let selected = select_workflow_primitives(&loaded, graph_loops);
    let lane_policy = build_lane_policy(&selected.queue_claim_policies, graph_loops)?;
    validate_workflow_primitive_authority(paths, &selected, &lane_policy, graph_loops)?;

    Ok(CompiledWorkflowPrimitiveAuthority {
        workflow_primitives: selected,
        workflow_primitive_fingerprints: loaded.fingerprints,
        lane_policy,
        resolved_assets: loaded.resolved_assets,
    })
}

/// Apply selected workflow primitive authority to one frozen graph plan.
pub fn materialize_workflow_primitive_graph_authority(
    graph: &mut FrozenGraphPlanePlan,
    primitives: &CompiledWorkflowPrimitiveBundle,
    lane_policy: &WorkflowPlaneSchedulerPolicyDefinition,
) -> Result<(), super::contracts::CompilerContractError> {
    let rules_by_node_outcome = rules_by_node_outcome(&primitives.runtime_effect_rules);
    let terminal_states_by_id: HashMap<_, _> = graph
        .terminal_states
        .iter()
        .map(|state| {
            (
                state.terminal_state_id.clone(),
                state.terminal_class.as_str().to_owned(),
            )
        })
        .collect();
    let compiled_transitions = graph.compiled_transitions.clone();

    for node in &mut graph.nodes {
        node.lane_id = lane_id_for_plane(lane_policy, graph.plane);
        node.request_context_profile_id =
            request_context_profile_id(node, &primitives.request_context_profiles);
        node.context_render_plan_id = context_render_plan_id(
            node,
            &primitives.request_context_profiles,
            &primitives.request_context_render_plans,
        )?;
        node.runtime_effect_rule_selections =
            runtime_effect_rule_ids_for_node(node, &primitives.runtime_effect_rules);
        node.terminal_action_mappings = terminal_action_mappings_for_node(
            node,
            &compiled_transitions,
            primitives,
            &rules_by_node_outcome,
            &terminal_states_by_id,
        )?;
        node.validate()?;
    }
    graph.validate()
}

#[derive(Default)]
struct LoadedWorkflowPrimitives {
    artifact_contracts: Vec<ArtifactContractDefinition>,
    request_context_profiles: Vec<RequestContextProfileDefinition>,
    request_context_providers: Vec<RequestContextProviderDefinition>,
    request_context_render_plans: Vec<RequestContextRenderPlan>,
    work_item_families: Vec<WorkItemFamilyDefinition>,
    document_adapters: Vec<WorkItemDocumentAdapterDefinition>,
    queue_claim_policies: Vec<PlaneQueueClaimPolicyDefinition>,
    terminal_actions: Vec<TerminalActionDefinition>,
    lifecycle_mutation_plans: Vec<LifecycleMutationPlanDefinition>,
    runtime_effect_handlers: Vec<RuntimeEffectHandlerDefinition>,
    runtime_effect_runners: Vec<RuntimeEffectOperationRunnerDefinition>,
    runtime_effect_rules: Vec<RuntimeEffectRuleDefinition>,
    effect_stores: Vec<RuntimeEffectStoreDefinition>,
    effect_validators: Vec<RuntimeEffectValidatorDefinition>,
    runtime_effect_operations: Vec<RuntimeEffectOperationDefinition>,
    recovery_policies: Vec<WorkflowRecoveryPolicyDefinition>,
    runtime_failure_policies: Vec<RuntimeFailurePolicyDefinition>,
    workspace_schema_epoch: Option<WorkspaceSchemaEpochDefinition>,
    resolved_assets: Vec<ResolvedAssetRef>,
    fingerprints: BTreeMap<String, String>,
}

fn load_all_workflow_primitives(
    paths: &WorkspacePaths,
) -> CompilerAssetResult<LoadedWorkflowPrimitives> {
    let mut loaded = LoadedWorkflowPrimitives::default();

    loaded.artifact_contracts = load_collection(
        paths,
        ARTIFACT_CONTRACTS_PATH,
        "artifact_contracts",
        &mut loaded,
    )?;
    let builtin_document_adapters = load_single(
        paths,
        BUILTIN_DOCUMENT_ADAPTER_PATH,
        "document_adapters",
        &mut loaded,
    )?;
    loaded.document_adapters.extend(builtin_document_adapters);
    let blueprint_document_adapters = load_single(
        paths,
        BLUEPRINT_DOCUMENT_ADAPTER_PATH,
        "document_adapters",
        &mut loaded,
    )?;
    loaded.document_adapters.extend(blueprint_document_adapters);
    loaded.lifecycle_mutation_plans = load_collection(
        paths,
        LIFECYCLE_MUTATION_PLANS_PATH,
        "lifecycle_mutation_plans",
        &mut loaded,
    )?;
    loaded.queue_claim_policies = load_collection(
        paths,
        QUEUE_CLAIM_POLICIES_PATH,
        "queue_claim_policies",
        &mut loaded,
    )?;
    loaded.request_context_profiles = load_collection(
        paths,
        REQUEST_CONTEXT_PROFILES_PATH,
        "request_context_profiles",
        &mut loaded,
    )?;
    loaded.request_context_providers = load_collection(
        paths,
        REQUEST_CONTEXT_PROVIDERS_PATH,
        "request_context_providers",
        &mut loaded,
    )?;
    loaded.request_context_render_plans = load_collection(
        paths,
        REQUEST_CONTEXT_RENDER_PLANS_PATH,
        "request_context_render_plans",
        &mut loaded,
    )?;
    loaded.runtime_effect_handlers = load_collection(
        paths,
        RUNTIME_EFFECT_HANDLERS_PATH,
        "runtime_effect_handlers",
        &mut loaded,
    )?;
    loaded.runtime_effect_runners = load_collection(
        paths,
        RUNTIME_EFFECT_RUNNERS_PATH,
        "runtime_effect_runners",
        &mut loaded,
    )?;
    let planner_effect_rules = load_collection(
        paths,
        PLANNER_EFFECT_RULES_PATH,
        "runtime_effect_rules",
        &mut loaded,
    )?;
    loaded.runtime_effect_rules.extend(planner_effect_rules);
    let blueprint_effect_rules = load_collection(
        paths,
        BLUEPRINT_EFFECT_RULES_PATH,
        "runtime_effect_rules",
        &mut loaded,
    )?;
    loaded.runtime_effect_rules.extend(blueprint_effect_rules);
    loaded.effect_stores =
        load_collection(paths, EFFECT_STORES_PATH, "effect_stores", &mut loaded)?;
    loaded.effect_validators = load_collection(
        paths,
        EFFECT_VALIDATORS_PATH,
        "effect_validators",
        &mut loaded,
    )?;
    loaded.runtime_effect_operations = load_collection(
        paths,
        RUNTIME_EFFECT_OPERATIONS_PATH,
        "runtime_effect_operations",
        &mut loaded,
    )?;
    loaded.runtime_failure_policies = load_collection(
        paths,
        RUNTIME_FAILURE_POLICIES_PATH,
        "runtime_failure_policies",
        &mut loaded,
    )?;
    loaded.terminal_actions = load_collection(
        paths,
        TERMINAL_ACTIONS_PATH,
        "terminal_actions",
        &mut loaded,
    )?;
    loaded.recovery_policies = load_collection(
        paths,
        RECOVERY_POLICIES_PATH,
        "recovery_policies",
        &mut loaded,
    )?;
    loaded.workspace_schema_epoch = Some(load_single_definition(
        paths,
        WORKSPACE_SCHEMA_EPOCH_PATH,
        "workspace_schema_epoch",
        &mut loaded,
    )?);

    for relative_path in WORK_ITEM_FAMILY_PATHS {
        let families = load_single(paths, relative_path, "work_item_families", &mut loaded)?;
        loaded.work_item_families.extend(families);
    }

    Ok(loaded)
}

fn select_workflow_primitives(
    loaded: &LoadedWorkflowPrimitives,
    graph_loops: &[super::assets::ResolvedGraphLoopAsset],
) -> CompiledWorkflowPrimitiveBundle {
    let selected_planes: HashSet<_> = graph_loops.iter().map(|graph| graph.plane).collect();
    let selected_nodes = selected_node_ids(graph_loops);
    let selected_profile_ids =
        selected_request_context_profile_ids(graph_loops, &loaded.request_context_profiles);
    let selected_rules: Vec<_> = loaded
        .runtime_effect_rules
        .iter()
        .filter(|rule| selected_nodes.contains(rule.source_node_id.as_str()))
        .cloned()
        .collect();
    let selected_recovery_policies = loaded
        .recovery_policies
        .iter()
        .filter(|policy| {
            policy
                .source_node_ids
                .iter()
                .any(|node_id| selected_nodes.contains(node_id.as_str()))
        })
        .cloned()
        .collect();

    CompiledWorkflowPrimitiveBundle {
        artifact_contracts: loaded.artifact_contracts.clone(),
        request_context_profiles: loaded
            .request_context_profiles
            .iter()
            .filter(|profile| selected_profile_ids.contains(profile.profile_id.as_str()))
            .cloned()
            .collect(),
        request_context_providers: loaded.request_context_providers.clone(),
        request_context_render_plans: loaded.request_context_render_plans.clone(),
        work_item_families: loaded.work_item_families.clone(),
        document_adapters: loaded.document_adapters.clone(),
        queue_claim_policies: loaded
            .queue_claim_policies
            .iter()
            .filter(|policy| selected_planes.contains(&policy.plane))
            .cloned()
            .collect(),
        terminal_actions: loaded.terminal_actions.clone(),
        lifecycle_mutation_plans: loaded.lifecycle_mutation_plans.clone(),
        runtime_effect_handlers: loaded.runtime_effect_handlers.clone(),
        runtime_effect_runners: loaded.runtime_effect_runners.clone(),
        runtime_effect_rules: selected_rules,
        effect_stores: loaded.effect_stores.clone(),
        effect_validators: loaded.effect_validators.clone(),
        runtime_effect_operations: loaded.runtime_effect_operations.clone(),
        recovery_policies: selected_recovery_policies,
        runtime_failure_policies: loaded.runtime_failure_policies.clone(),
        workspace_schema_epoch: loaded.workspace_schema_epoch.clone(),
    }
}

fn validate_workflow_primitive_authority(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
    lane_policy: &WorkflowPlaneSchedulerPolicyDefinition,
    graph_loops: &[super::assets::ResolvedGraphLoopAsset],
) -> CompilerAssetResult<()> {
    validate_work_item_family_document_adapters(paths, primitives)?;
    validate_request_context_profiles(paths, primitives)?;
    validate_runtime_effect_rules(paths, primitives)?;
    validate_runtime_effect_operations(paths, primitives)?;
    validate_artifact_contract_operation_bindings(paths, primitives)?;
    validate_runtime_failure_policy_effect_scopes(paths, primitives)?;
    validate_lifecycle_and_terminal_actions(paths, primitives)?;
    validate_lane_policy(paths, lane_policy)?;
    validate_workspace_schema_epoch(paths, primitives)?;
    validate_graph_workflow_references(paths, primitives, graph_loops)?;
    Ok(())
}

fn validate_work_item_family_document_adapters(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let families: HashMap<_, _> = primitives
        .work_item_families
        .iter()
        .map(|family| (family.family_id.as_str(), family))
        .collect();
    let adapters: HashMap<_, _> = primitives
        .document_adapters
        .iter()
        .map(|adapter| (adapter.adapter_id.as_str(), adapter))
        .collect();
    for family in &primitives.work_item_families {
        let Some(adapter) = adapters.get(family.document_adapter_id.as_str()) else {
            return Err(invalid_primitive(
                paths,
                "invalid work-item family document adapter reference",
            ));
        };
        if !adapter.family_ids.iter().any(|id| id == &family.family_id) {
            return Err(invalid_primitive(
                paths,
                "document adapter family_ids must include referencing work-item family",
            ));
        }
    }
    for adapter in &primitives.document_adapters {
        for family_id in &adapter.family_ids {
            if !families.contains_key(family_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "document adapter references unknown work-item family",
                ));
            }
        }
    }
    Ok(())
}

fn validate_request_context_profiles(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let providers_by_id: HashMap<_, _> = primitives
        .request_context_providers
        .iter()
        .map(|provider| (provider.provider_id.as_str(), provider))
        .collect();
    let render_plans_by_id: HashMap<_, _> = primitives
        .request_context_render_plans
        .iter()
        .map(|render_plan| (render_plan.render_plan_id.as_str(), render_plan))
        .collect();
    let artifact_contracts_by_id: HashMap<_, _> = primitives
        .artifact_contracts
        .iter()
        .map(|contract| (contract.artifact_id.as_str(), contract))
        .collect();

    for profile in &primitives.request_context_profiles {
        let Some(provider) = providers_by_id.get(profile.provider_id.as_str()) else {
            return Err(invalid_primitive(
                paths,
                format!(
                    "request context profile {} references unknown provider {}",
                    profile.profile_id, profile.provider_id
                ),
            ));
        };
        if !provider
            .supported_request_kinds
            .iter()
            .any(|kind| kind == &profile.request_kind)
        {
            return Err(invalid_primitive(
                paths,
                format!(
                    "request context profile {} request kind {} is not supported by provider {}",
                    profile.profile_id, profile.request_kind, profile.provider_id
                ),
            ));
        }
        let Some(render_plan) = render_plans_by_id.get(profile.primary_render_plan_id.as_str())
        else {
            return Err(invalid_primitive(
                paths,
                format!(
                    "request context profile {} references unknown render plan {}",
                    profile.profile_id, profile.primary_render_plan_id
                ),
            ));
        };
        let provider_capabilities: HashSet<_> =
            provider.capabilities.iter().map(String::as_str).collect();
        let missing_render_capabilities: Vec<_> = render_plan
            .required_provider_capabilities
            .iter()
            .map(String::as_str)
            .filter(|capability| !provider_capabilities.contains(capability))
            .collect();
        if !missing_render_capabilities.is_empty() {
            return Err(invalid_primitive(
                paths,
                format!(
                    "request context profile {} render plan {} requires provider capabilities not declared by {}: {}",
                    profile.profile_id,
                    render_plan.render_plan_id,
                    provider.provider_id,
                    missing_render_capabilities.join(", ")
                ),
            ));
        }
        let missing_required_providers: Vec<_> = profile
            .required_providers
            .iter()
            .map(String::as_str)
            .filter(|provider_capability| !provider_capabilities.contains(provider_capability))
            .collect();
        if !missing_required_providers.is_empty() {
            return Err(invalid_primitive(
                paths,
                format!(
                    "request context profile {} requires provider capabilities not declared by {}: {}",
                    profile.profile_id,
                    provider.provider_id,
                    missing_required_providers.join(", ")
                ),
            ));
        }
        for (artifact_id, filename) in &profile.output_path_preferences {
            let Some(contract) = artifact_contracts_by_id.get(artifact_id.as_str()) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "request context profile {} references unknown output artifact {}",
                        profile.profile_id, artifact_id
                    ),
                ));
            };
            let mut allowed_filenames = vec![contract.canonical_filename.as_str()];
            allowed_filenames.extend(contract.accepted_filenames.iter().map(String::as_str));
            if !allowed_filenames.contains(&filename.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "request context profile {} maps artifact {} to filename {}; artifact contract mismatch",
                        profile.profile_id, artifact_id, filename
                    ),
                ));
            }
        }
    }
    Ok(())
}

fn validate_runtime_effect_rules(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let artifact_contracts_by_id: HashMap<_, _> = primitives
        .artifact_contracts
        .iter()
        .map(|contract| (contract.artifact_id.as_str(), contract))
        .collect();
    let handlers_by_id: BTreeMap<_, _> = primitives
        .runtime_effect_handlers
        .iter()
        .map(|handler| (handler.handler_id.as_str(), handler))
        .collect();
    let implemented: HashSet<_> = RUNTIME_EFFECT_HANDLER_IMPLEMENTATIONS
        .iter()
        .copied()
        .collect();
    let families: HashSet<_> = primitives
        .work_item_families
        .iter()
        .map(|family| family.family_id.as_str())
        .collect();
    let mut bindings = BTreeMap::new();

    for handler in &primitives.runtime_effect_handlers {
        if !implemented.contains(handler.handler_id.as_str()) {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect handler {} has no packaged Rust implementation",
                    handler.handler_id
                ),
            ));
        }
        for family_id in &handler.allowed_source_families {
            if !families.contains(family_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "runtime effect handler references unknown source family",
                ));
            }
        }
    }

    for rule in &primitives.runtime_effect_rules {
        for artifact_id in &rule.required_run_artifacts {
            if !artifact_contracts_by_id.contains_key(artifact_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect rule {} requires unknown artifact {}",
                        rule.rule_id, artifact_id
                    ),
                ));
            }
        }
        if let Some(handler_id) = rule.handler_id.as_deref() {
            let Some(handler) = handlers_by_id.get(handler_id) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect rule {} references unknown handler {}",
                        rule.rule_id, handler_id
                    ),
                ));
            };
            let handler_required_artifacts: HashSet<_> = handler
                .required_artifacts
                .iter()
                .map(String::as_str)
                .collect();
            let handler_declared_artifacts: HashSet<_> = handler
                .required_artifacts
                .iter()
                .chain(handler.optional_artifacts.iter())
                .map(String::as_str)
                .collect();
            let rule_required_artifacts: HashSet<_> = rule
                .required_run_artifacts
                .iter()
                .map(String::as_str)
                .collect();
            for artifact_id in &rule.required_run_artifacts {
                if !handler_declared_artifacts.contains(artifact_id.as_str()) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect rule {} requires artifact {} not declared by handler {}",
                            rule.rule_id, artifact_id, handler_id
                        ),
                    ));
                }
            }
            for artifact_id in handler_required_artifacts {
                if !rule_required_artifacts.contains(artifact_id) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect handler {} requires artifact {} missing from runtime effect rule {}",
                            handler_id, artifact_id, rule.rule_id
                        ),
                    ));
                }
            }
            let declared_capabilities: BTreeSet<_> = handler
                .declared_capabilities
                .iter()
                .map(String::as_str)
                .collect();
            let missing_capabilities: Vec<_> = rule
                .required_handler_capabilities
                .iter()
                .map(String::as_str)
                .filter(|capability_id| !declared_capabilities.contains(capability_id))
                .collect();
            if !missing_capabilities.is_empty() {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect rule {} selected handler {} missing required capabilities: {}",
                        rule.rule_id,
                        handler_id,
                        missing_capabilities.join(", ")
                    ),
                ));
            }
        }
        if let Some(destination_family_id) = &rule.destination_family_id {
            if !families.contains(destination_family_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "runtime effect rule references unknown destination family",
                ));
            }
        }
        for outcome in &rule.on_outcomes {
            let key = (rule.source_node_id.as_str(), outcome.as_str());
            if let Some(existing) = bindings.insert(key, rule.rule_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "duplicate runtime-effect binding for {} {outcome}: {existing} and {}",
                        rule.source_node_id, rule.rule_id
                    ),
                ));
            }
        }
    }
    Ok(())
}

fn validate_runtime_effect_operations(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let artifact_contracts_by_id: HashMap<_, _> = primitives
        .artifact_contracts
        .iter()
        .map(|contract| (contract.artifact_id.as_str(), contract))
        .collect();
    let effect_stores_by_id: HashMap<_, _> = primitives
        .effect_stores
        .iter()
        .map(|store| (store.store_id.as_str(), store))
        .collect();
    let effect_validators_by_id: HashMap<_, _> = primitives
        .effect_validators
        .iter()
        .map(|validator| (validator.validator_id.as_str(), validator))
        .collect();
    let runtime_effect_operations_by_id: HashMap<_, _> = primitives
        .runtime_effect_operations
        .iter()
        .map(|operation| (operation.operation_id.as_str(), operation))
        .collect();

    let mut runners_by_operation_id: HashMap<&str, &RuntimeEffectOperationRunnerDefinition> =
        HashMap::new();
    let mut aliases_by_handler_id: HashMap<&str, (String, &str)> = HashMap::new();
    for runner in &primitives.runtime_effect_runners {
        for operation_id in &runner.operation_ids {
            if !runtime_effect_operations_by_id.contains_key(operation_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect runner {} references unknown operation {}",
                        runner.runner_id, operation_id
                    ),
                ));
            }
            if let Some(previous_runner) =
                runners_by_operation_id.insert(operation_id.as_str(), runner)
            {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect operation {} is owned by multiple runners: {}, {}",
                        operation_id, previous_runner.runner_id, runner.runner_id
                    ),
                ));
            }
        }
        for handler_id in &runner.legacy_handler_ids {
            let Some(operation_id) = runner.operation_id_for_legacy_handler(handler_id) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect runner {} maps legacy handler {} ambiguously",
                        runner.runner_id, handler_id
                    ),
                ));
            };
            let current = (operation_id, runner.runner_id.as_str());
            if let Some(previous) =
                aliases_by_handler_id.insert(handler_id.as_str(), current.clone())
            {
                if previous != current {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "legacy runtime effect handler {} is mapped to multiple operations or runners: {}/{}, {}/{}",
                            handler_id, previous.0, previous.1, current.0, current.1
                        ),
                    ));
                }
            }
        }
    }

    for operation in &primitives.runtime_effect_operations {
        let Some(operation_runner) = runners_by_operation_id.get(operation.operation_id.as_str())
        else {
            continue;
        };
        for handler_id in &operation.legacy_handler_ids {
            if operation_runner
                .operation_id_for_legacy_handler(handler_id)
                .as_deref()
                != Some(operation.operation_id.as_str())
            {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect operation {} legacy handler {} is not mapped by runner {}",
                        operation.operation_id, handler_id, operation_runner.runner_id
                    ),
                ));
            }
        }
    }

    for rule in &primitives.runtime_effect_rules {
        let Some(operation) =
            runtime_effect_operations_by_id.get(rule.effect_operation_id.as_str())
        else {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} references unknown operation {}",
                    rule.rule_id, rule.effect_operation_id
                ),
            ));
        };
        let Some(runner) = runners_by_operation_id.get(operation.operation_id.as_str()) else {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} references operation {} without a runtime effect runner",
                    rule.rule_id, operation.operation_id
                ),
            ));
        };
        if let Some(handler_id) = rule.handler_id.as_deref() {
            if runner
                .operation_id_for_legacy_handler(handler_id)
                .as_deref()
                != Some(operation.operation_id.as_str())
            {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect rule {} handler {} is not a legacy alias for operation {}",
                        rule.rule_id, handler_id, operation.operation_id
                    ),
                ));
            }
        }

        let missing_artifacts: Vec<_> = rule
            .required_run_artifacts
            .iter()
            .filter(|artifact_id| !operation.required_artifacts.contains(*artifact_id))
            .cloned()
            .collect();
        if !missing_artifacts.is_empty() {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} requires artifacts not declared by operation {}: {}",
                    rule.rule_id,
                    operation.operation_id,
                    missing_artifacts.join(", ")
                ),
            ));
        }
        if rule.duplicate_policy != operation.idempotency.duplicate_policy {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} duplicate_policy does not match operation {}",
                    rule.rule_id, operation.operation_id
                ),
            ));
        }
        if rule.replay_policy != operation.idempotency.replay_policy {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} replay_policy does not match operation {}",
                    rule.rule_id, operation.operation_id
                ),
            ));
        }
        let operation_runtime_capabilities: HashSet<_> = runner
            .runtime_capabilities_for_operation(operation.operation_id.as_str())
            .iter()
            .map(String::as_str)
            .collect();
        let missing_capabilities: Vec<_> = rule
            .required_handler_capabilities
            .iter()
            .map(String::as_str)
            .filter(|capability_id| !operation_runtime_capabilities.contains(capability_id))
            .collect();
        if !missing_capabilities.is_empty() {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect rule {} requires runner capability {} not declared by runner {}",
                    rule.rule_id, missing_capabilities[0], runner.runner_id
                ),
            ));
        }
    }

    for validator in &primitives.effect_validators {
        validate_runtime_effect_primitive_id(
            validator.primitive_id.as_str(),
            format!("runtime effect validator {}", validator.validator_id),
            paths,
        )?;
        for artifact_id in &validator.input_artifact_ids {
            if !artifact_contracts_by_id.contains_key(artifact_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect validator {} references unknown artifact {}",
                        validator.validator_id, artifact_id
                    ),
                ));
            }
        }
        for store_id in &validator.store_ids {
            if !effect_stores_by_id.contains_key(store_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect validator {} references unknown store {}",
                        validator.validator_id, store_id
                    ),
                ));
            }
        }
    }

    for operation in &primitives.runtime_effect_operations {
        let known_operation_artifacts: HashSet<_> = operation
            .required_artifacts
            .iter()
            .chain(operation.produced_artifacts.iter())
            .map(String::as_str)
            .collect();
        for artifact_id in &known_operation_artifacts {
            if !artifact_contracts_by_id.contains_key(*artifact_id) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect operation {} references unknown artifact {}",
                        operation.operation_id, artifact_id
                    ),
                ));
            }
        }

        let mut write_step_count = 0usize;
        for step in &operation.steps {
            validate_runtime_effect_primitive_id(
                step.primitive_id.as_str(),
                format!(
                    "runtime effect operation {} step {}",
                    operation.operation_id, step.step_id
                ),
                paths,
            )?;
            for artifact_id in &step.reads_artifact_ids {
                if !known_operation_artifacts.contains(artifact_id.as_str()) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect operation {} step {} references artifact {} not declared by the operation",
                            operation.operation_id, step.step_id, artifact_id
                        ),
                    ));
                }
            }
            if let Some(store_id) = &step.store_id {
                if !effect_stores_by_id.contains_key(store_id.as_str()) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect operation {} step {} references unknown store {}",
                            operation.operation_id, step.step_id, store_id
                        ),
                    ));
                }
            }
            for validator_id in &step.validator_ids {
                let Some(validator) = effect_validators_by_id.get(validator_id.as_str()) else {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect operation {} step {} references unknown validator {}",
                            operation.operation_id, step.step_id, validator_id
                        ),
                    ));
                };
                validate_operation_validator_binding(paths, operation, validator)?;
            }
            if step.writes_store {
                write_step_count += 1;
            }
        }

        for mapping in &operation.failure_mappings {
            if let Some(validator_id) = &mapping.validator_id {
                if !effect_validators_by_id.contains_key(validator_id.as_str()) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime effect operation {} failure mapping {} references unknown validator {}",
                            operation.operation_id, mapping.failure_class, validator_id
                        ),
                    ));
                }
            }
        }
        for validator_id in &operation.idempotency.equivalence_validator_ids {
            let Some(validator) = effect_validators_by_id.get(validator_id.as_str()) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect operation {} idempotency references unknown validator {}",
                        operation.operation_id, validator_id
                    ),
                ));
            };
            validate_operation_validator_binding(paths, operation, validator)?;
        }
        if write_step_count > 1 && operation.partial_commit_policy.is_none() {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect operation {} has multiple write steps without partial_commit_policy",
                    operation.operation_id
                ),
            ));
        }
    }

    Ok(())
}

fn validate_artifact_contract_operation_bindings(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let operations_by_id: HashMap<_, _> = primitives
        .runtime_effect_operations
        .iter()
        .map(|operation| (operation.operation_id.as_str(), operation))
        .collect();
    let mut operation_by_legacy_handler: HashMap<&str, String> = HashMap::new();
    for runner in &primitives.runtime_effect_runners {
        for handler_id in &runner.legacy_handler_ids {
            let Some(operation_id) = runner.operation_id_for_legacy_handler(handler_id) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime effect runner {} maps legacy handler {} ambiguously",
                        runner.runner_id, handler_id
                    ),
                ));
            };
            if let Some(previous) =
                operation_by_legacy_handler.insert(handler_id.as_str(), operation_id.clone())
            {
                if previous != operation_id {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "legacy runtime effect handler {} is mapped to multiple operations: {}, {}",
                            handler_id, previous, operation_id
                        ),
                    ));
                }
            }
        }
    }

    for contract in &primitives.artifact_contracts {
        let declared_consumer_operation_ids: HashSet<_> = contract
            .consumer_operation_ids
            .iter()
            .map(String::as_str)
            .collect();
        for consumer_operation_id in &contract.consumer_operation_ids {
            let Some(operation) = operations_by_id.get(consumer_operation_id.as_str()) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} references unknown consumer operation {}",
                        contract.artifact_id, consumer_operation_id
                    ),
                ));
            };
            let references_artifact = operation
                .required_artifacts
                .iter()
                .chain(operation.produced_artifacts.iter())
                .any(|artifact_id| artifact_id == &contract.artifact_id);
            if !references_artifact {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} consumer operation {} does not reference artifact {}",
                        contract.artifact_id, consumer_operation_id, contract.artifact_id
                    ),
                ));
            }
        }
        for consumer_handler_id in &contract.consumer_handler_ids {
            let Some(mapped_operation_id) =
                operation_by_legacy_handler.get(consumer_handler_id.as_str())
            else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} references unknown consumer handler {}",
                        contract.artifact_id, consumer_handler_id
                    ),
                ));
            };
            let Some(operation) = operations_by_id.get(mapped_operation_id.as_str()) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} consumer handler {} maps to unknown operation {}",
                        contract.artifact_id, consumer_handler_id, mapped_operation_id
                    ),
                ));
            };
            let references_artifact = operation
                .required_artifacts
                .iter()
                .chain(operation.produced_artifacts.iter())
                .any(|artifact_id| artifact_id == &contract.artifact_id);
            if !references_artifact {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} consumer handler {} maps to operation {} that does not reference artifact {}",
                        contract.artifact_id,
                        consumer_handler_id,
                        mapped_operation_id,
                        contract.artifact_id
                    ),
                ));
            }
            if !declared_consumer_operation_ids.is_empty()
                && !declared_consumer_operation_ids.contains(mapped_operation_id.as_str())
            {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "artifact contract {} consumer handler {} maps to operation {} not declared in consumer_operation_ids",
                        contract.artifact_id, consumer_handler_id, mapped_operation_id
                    ),
                ));
            }
        }
    }
    Ok(())
}

fn validate_runtime_failure_policy_effect_scopes(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let runtime_effect_handlers_by_id: HashMap<_, _> = primitives
        .runtime_effect_handlers
        .iter()
        .map(|handler| (handler.handler_id.as_str(), handler))
        .collect();
    let runtime_effect_operations_by_id: HashMap<_, _> = primitives
        .runtime_effect_operations
        .iter()
        .map(|operation| (operation.operation_id.as_str(), operation))
        .collect();
    let runtime_effect_failure_classes: HashSet<_> = primitives
        .runtime_effect_handlers
        .iter()
        .flat_map(|handler| handler.failure_classes.iter().map(String::as_str))
        .chain(
            primitives
                .runtime_effect_operations
                .iter()
                .flat_map(|operation| {
                    operation
                        .failure_mappings
                        .iter()
                        .map(|mapping| mapping.failure_class.as_str())
                }),
        )
        .collect();

    for policy in &primitives.runtime_failure_policies {
        if !policy
            .applies_to_origins
            .iter()
            .any(|origin| origin == "runtime_effect")
        {
            continue;
        }
        for handler_id in &policy.applies_to_handler_ids {
            if !runtime_effect_handlers_by_id.contains_key(handler_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime failure policy {} references unknown runtime effect handler {}",
                        policy.policy_id, handler_id
                    ),
                ));
            }
        }
        for operation_id in &policy.applies_to_operation_ids {
            if !runtime_effect_operations_by_id.contains_key(operation_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime failure policy {} references unknown runtime effect operation {}",
                        policy.policy_id, operation_id
                    ),
                ));
            }
        }
        if !policy.applies_to_handler_ids.is_empty() && !policy.applies_to_operation_ids.is_empty()
        {
            let selected_handler_ids: HashSet<_> = policy
                .applies_to_handler_ids
                .iter()
                .map(String::as_str)
                .collect();
            let mut compatible_handler_ids: HashSet<&str> = HashSet::new();
            for operation_id in &policy.applies_to_operation_ids {
                let legacy_handler_ids = legacy_handler_ids_for_operation(operation_id, primitives);
                compatible_handler_ids.extend(legacy_handler_ids.iter().copied());
                if selected_handler_ids.is_disjoint(&legacy_handler_ids) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime failure policy {} operation {} has no selected legacy handler alias",
                            policy.policy_id, operation_id
                        ),
                    ));
                }
            }
            for handler_id in selected_handler_ids {
                if !compatible_handler_ids.contains(handler_id) {
                    return Err(invalid_primitive(
                        paths,
                        format!(
                            "runtime failure policy {} handler {} is not a legacy alias for operation ids {}",
                            policy.policy_id,
                            handler_id,
                            policy.applies_to_operation_ids.join(", ")
                        ),
                    ));
                }
            }
        }
        for failure_class in &policy.applies_to_failure_classes {
            if !runtime_effect_failure_classes.contains(failure_class.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime failure policy {} references undeclared runtime effect failure class {}",
                        policy.policy_id, failure_class
                    ),
                ));
            }
        }
        for mapping in &policy.repair_closure_mappings {
            if !runtime_effect_operations_by_id.contains_key(mapping.source_operation_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime failure policy {} repair mapping references unknown source operation {}",
                        policy.policy_id, mapping.source_operation_id
                    ),
                ));
            }
            if !runtime_effect_operations_by_id.contains_key(mapping.repair_operation_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "runtime failure policy {} repair mapping references unknown repair operation {}",
                        policy.policy_id, mapping.repair_operation_id
                    ),
                ));
            }
        }
    }

    Ok(())
}

fn validate_operation_validator_binding(
    paths: &WorkspacePaths,
    operation: &RuntimeEffectOperationDefinition,
    validator: &RuntimeEffectValidatorDefinition,
) -> CompilerAssetResult<()> {
    let known_operation_artifacts: HashSet<_> = operation
        .required_artifacts
        .iter()
        .chain(operation.produced_artifacts.iter())
        .map(String::as_str)
        .collect();
    for artifact_id in &validator.input_artifact_ids {
        if !known_operation_artifacts.contains(artifact_id.as_str()) {
            return Err(invalid_primitive(
                paths,
                format!(
                    "runtime effect operation {} binds validator {} to artifact {} not declared by the operation",
                    operation.operation_id, validator.validator_id, artifact_id
                ),
            ));
        }
    }
    let mapped_failure_classes: HashSet<_> = operation
        .failure_mappings
        .iter()
        .map(|mapping| mapping.failure_class.as_str())
        .collect();
    if !mapped_failure_classes.contains(validator.failure_class.as_str()) {
        return Err(invalid_primitive(
            paths,
            format!(
                "runtime effect operation {} binds validator {} with unmapped failure class {}",
                operation.operation_id, validator.validator_id, validator.failure_class
            ),
        ));
    }
    Ok(())
}

fn legacy_handler_ids_for_operation<'a>(
    operation_id: &str,
    primitives: &'a CompiledWorkflowPrimitiveBundle,
) -> HashSet<&'a str> {
    let mut legacy_handler_ids = HashSet::new();
    for runner in &primitives.runtime_effect_runners {
        for handler_id in &runner.legacy_handler_ids {
            if runner
                .operation_id_for_legacy_handler(handler_id)
                .as_deref()
                == Some(operation_id)
            {
                legacy_handler_ids.insert(handler_id.as_str());
            }
        }
    }
    legacy_handler_ids
}

fn validate_runtime_effect_primitive_id(
    primitive_id: &str,
    source_label: String,
    paths: &WorkspacePaths,
) -> CompilerAssetResult<()> {
    if SUPPORTED_RUNTIME_EFFECT_PRIMITIVE_IDS.contains(&primitive_id) {
        Ok(())
    } else {
        Err(invalid_primitive(
            paths,
            format!("{source_label} references unknown primitive {primitive_id}"),
        ))
    }
}

fn validate_lifecycle_and_terminal_actions(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let lifecycle_plan_ids: HashSet<_> = primitives
        .lifecycle_mutation_plans
        .iter()
        .map(|plan| plan.plan_id.as_str())
        .collect();
    let terminal_action_ids: HashSet<_> = primitives
        .terminal_actions
        .iter()
        .map(|action| action.terminal_action_id.as_str())
        .collect();
    for action in &primitives.terminal_actions {
        if !action.non_mutating {
            let Some(plan_id) = &action.lifecycle_mutation_plan_id else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "terminal action {} is mutating without a lifecycle mutation plan",
                        action.terminal_action_id
                    ),
                ));
            };
            if !lifecycle_plan_ids.contains(plan_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "terminal action references unknown lifecycle mutation plan",
                ));
            }
        }
    }
    for rule in &primitives.runtime_effect_rules {
        if let Some(plan_id) = &rule.lifecycle_mutation_plan_id {
            if !lifecycle_plan_ids.contains(plan_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "runtime effect rule references unknown lifecycle mutation plan",
                ));
            }
            if terminal_action_for_lifecycle_plan(primitives, plan_id).is_none() {
                return Err(invalid_primitive(
                    paths,
                    "runtime effect rule lifecycle plan has no terminal action mapping",
                ));
            }
        }
    }
    for action in &primitives.terminal_actions {
        for effect_rule_id in &action.effect_rule_ids {
            if !primitives
                .runtime_effect_rules
                .iter()
                .any(|rule| &rule.rule_id == effect_rule_id)
            {
                return Err(invalid_primitive(
                    paths,
                    "terminal action references unknown runtime effect rule",
                ));
            }
        }
        if !terminal_action_ids.contains(action.terminal_action_id.as_str()) {
            return Err(invalid_primitive(paths, "invalid terminal action"));
        }
    }
    Ok(())
}

fn validate_lane_policy(
    paths: &WorkspacePaths,
    lane_policy: &WorkflowPlaneSchedulerPolicyDefinition,
) -> CompilerAssetResult<()> {
    let conflict_ids: HashSet<_> = lane_policy
        .lane_conflict_policies
        .iter()
        .map(|policy| policy.policy_id.as_str())
        .collect();
    for lane in &lane_policy.lanes {
        let Some(conflict_policy_id) = &lane.conflict_policy_id else {
            return Err(invalid_primitive(
                paths,
                format!("lane {} is missing lane conflict policy", lane.lane_id),
            ));
        };
        if !conflict_ids.contains(conflict_policy_id.as_str()) {
            return Err(invalid_primitive(
                paths,
                format!(
                    "lane {} references unknown lane conflict policy {}",
                    lane.lane_id, conflict_policy_id
                ),
            ));
        }
    }
    Ok(())
}

fn validate_workspace_schema_epoch(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
) -> CompilerAssetResult<()> {
    let Some(epoch) = &primitives.workspace_schema_epoch else {
        return Err(invalid_primitive(
            paths,
            "missing workspace schema epoch authority",
        ));
    };
    if epoch.epoch_id != WORKSPACE_SCHEMA_EPOCH_ID
        || epoch.minimum_supported_epoch_id != WORKSPACE_SCHEMA_EPOCH_ID
    {
        return Err(invalid_primitive(
            paths,
            format!(
                "stale workspace schema epoch: expected {WORKSPACE_SCHEMA_EPOCH_ID}, got {}",
                epoch.epoch_id
            ),
        ));
    }
    Ok(())
}

fn validate_graph_workflow_references(
    paths: &WorkspacePaths,
    primitives: &CompiledWorkflowPrimitiveBundle,
    graph_loops: &[super::assets::ResolvedGraphLoopAsset],
) -> CompilerAssetResult<()> {
    let family_ids: HashSet<_> = primitives
        .work_item_families
        .iter()
        .map(|family| family.family_id.as_str())
        .collect();
    let profiles_by_id: HashMap<_, _> = primitives
        .request_context_profiles
        .iter()
        .map(|profile| (profile.profile_id.as_str(), profile))
        .collect();
    let render_plans_by_id: HashSet<_> = primitives
        .request_context_render_plans
        .iter()
        .map(|plan| plan.render_plan_id.as_str())
        .collect();
    let claim_policies_by_plane: HashMap<_, _> = primitives
        .queue_claim_policies
        .iter()
        .map(|policy| (policy.plane, policy))
        .collect();
    for graph_asset in graph_loops {
        let graph = &graph_asset.graph_loop;
        for entry in &graph.entry_nodes {
            if !family_ids.contains(entry.entry_key.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "graph {} entry {} references unknown work-item family",
                        graph.loop_id,
                        entry.entry_key.as_str()
                    ),
                ));
            }
        }
        let Some(claim_policy) = claim_policies_by_plane.get(&graph.plane) else {
            return Err(invalid_primitive(
                paths,
                format!(
                    "missing queue claim policy for plane {}",
                    graph.plane.as_str()
                ),
            ));
        };
        for family_id in &claim_policy.family_order {
            if !family_ids.contains(family_id.as_str()) {
                return Err(invalid_primitive(
                    paths,
                    "queue claim policy references unknown work-item family",
                ));
            }
        }
        if claim_policy.empty_behavior == "check_completion" && graph.completion_behavior.is_none()
        {
            return Err(invalid_primitive(
                paths,
                format!(
                    "graph {} is selected for completion checks but has no completion behavior",
                    graph.loop_id
                ),
            ));
        }
        for node in &graph.nodes {
            let Some(profile_id) = request_context_profile_id_for_stage_kind(
                node.stage_kind_id.as_str(),
                graph.plane,
                &primitives.request_context_profiles,
            ) else {
                continue;
            };
            let Some(profile) = profiles_by_id.get(profile_id.as_str()) else {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "graph node {} resolved unknown request context profile {}",
                        node.node_id, profile_id
                    ),
                ));
            };
            let Some(render_plan_id) = node.context_render_plan_id.as_deref() else {
                continue;
            };
            if !render_plans_by_id.contains(render_plan_id) {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "graph node {} references unknown context render plan {}",
                        node.node_id, render_plan_id
                    ),
                ));
            }
            if !profile.allow_render_plan_override
                && render_plan_id != profile.primary_render_plan_id
            {
                return Err(invalid_primitive(
                    paths,
                    format!(
                        "graph node {} render plan override {} is disallowed by profile {}",
                        node.node_id, render_plan_id, profile.profile_id
                    ),
                ));
            }
        }
        if graph.loop_id == "planning.blueprint" {
            for required in [
                "blueprint_draft",
                "manager_blueprint",
                "contractor_blueprint",
                "evaluator_blueprint",
                "mechanic_blueprint",
            ] {
                let has_entry = graph
                    .entry_nodes
                    .iter()
                    .any(|entry| entry.entry_key.as_str() == required);
                let has_node = graph
                    .nodes
                    .iter()
                    .any(|node| node.node_id == required || node.stage_kind_id == required);
                if !(has_entry || has_node) {
                    return Err(invalid_primitive(
                        paths,
                        "blueprint graph mode reference drift",
                    ));
                }
            }
        }
    }
    Ok(())
}

fn build_lane_policy(
    claim_policies: &[PlaneQueueClaimPolicyDefinition],
    graph_loops: &[super::assets::ResolvedGraphLoopAsset],
) -> CompilerAssetResult<WorkflowPlaneSchedulerPolicyDefinition> {
    let selected_planes: Vec<_> = graph_loops.iter().map(|graph| graph.plane).collect();
    let selected_lane_ids: Vec<_> = selected_planes
        .iter()
        .map(|plane| format!("{}.main", plane.as_str()))
        .collect();
    let mut claim_policies_by_plane = HashMap::new();
    let mut lanes = Vec::new();
    let mut lane_conflict_policies = Vec::new();
    let mut completion_check_order = Vec::new();

    for plane in selected_planes {
        let claim_policy = claim_policies
            .iter()
            .find(|policy| policy.plane == plane)
            .ok_or_else(|| CompilerAssetError::MissingReferencedAsset {
                asset_family: "queue_claim_policy",
                logical_id: format!("queue_claim_policy:{}", plane.as_str()),
                path: PathBuf::from(QUEUE_CLAIM_POLICIES_PATH),
            })?
            .clone();
        if claim_policy.empty_behavior == "check_completion" {
            completion_check_order.push(plane);
        }
        let lane_id = format!("{}.main", plane.as_str());
        let conflict_policy_id = format!("{}.single_writer", plane.as_str());
        lanes.push(WorkflowLaneDefinition {
            schema_version: "1.0".to_owned(),
            kind: "workflow_lane".to_owned(),
            lane_id: lane_id.clone(),
            plane,
            allowed_family_ids: claim_policy.family_order.clone(),
            claim_policy_id: claim_policy.policy_id.clone(),
            max_active_runs: 1,
            one_active_scope: "plane".to_owned(),
            partition_selector_id: None,
            mutation_lock_scope: "plane".to_owned(),
            result_application_policy: "single_writer_serialized".to_owned(),
            conflict_policy_id: Some(conflict_policy_id.clone()),
        });
        lane_conflict_policies.push(LaneConflictPolicyDefinition {
            schema_version: "1.0".to_owned(),
            kind: "lane_conflict_policy".to_owned(),
            policy_id: conflict_policy_id,
            lane_ids: vec![lane_id.clone()],
            concurrent_with_lane_ids: selected_lane_ids.clone(),
            conflict_scopes: vec!["workspace_write".to_owned()],
            lock_acquisition_order: vec![lane_id],
            release_policy: "after_result_application".to_owned(),
            missing_lock_policy: "reject_compile".to_owned(),
        });
        claim_policies_by_plane.insert(plane, claim_policy);
    }

    Ok(WorkflowPlaneSchedulerPolicyDefinition {
        schema_version: "1.0".to_owned(),
        kind: "workflow_plane_scheduler_policy".to_owned(),
        policy_id: "compiled.default".to_owned(),
        plane_order: graph_loops.iter().map(|graph| graph.plane).collect(),
        concurrency_policy_id: None,
        lanes,
        claim_policies_by_plane,
        completion_check_order,
        experimental_multi_lane: false,
        lane_conflict_policies,
    })
}

fn selected_node_ids(graph_loops: &[super::assets::ResolvedGraphLoopAsset]) -> HashSet<&str> {
    graph_loops
        .iter()
        .flat_map(|graph| graph.graph_loop.nodes.iter())
        .map(|node| node.node_id.as_str())
        .collect()
}

fn selected_request_context_profile_ids<'a>(
    graph_loops: &'a [super::assets::ResolvedGraphLoopAsset],
    profiles: &'a [RequestContextProfileDefinition],
) -> BTreeSet<&'a str> {
    let profile_ids: HashSet<_> = profiles
        .iter()
        .map(|profile| profile.profile_id.as_str())
        .collect();
    let mut selected = BTreeSet::new();
    for graph in graph_loops {
        for node in &graph.graph_loop.nodes {
            let stage_profile = format!("{}.default", node.stage_kind_id);
            if profile_ids.contains(stage_profile.as_str()) {
                selected.insert(profile_id_str(profiles, &stage_profile));
            } else if graph.plane == Plane::Learning && profile_ids.contains("learning.default") {
                selected.insert(profile_id_str(profiles, "learning.default"));
            }
        }
    }
    selected
}

fn profile_id_str<'a>(
    profiles: &'a [RequestContextProfileDefinition],
    profile_id: &str,
) -> &'a str {
    profiles
        .iter()
        .find(|profile| profile.profile_id == profile_id)
        .map(|profile| profile.profile_id.as_str())
        .expect("profile id came from loaded profiles")
}

fn lane_id_for_plane(
    lane_policy: &WorkflowPlaneSchedulerPolicyDefinition,
    plane: Plane,
) -> Option<String> {
    lane_policy
        .lanes
        .iter()
        .find(|lane| lane.plane == plane)
        .map(|lane| lane.lane_id.clone())
}

fn request_context_profile_id(
    node: &MaterializedGraphNodePlan,
    profiles: &[RequestContextProfileDefinition],
) -> Option<String> {
    request_context_profile_id_for_stage_kind(node.stage_kind_id.as_str(), node.plane, profiles)
}

fn request_context_profile_id_for_stage_kind(
    stage_kind_id: &str,
    plane: Plane,
    profiles: &[RequestContextProfileDefinition],
) -> Option<String> {
    let profile_ids: HashSet<_> = profiles
        .iter()
        .map(|profile| profile.profile_id.as_str())
        .collect();
    let stage_profile = format!("{stage_kind_id}.default");
    if profile_ids.contains(stage_profile.as_str()) {
        return Some(stage_profile);
    }
    if plane == Plane::Learning && profile_ids.contains("learning.default") {
        return Some("learning.default".to_owned());
    }
    None
}

fn context_render_plan_id(
    node: &MaterializedGraphNodePlan,
    profiles: &[RequestContextProfileDefinition],
    render_plans: &[RequestContextRenderPlan],
) -> Result<Option<String>, super::contracts::CompilerContractError> {
    let Some(profile_id) = node.request_context_profile_id.as_deref() else {
        return Ok(None);
    };
    let profile = profiles
        .iter()
        .find(|candidate| candidate.profile_id == profile_id)
        .ok_or_else(
            || super::contracts::CompilerContractError::InvalidDocument {
                message: format!(
                    "node {} resolved unknown request context profile {}",
                    node.node_id, profile_id
                ),
            },
        )?;
    let selected_render_plan_id = if let Some(render_plan_id) = node.context_render_plan_id.as_ref()
    {
        if !profile.allow_render_plan_override && render_plan_id != &profile.primary_render_plan_id
        {
            return Err(super::contracts::CompilerContractError::InvalidDocument {
                message: format!(
                    "node {} render plan override {} is disallowed by profile {}",
                    node.node_id, render_plan_id, profile.profile_id
                ),
            });
        }
        render_plan_id.clone()
    } else {
        profile.primary_render_plan_id.clone()
    };
    if !render_plans
        .iter()
        .any(|plan| plan.render_plan_id == selected_render_plan_id)
    {
        return Err(super::contracts::CompilerContractError::InvalidDocument {
            message: format!(
                "node {} references unknown request context render plan {}",
                node.node_id, selected_render_plan_id
            ),
        });
    }
    Ok(Some(selected_render_plan_id))
}

fn runtime_effect_rule_ids_for_node(
    node: &MaterializedGraphNodePlan,
    rules: &[RuntimeEffectRuleDefinition],
) -> Vec<String> {
    rules
        .iter()
        .filter(|rule| rule.source_node_id == node.node_id)
        .map(|rule| rule.rule_id.clone())
        .collect()
}

fn rules_by_node_outcome(
    rules: &[RuntimeEffectRuleDefinition],
) -> BTreeMap<(String, String), Vec<&RuntimeEffectRuleDefinition>> {
    let mut by_key: BTreeMap<(String, String), Vec<&RuntimeEffectRuleDefinition>> = BTreeMap::new();
    for rule in rules {
        for outcome in &rule.on_outcomes {
            by_key
                .entry((rule.source_node_id.clone(), outcome.clone()))
                .or_default()
                .push(rule);
        }
    }
    by_key
}

fn terminal_action_mappings_for_node(
    node: &MaterializedGraphNodePlan,
    compiled_transitions: &[super::contracts::CompiledGraphTransitionPlan],
    primitives: &CompiledWorkflowPrimitiveBundle,
    rules_by_node_outcome: &BTreeMap<(String, String), Vec<&RuntimeEffectRuleDefinition>>,
    terminal_states_by_id: &HashMap<String, String>,
) -> Result<BTreeMap<String, String>, super::contracts::CompilerContractError> {
    let mut mappings = BTreeMap::new();
    for ((source_node_id, outcome), rules) in rules_by_node_outcome {
        if source_node_id != &node.node_id {
            continue;
        }
        for rule in rules {
            if let Some(plan_id) = &rule.lifecycle_mutation_plan_id {
                let action_id = terminal_action_for_lifecycle_plan(primitives, plan_id)
                    .ok_or_else(|| super::contracts::CompilerContractError::InvalidDocument {
                        message: format!(
                            "runtime effect rule {} has no terminal action for lifecycle plan {plan_id}",
                            rule.rule_id
                        ),
                    })?;
                mappings.insert(outcome.clone(), action_id);
            }
        }
    }

    for transition in compiled_transitions {
        if transition.source_node_id != node.node_id {
            continue;
        }
        if mappings.contains_key(&transition.outcome) {
            continue;
        }
        let Some(terminal_state_id) = &transition.terminal_state_id else {
            continue;
        };
        let Some(terminal_class) = terminal_states_by_id.get(terminal_state_id) else {
            continue;
        };
        let action_id = terminal_action_for_class(primitives, terminal_class)
            .ok_or_else(|| super::contracts::CompilerContractError::InvalidDocument {
                message: format!(
                    "terminal state {terminal_state_id} has no terminal action for class {terminal_class}"
                ),
            })?;
        mappings.insert(transition.outcome.clone(), action_id);
    }
    Ok(mappings)
}

fn terminal_action_for_lifecycle_plan(
    primitives: &CompiledWorkflowPrimitiveBundle,
    plan_id: &str,
) -> Option<String> {
    primitives
        .terminal_actions
        .iter()
        .find(|action| action.lifecycle_mutation_plan_id.as_deref() == Some(plan_id))
        .map(|action| action.terminal_action_id.clone())
}

fn terminal_action_for_class(
    primitives: &CompiledWorkflowPrimitiveBundle,
    terminal_class: &str,
) -> Option<String> {
    primitives
        .terminal_actions
        .iter()
        .find(|action| action.terminal_class == terminal_class)
        .map(|action| action.terminal_action_id.clone())
}

fn load_collection<T>(
    paths: &WorkspacePaths,
    relative_path: &str,
    collection_id: &str,
    loaded: &mut LoadedWorkflowPrimitives,
) -> CompilerAssetResult<Vec<T>>
where
    T: RuntimeJsonContract + DeserializeOwned,
{
    let value = load_json_value(paths, relative_path, collection_id, loaded)?;
    let definitions = value
        .get("definitions")
        .and_then(Value::as_array)
        .ok_or_else(|| {
            invalid_primitive(paths, format!("{relative_path} must contain definitions"))
        })?;
    definitions
        .iter()
        .cloned()
        .map(|value| {
            T::from_json_value(value).map_err(|error| CompilerAssetError::Contract {
                path: paths.runtime_root.join(relative_path),
                artifact: T::ARTIFACT,
                message: error.to_string(),
            })
        })
        .collect()
}

fn load_single<T>(
    paths: &WorkspacePaths,
    relative_path: &str,
    collection_id: &str,
    loaded: &mut LoadedWorkflowPrimitives,
) -> CompilerAssetResult<Vec<T>>
where
    T: RuntimeJsonContract + DeserializeOwned,
{
    Ok(vec![load_single_definition(
        paths,
        relative_path,
        collection_id,
        loaded,
    )?])
}

fn load_single_definition<T>(
    paths: &WorkspacePaths,
    relative_path: &str,
    collection_id: &str,
    loaded: &mut LoadedWorkflowPrimitives,
) -> CompilerAssetResult<T>
where
    T: RuntimeJsonContract + DeserializeOwned,
{
    let value = load_json_value(paths, relative_path, collection_id, loaded)?;
    T::from_json_value(value).map_err(|error| CompilerAssetError::Contract {
        path: paths.runtime_root.join(relative_path),
        artifact: T::ARTIFACT,
        message: error.to_string(),
    })
}

fn load_json_value(
    paths: &WorkspacePaths,
    relative_path: &str,
    collection_id: &str,
    loaded: &mut LoadedWorkflowPrimitives,
) -> CompilerAssetResult<Value> {
    let path = paths.runtime_root.join(relative_path);
    let bytes = fs::read(&path).map_err(|error| io_error(&path, error))?;
    let content_sha256 = sha256_hex(&bytes);
    loaded.resolved_assets.push(ResolvedAssetRef {
        asset_family: "workflow_primitive".to_owned(),
        logical_id: format!(
            "workflow_primitive:{collection_id}:{}",
            primitive_asset_id(relative_path)
        ),
        compile_time_path: relative_path.to_owned(),
        content_sha256: content_sha256.clone(),
    });
    loaded.fingerprints.insert(
        format!("{collection_id}.{}", primitive_asset_id(relative_path)),
        content_sha256.clone(),
    );
    let collection_fingerprint = loaded
        .fingerprints
        .get(collection_id)
        .map(|previous| {
            sha256_hex(format!("{previous}\n{relative_path}\n{content_sha256}").as_bytes())
        })
        .unwrap_or_else(|| content_sha256.clone());
    loaded
        .fingerprints
        .insert(collection_id.to_owned(), collection_fingerprint);
    serde_json::from_slice(&bytes).map_err(|error| CompilerAssetError::Contract {
        path,
        artifact: "workflow_primitive",
        message: error.to_string(),
    })
}

fn primitive_asset_id(relative_path: &str) -> String {
    Path::new(relative_path)
        .file_stem()
        .and_then(|stem| stem.to_str())
        .unwrap_or(relative_path)
        .replace('-', "_")
}

fn invalid_primitive(paths: &WorkspacePaths, message: impl Into<String>) -> CompilerAssetError {
    CompilerAssetError::InvalidReferencedAsset {
        asset_family: "workflow_primitive",
        logical_id: "workflow_primitives".to_owned(),
        path: paths.runtime_root.join("registry"),
        message: message.into(),
    }
}

fn io_error(path: &Path, error: io::Error) -> CompilerAssetError {
    CompilerAssetError::Io {
        path: path.to_path_buf(),
        message: error.to_string(),
    }
}

fn sha256_hex(contents: &[u8]) -> String {
    let digest = Sha256::digest(contents);
    let mut rendered = String::with_capacity(digest.len() * 2);
    for byte in digest {
        rendered.push_str(&format!("{byte:02x}"));
    }
    rendered
}
