xlog_prob/

epistemic_production.rs

//! Production GPU exact-path adapter for accepted epistemic evidence.
//!
//! This module is intentionally thin. It gates probabilistic execution on
//! accepted world-view evidence, then routes into the existing GPU-native exact
//! provenance path instead of using the bounded epistemic fixture circuit.

#[cfg(feature = "host-io")]
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::sync::Arc;
#[cfg(feature = "host-io")]
use std::sync::{Mutex, MutexGuard};

use xlog_core::{symbol, Result, XlogError};
use xlog_cuda::CudaKernelProvider;
use xlog_ir::EirEpistemicMode;
#[cfg(feature = "host-io")]
use xlog_logic::ast::ProbEngine;
use xlog_logic::ast::{Atom, Evidence, Program, Term};
use xlog_logic::parse_program;
use xlog_runtime::{EpistemicGpuBatchExecutionResult, EpistemicGpuExecutionResult};

use crate::compilation::{encode_cnf_gpu, GpuPirGraph, GpuPirRoots};
#[cfg(feature = "host-io")]
use crate::epistemic::EpistemicAssumptionKind;
use crate::epistemic::EpistemicEvidenceTerm;
use crate::epistemic::{
    AcceptedWorldViewEvidence, CircuitUpdate, CircuitUpdateMode, EpistemicAssumption,
    EpistemicCircuit,
};
#[cfg(feature = "host-io")]
use crate::exact::ExactProgramOrigin;
#[cfg(feature = "host-io")]
use crate::exact::{ExactCircuitWitness, ExactResult, ExactResultWithGrads, ProbVarInfo};
use crate::exact::{ExactDdnnfProgram, GpuConfig};
use crate::pir::{PirNode, PirNodeId};
#[cfg(feature = "host-io")]
use crate::provenance::AggregateLiftStatus;
use crate::provenance::Value;
use crate::provenance::{extract_from_program, Provenance};

macro_rules! epistemic_prob_trace_transaction {
    ($adapter:ident, $body:block) => {{
        let trace_before = $adapter.trace;
        let result: Result<_> = (|| $body)();
        match result {
            Ok(value) => Ok(value),
            Err(err) => {
                $adapter.trace = trace_before;
                Err(err)
            }
        }
    }};
}

macro_rules! checked_prob_trace_counter_inc {
    ($adapter:ident, $field:ident) => {{
        $adapter.trace.$field = EpistemicProbProductionAdapter::checked_trace_counter_add(
            $adapter.trace.$field,
            1,
            stringify!($field),
        )?;
    }};
}

/// Production capability status for probabilistic adapter paths.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EpistemicProbProductionCapabilityStatus {
    /// Existing GPU-native production path is available.
    Available,
    /// Required GPU-native production path is not implemented.
    Blocked,
}

/// Capability report for the probabilistic production adapter.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EpistemicProbProductionCapabilities {
    /// Exact/provenance compilation through `ExactDdnnfProgram`.
    pub gpu_exact_provenance: EpistemicProbProductionCapabilityStatus,
    /// GPU PIR upload and CNF encoding path.
    pub gpu_pir_cnf: EpistemicProbProductionCapabilityStatus,
    /// Bounded compile-plus-evaluate knowledge-compilation path.
    pub gpu_knowledge_compilation: EpistemicProbProductionCapabilityStatus,
    /// GPU query and gradient evaluation path.
    pub gpu_exact_query_and_gradient: EpistemicProbProductionCapabilityStatus,
    /// Whether the bounded fixture circuit may satisfy production metrics.
    pub fixture_circuit_allowed: bool,
    /// Blocker reason for knowledge-compilation production coverage, or empty when available.
    pub gpu_knowledge_compilation_blocker: &'static str,
}

/// Return a static inventory of implemented probabilistic backend stages.
///
/// This does not probe CUDA availability, reflect Cargo feature selection, or
/// assert that every epistemic execution route produces a compatible evidence
/// record.
pub fn production_capabilities() -> EpistemicProbProductionCapabilities {
    EpistemicProbProductionCapabilities {
        gpu_exact_provenance: EpistemicProbProductionCapabilityStatus::Available,
        gpu_pir_cnf: EpistemicProbProductionCapabilityStatus::Available,
        gpu_knowledge_compilation: EpistemicProbProductionCapabilityStatus::Available,
        gpu_exact_query_and_gradient: EpistemicProbProductionCapabilityStatus::Available,
        fixture_circuit_allowed: false,
        gpu_knowledge_compilation_blocker: "",
    }
}

/// Trace counters proving the production adapter stayed on the GPU exact path.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct EpistemicProbProductionTrace {
    /// Number of source compiles routed through `ExactDdnnfProgram`.
    pub gpu_exact_source_compiles: u64,
    /// Number of parsed-program compiles routed through `ExactDdnnfProgram`.
    pub gpu_exact_program_compiles: u64,
    /// Number of evaluations routed through an already compiled conditioned circuit.
    pub gpu_conditioned_circuit_reuses: u64,
    /// Actual GPU circuit-compiler invocations performed while preparing this handle.
    pub gpu_conditioned_circuit_preparation_compiles: u64,
    /// Successful exact-circuit materializations performed while preparing this handle.
    pub gpu_conditioned_circuit_materializations: u64,
    /// Verified disk-cache restorations used while preparing this handle.
    pub gpu_conditioned_circuit_disk_cache_restores: u64,
    /// GPU circuit-cache hits used while preparing this handle.
    pub gpu_conditioned_circuit_gpu_cache_hits: u64,
    /// Process-local immutable generation of the exact circuit used by this evaluation.
    pub gpu_conditioned_circuit_generation: u64,
    /// Device-cache slot held by the exact circuit used by this evaluation.
    pub gpu_conditioned_circuit_cache_slot: u64,
    /// Number of accepted world-view evidence objects consumed as a gate.
    pub accepted_world_view_evidence_consumed: u64,
    /// Number of accepted Gelfond-1991 compatibility-mode GPU world-view
    /// evidence objects consumed as a gate.
    pub accepted_g91_world_view_evidence_consumed: u64,
    /// Number of accepted FAEEL GPU world-view evidence objects consumed as a gate.
    pub accepted_faeel_world_view_evidence_consumed: u64,
    /// Number of accepted epistemic assumptions consumed from world-view evidence.
    pub accepted_evidence_assumptions_consumed: u64,
    /// Number of accepted nonzero-arity epistemic assumptions consumed from GPU evidence.
    pub accepted_gpu_nonzero_arity_evidence_assumptions_consumed: u64,
    /// Maximum accepted GPU evidence tuple arity consumed by this adapter.
    pub accepted_gpu_max_evidence_arity_consumed: u32,
    /// GPU tuple-key column reads consumed from accepted world-view evidence.
    pub accepted_gpu_tuple_key_column_reads_consumed: u64,
    /// GPU final-tuple row filters consumed from accepted world-view evidence.
    pub accepted_gpu_final_tuple_row_filters_consumed: u64,
    /// Negated GPU final-tuple row filters consumed from accepted world-view evidence.
    pub accepted_gpu_final_tuple_negated_row_filters_consumed: u64,
    /// Row-specific GPU model-slot capacity consumed from accepted world-view evidence.
    pub accepted_gpu_row_specific_membership_row_capacity_consumed: u64,
    /// Fallback GPU row-filter capacity consumed outside bounded model-slot windows.
    pub accepted_gpu_row_filter_fallback_row_capacity_consumed: u64,
    /// Reduced integrity-constraint relations checked by accepted GPU evidence.
    pub accepted_gpu_constraint_relations_checked_consumed: u64,
    /// Constraint row-count metadata reads consumed from accepted GPU evidence.
    pub accepted_gpu_constraint_row_count_device_reads_consumed: u64,
    /// Number of accepted GPU batch evidence records consumed as a gate.
    pub accepted_gpu_batch_evidence_consumed: u64,
    /// Number of accepted GPU batch components consumed as individual evidence records.
    pub accepted_gpu_batch_component_evidence_consumed: u64,
    /// Number of accepted evidence applications that updated caller-owned incremental circuits.
    ///
    /// This is fixture coverage only and is intentionally excluded from production path events.
    pub accepted_incremental_circuit_updates: u64,
    /// Number of GPU exact query evaluations routed through `ExactDdnnfProgram`.
    pub gpu_exact_query_evaluations: u64,
    /// Number of source GPU exact query evaluations routed through `ExactDdnnfProgram`.
    pub gpu_source_exact_query_evaluations: u64,
    /// Number of parsed-program GPU exact query evaluations routed through `ExactDdnnfProgram`.
    pub gpu_program_exact_query_evaluations: u64,
    /// Number of GPU gradient evaluations routed through `ExactDdnnfProgram`.
    pub gpu_exact_gradient_evaluations: u64,
    /// Number of source GPU gradient evaluations routed through `ExactDdnnfProgram`.
    pub gpu_source_exact_gradient_evaluations: u64,
    /// Number of parsed-program GPU gradient evaluations routed through `ExactDdnnfProgram`.
    pub gpu_program_exact_gradient_evaluations: u64,
    /// Number of source-conditioned GPU gradient evaluations routed through `ExactDdnnfProgram`.
    pub gpu_source_conditioned_gradient_evaluations: u64,
    /// Number of parsed-program-conditioned GPU gradient evaluations routed through `ExactDdnnfProgram`.
    pub gpu_program_conditioned_gradient_evaluations: u64,
    /// Number of accepted PIR graphs uploaded through the existing GPU PIR layout.
    pub gpu_pir_graph_uploads: u64,
    /// Number of source accepted PIR graphs uploaded through the existing GPU PIR layout.
    pub gpu_source_pir_graph_uploads: u64,
    /// Number of parsed-program accepted PIR graphs uploaded through the existing GPU PIR layout.
    pub gpu_program_pir_graph_uploads: u64,
    /// Number of accepted PIR root sets encoded through the existing GPU CNF encoder.
    pub gpu_cnf_encodes: u64,
    /// Number of source accepted PIR root sets encoded through the existing GPU CNF encoder.
    pub gpu_source_cnf_encodes: u64,
    /// Number of parsed-program accepted PIR root sets encoded through the existing GPU CNF encoder.
    pub gpu_program_cnf_encodes: u64,
    /// Number of accepted compile-and-evaluate runs through the GPU exact path.
    pub gpu_knowledge_compilation_end_to_end_runs: u64,
    /// GPU exact/provenance/PIR/CNF/knowledge-compilation events that occurred inside accepted evidence gates.
    pub accepted_gpu_production_path_events: u64,
    /// Number of accepted source compile-and-evaluate runs through the GPU exact path.
    pub gpu_source_knowledge_compilation_end_to_end_runs: u64,
    /// Number of accepted parsed-program compile-and-evaluate runs through the GPU exact path.
    pub gpu_program_knowledge_compilation_end_to_end_runs: u64,
    /// Number of accepted assumptions compiled as exact evidence facts.
    pub gpu_conditioned_evidence_facts: u64,
    /// Number of accepted world-view evidence objects compiled into conditioned exact evidence.
    pub accepted_conditioned_world_view_evidence_consumed: u64,
    /// Number of source-conditioned accepted world-view evidence objects compiled as exact evidence.
    pub accepted_source_conditioned_world_view_evidence_consumed: u64,
    /// Number of parsed-program-conditioned accepted world-view evidence objects compiled as exact evidence.
    pub accepted_program_conditioned_world_view_evidence_consumed: u64,
    /// Number of accepted nonzero-arity assumptions compiled as exact evidence facts.
    pub gpu_conditioned_nonzero_arity_evidence_facts: u64,
    /// Maximum accepted exact evidence tuple arity observed across conditioned paths.
    pub gpu_conditioned_max_evidence_arity: u32,
    /// Number of false accepted assumptions compiled as exact evidence facts.
    pub gpu_conditioned_negative_evidence_facts: u64,
    /// Number of source-conditioned accepted assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_evidence_facts: u64,
    /// Number of source-conditioned nonzero-arity assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_nonzero_arity_evidence_facts: u64,
    /// Maximum source-conditioned accepted exact evidence tuple arity observed.
    pub gpu_source_conditioned_max_evidence_arity: u32,
    /// Number of parsed-program-conditioned accepted assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_evidence_facts: u64,
    /// Number of parsed-program-conditioned nonzero-arity assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_nonzero_arity_evidence_facts: u64,
    /// Maximum parsed-program-conditioned accepted exact evidence tuple arity observed.
    pub gpu_program_conditioned_max_evidence_arity: u32,
    /// Number of false source-conditioned assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_negative_evidence_facts: u64,
    /// Number of false parsed-program-conditioned assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_negative_evidence_facts: u64,
    /// Number of true `know` assumptions compiled as exact evidence facts.
    pub gpu_conditioned_know_evidence_facts: u64,
    /// Number of true `possible` assumptions compiled as exact evidence facts.
    pub gpu_conditioned_possible_evidence_facts: u64,
    /// Number of false `know` assumptions compiled as exact evidence facts.
    pub gpu_conditioned_not_known_evidence_facts: u64,
    /// Number of false `possible` assumptions compiled as exact evidence facts.
    pub gpu_conditioned_not_possible_evidence_facts: u64,
    /// Number of source-conditioned true `know` assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_know_evidence_facts: u64,
    /// Number of source-conditioned true `possible` assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_possible_evidence_facts: u64,
    /// Number of source-conditioned false `know` assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_not_known_evidence_facts: u64,
    /// Number of source-conditioned false `possible` assumptions compiled as exact evidence facts.
    pub gpu_source_conditioned_not_possible_evidence_facts: u64,
    /// Number of parsed-program-conditioned true `know` assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_know_evidence_facts: u64,
    /// Number of parsed-program-conditioned true `possible` assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_possible_evidence_facts: u64,
    /// Number of parsed-program-conditioned false `know` assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_not_known_evidence_facts: u64,
    /// Number of parsed-program-conditioned false `possible` assumptions compiled as exact evidence facts.
    pub gpu_program_conditioned_not_possible_evidence_facts: u64,
}

impl EpistemicProbProductionTrace {
    fn checked_gpu_production_path_events(&self) -> Result<u64> {
        Self::checked_production_event_sum(
            "gpu_production_path_events",
            &[
                self.gpu_exact_source_compiles,
                self.gpu_exact_program_compiles,
                self.gpu_conditioned_circuit_reuses,
                self.gpu_exact_query_evaluations,
                self.gpu_source_exact_query_evaluations,
                self.gpu_program_exact_query_evaluations,
                self.gpu_exact_gradient_evaluations,
                self.gpu_source_exact_gradient_evaluations,
                self.gpu_program_exact_gradient_evaluations,
                self.gpu_source_conditioned_gradient_evaluations,
                self.gpu_program_conditioned_gradient_evaluations,
                self.gpu_pir_graph_uploads,
                self.gpu_source_pir_graph_uploads,
                self.gpu_program_pir_graph_uploads,
                self.gpu_cnf_encodes,
                self.gpu_source_cnf_encodes,
                self.gpu_program_cnf_encodes,
                self.gpu_knowledge_compilation_end_to_end_runs,
                self.gpu_source_knowledge_compilation_end_to_end_runs,
                self.gpu_program_knowledge_compilation_end_to_end_runs,
            ],
        )
    }

    fn checked_production_event_sum(counter: &str, values: &[u64]) -> Result<u64> {
        values.iter().try_fold(0u64, |acc, value| {
            acc.checked_add(*value)
                .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic production trace accounting".to_string(),
                    context: format!(
                        "GPU probability production counter {counter} overflowed while adding \
                         {value} to {acc}"
                    ),
                })
        })
    }

    fn require_pir_cnf_accounting_pair(
        construct: &'static str,
        pir_graph_uploads: u64,
        cnf_encodes: u64,
        path: &'static str,
    ) -> Result<()> {
        if pir_graph_uploads != cnf_encodes {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: construct.to_string(),
                context: format!(
                    "PIR/CNF production accounting must match for {path} path, got \
                     pir_graph_uploads={} cnf_encodes={}",
                    pir_graph_uploads, cnf_encodes
                ),
            });
        }
        Ok(())
    }

    fn require_pir_cnf_accounting(&self) -> Result<()> {
        let construct = "epistemic probabilistic production metric gate";
        Self::require_pir_cnf_accounting_pair(
            construct,
            self.gpu_pir_graph_uploads,
            self.gpu_cnf_encodes,
            "aggregate",
        )?;
        Self::require_pir_cnf_accounting_pair(
            construct,
            self.gpu_source_pir_graph_uploads,
            self.gpu_source_cnf_encodes,
            "source",
        )?;
        Self::require_pir_cnf_accounting_pair(
            construct,
            self.gpu_program_pir_graph_uploads,
            self.gpu_program_cnf_encodes,
            "program",
        )
    }

    /// Require internally consistent GPU tuple-membership evidence counters.
    pub fn require_accepted_gpu_tuple_evidence_trace(&self) -> Result<()> {
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
            > self.accepted_evidence_assumptions_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted nonzero-arity evidence assumptions cannot exceed accepted \
                     evidence assumptions: nonzero={} total={}",
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed,
                    self.accepted_evidence_assumptions_consumed
                ),
            });
        }
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed == 0
            && self.accepted_gpu_max_evidence_arity_consumed > 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted max evidence arity {} requires at least one accepted \
                     nonzero-arity GPU evidence assumption",
                    self.accepted_gpu_max_evidence_arity_consumed
                ),
            });
        }
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed > 0
            && self.accepted_gpu_max_evidence_arity_consumed == 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted nonzero-arity GPU evidence requires accepted max evidence arity, \
                     got nonzero_assumptions={} max_arity=0",
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
                ),
            });
        }
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed == 0
            && self.accepted_gpu_tuple_key_column_reads_consumed != 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted tuple-key reads require accepted nonzero-arity GPU evidence, got \
                     nonzero_assumptions=0 tuple_key_reads={}",
                    self.accepted_gpu_tuple_key_column_reads_consumed
                ),
            });
        }
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed > 0
            && self.accepted_gpu_tuple_key_column_reads_consumed == 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted nonzero-arity GPU evidence requires tuple-key device column reads, \
                     got nonzero_assumptions={} tuple_key_reads=0",
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
                ),
            });
        }
        if self.accepted_gpu_final_tuple_negated_row_filters_consumed
            > self.accepted_gpu_final_tuple_row_filters_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted negated final-tuple row filters cannot exceed total row filters: \
                     negated={} total={}",
                    self.accepted_gpu_final_tuple_negated_row_filters_consumed,
                    self.accepted_gpu_final_tuple_row_filters_consumed
                ),
            });
        }
        if self.accepted_gpu_final_tuple_row_filters_consumed == 0
            && (self.accepted_gpu_row_specific_membership_row_capacity_consumed != 0
                || self.accepted_gpu_row_filter_fallback_row_capacity_consumed != 0)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted row-specific/fallback tuple capacity requires accepted GPU row \
                     filters, got row_filters=0 row_specific_capacity={} fallback_capacity={}",
                    self.accepted_gpu_row_specific_membership_row_capacity_consumed,
                    self.accepted_gpu_row_filter_fallback_row_capacity_consumed
                ),
            });
        }
        if self.accepted_gpu_final_tuple_row_filters_consumed > 0
            && self.accepted_gpu_row_specific_membership_row_capacity_consumed == 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU final-tuple row filters require row-specific model-slot \
                     capacity, got row_filters={} row_specific_capacity=0",
                    self.accepted_gpu_final_tuple_row_filters_consumed
                ),
            });
        }
        if self.accepted_gpu_constraint_row_count_device_reads_consumed
            > self.accepted_gpu_constraint_relations_checked_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted constraint row-count device reads cannot exceed checked reduced \
                     constraint relations, got reads={} checked={}",
                    self.accepted_gpu_constraint_row_count_device_reads_consumed,
                    self.accepted_gpu_constraint_relations_checked_consumed
                ),
            });
        }
        Ok(())
    }

    /// Require internally consistent accepted GPU world-view evidence counters.
    pub fn require_accepted_gpu_world_view_evidence_trace(&self) -> Result<()> {
        let mode_count = self
            .accepted_g91_world_view_evidence_consumed
            .checked_add(self.accepted_faeel_world_view_evidence_consumed)
            .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "accepted GPU world-view mode counters overflowed".to_string(),
            })?;
        if self.accepted_world_view_evidence_consumed != 0
            && mode_count != self.accepted_world_view_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU world-view evidence must be classified by epistemic mode, got \
                     evidence={} gelfond_1991={} faeel={}",
                    self.accepted_world_view_evidence_consumed,
                    self.accepted_g91_world_view_evidence_consumed,
                    self.accepted_faeel_world_view_evidence_consumed
                ),
            });
        }
        if self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
            > self.accepted_evidence_assumptions_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted nonzero-arity GPU evidence assumptions cannot exceed accepted \
                     assumptions, got nonzero={} assumptions={}",
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed,
                    self.accepted_evidence_assumptions_consumed
                ),
            });
        }
        if self.accepted_world_view_evidence_consumed != 0
            && self.accepted_evidence_assumptions_consumed
                < self.accepted_world_view_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU world-view evidence requires at least one accepted epistemic \
                     assumption per evidence record, got evidence={} assumptions={}",
                    self.accepted_world_view_evidence_consumed,
                    self.accepted_evidence_assumptions_consumed
                ),
            });
        }
        if self.accepted_gpu_batch_component_evidence_consumed
            < self.accepted_gpu_batch_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU batch component evidence must cover accepted batch evidence, \
                     got batches={} components={}",
                    self.accepted_gpu_batch_evidence_consumed,
                    self.accepted_gpu_batch_component_evidence_consumed
                ),
            });
        }
        if self.accepted_gpu_batch_component_evidence_consumed
            > self.accepted_world_view_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU batch component evidence cannot exceed accepted world-view \
                     evidence, got components={} evidence={}",
                    self.accepted_gpu_batch_component_evidence_consumed,
                    self.accepted_world_view_evidence_consumed
                ),
            });
        }
        if self.accepted_conditioned_world_view_evidence_consumed
            > self.accepted_world_view_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted conditioned world-view evidence cannot exceed accepted world-view \
                     evidence, got conditioned={} evidence={}",
                    self.accepted_conditioned_world_view_evidence_consumed,
                    self.accepted_world_view_evidence_consumed
                ),
            });
        }
        Ok(())
    }

    fn require_conditioned_counter_sum(
        counter: &'static str,
        aggregate: u64,
        source: u64,
        program: u64,
    ) -> Result<()> {
        let expected = source.checked_add(program).ok_or_else(|| {
            XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned evidence counter {counter} overflowed while adding source={} \
                     program={}",
                    source, program
                ),
            }
        })?;
        if aggregate != expected {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned evidence counter {counter} must equal source+program, got \
                     aggregate={} source={} program={}",
                    aggregate, source, program
                ),
            });
        }
        Ok(())
    }

    fn require_gpu_path_counter_sum(
        counter: &'static str,
        aggregate: u64,
        source: u64,
        program: u64,
    ) -> Result<()> {
        let expected = source.checked_add(program).ok_or_else(|| {
            XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "GPU production path counter {counter} overflowed while adding source={} \
                     program={}",
                    source, program
                ),
            }
        })?;
        if aggregate != expected {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "GPU production path accounting must match source+program for {counter}, \
                     got aggregate={} source={} program={}",
                    aggregate, source, program
                ),
            });
        }
        Ok(())
    }

    fn require_gpu_path_accounting(&self) -> Result<()> {
        Self::require_gpu_path_counter_sum(
            "exact_query_evaluations",
            self.gpu_exact_query_evaluations,
            self.gpu_source_exact_query_evaluations,
            self.gpu_program_exact_query_evaluations,
        )?;
        Self::require_gpu_path_counter_sum(
            "exact_gradient_evaluations",
            self.gpu_exact_gradient_evaluations,
            self.gpu_source_exact_gradient_evaluations,
            self.gpu_program_exact_gradient_evaluations,
        )?;
        Self::require_gpu_path_counter_sum(
            "pir_graph_uploads",
            self.gpu_pir_graph_uploads,
            self.gpu_source_pir_graph_uploads,
            self.gpu_program_pir_graph_uploads,
        )?;
        Self::require_gpu_path_counter_sum(
            "cnf_encodes",
            self.gpu_cnf_encodes,
            self.gpu_source_cnf_encodes,
            self.gpu_program_cnf_encodes,
        )?;
        Self::require_gpu_path_counter_sum(
            "knowledge_compilation_end_to_end_runs",
            self.gpu_knowledge_compilation_end_to_end_runs,
            self.gpu_source_knowledge_compilation_end_to_end_runs,
            self.gpu_program_knowledge_compilation_end_to_end_runs,
        )
    }

    /// Require internally consistent conditioned exact-evidence counters.
    pub fn require_conditioned_evidence_trace(&self) -> Result<()> {
        Self::require_conditioned_counter_sum(
            "accepted_world_view_evidence",
            self.accepted_conditioned_world_view_evidence_consumed,
            self.accepted_source_conditioned_world_view_evidence_consumed,
            self.accepted_program_conditioned_world_view_evidence_consumed,
        )?;
        Self::require_conditioned_counter_sum(
            "evidence_facts",
            self.gpu_conditioned_evidence_facts,
            self.gpu_source_conditioned_evidence_facts,
            self.gpu_program_conditioned_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "nonzero_arity_evidence_facts",
            self.gpu_conditioned_nonzero_arity_evidence_facts,
            self.gpu_source_conditioned_nonzero_arity_evidence_facts,
            self.gpu_program_conditioned_nonzero_arity_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "negative_evidence_facts",
            self.gpu_conditioned_negative_evidence_facts,
            self.gpu_source_conditioned_negative_evidence_facts,
            self.gpu_program_conditioned_negative_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "know_evidence_facts",
            self.gpu_conditioned_know_evidence_facts,
            self.gpu_source_conditioned_know_evidence_facts,
            self.gpu_program_conditioned_know_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "possible_evidence_facts",
            self.gpu_conditioned_possible_evidence_facts,
            self.gpu_source_conditioned_possible_evidence_facts,
            self.gpu_program_conditioned_possible_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "not_known_evidence_facts",
            self.gpu_conditioned_not_known_evidence_facts,
            self.gpu_source_conditioned_not_known_evidence_facts,
            self.gpu_program_conditioned_not_known_evidence_facts,
        )?;
        Self::require_conditioned_counter_sum(
            "not_possible_evidence_facts",
            self.gpu_conditioned_not_possible_evidence_facts,
            self.gpu_source_conditioned_not_possible_evidence_facts,
            self.gpu_program_conditioned_not_possible_evidence_facts,
        )?;

        if (self.gpu_conditioned_evidence_facts != 0
            && self.accepted_conditioned_world_view_evidence_consumed == 0)
            || (self.gpu_source_conditioned_evidence_facts != 0
                && self.accepted_source_conditioned_world_view_evidence_consumed == 0)
            || (self.gpu_program_conditioned_evidence_facts != 0
                && self.accepted_program_conditioned_world_view_evidence_consumed == 0)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned exact evidence facts require accepted conditioned world-view \
                     evidence, got facts={} evidence={} source_facts={} source_evidence={} \
                     program_facts={} program_evidence={}",
                    self.gpu_conditioned_evidence_facts,
                    self.accepted_conditioned_world_view_evidence_consumed,
                    self.gpu_source_conditioned_evidence_facts,
                    self.accepted_source_conditioned_world_view_evidence_consumed,
                    self.gpu_program_conditioned_evidence_facts,
                    self.accepted_program_conditioned_world_view_evidence_consumed
                ),
            });
        }

        if self.gpu_conditioned_evidence_facts
            < self.accepted_conditioned_world_view_evidence_consumed
            || self.gpu_source_conditioned_evidence_facts
                < self.accepted_source_conditioned_world_view_evidence_consumed
            || self.gpu_program_conditioned_evidence_facts
                < self.accepted_program_conditioned_world_view_evidence_consumed
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned exact evidence facts must cover each accepted conditioned \
                     world-view evidence record, got facts={} evidence={} source_facts={} \
                     source_evidence={} program_facts={} program_evidence={}",
                    self.gpu_conditioned_evidence_facts,
                    self.accepted_conditioned_world_view_evidence_consumed,
                    self.gpu_source_conditioned_evidence_facts,
                    self.accepted_source_conditioned_world_view_evidence_consumed,
                    self.gpu_program_conditioned_evidence_facts,
                    self.accepted_program_conditioned_world_view_evidence_consumed
                ),
            });
        }

        if self.gpu_conditioned_nonzero_arity_evidence_facts > self.gpu_conditioned_evidence_facts
            || self.gpu_source_conditioned_nonzero_arity_evidence_facts
                > self.gpu_source_conditioned_evidence_facts
            || self.gpu_program_conditioned_nonzero_arity_evidence_facts
                > self.gpu_program_conditioned_evidence_facts
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned nonzero-arity facts cannot exceed conditioned evidence facts: \
                     nonzero={} total={} source_nonzero={} source_total={} program_nonzero={} \
                     program_total={}",
                    self.gpu_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_conditioned_evidence_facts,
                    self.gpu_source_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_source_conditioned_evidence_facts,
                    self.gpu_program_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_program_conditioned_evidence_facts
                ),
            });
        }
        if self.gpu_conditioned_negative_evidence_facts > self.gpu_conditioned_evidence_facts
            || self.gpu_source_conditioned_negative_evidence_facts
                > self.gpu_source_conditioned_evidence_facts
            || self.gpu_program_conditioned_negative_evidence_facts
                > self.gpu_program_conditioned_evidence_facts
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned negative facts cannot exceed conditioned evidence facts: \
                     negative={} total={} source_negative={} source_total={} program_negative={} \
                     program_total={}",
                    self.gpu_conditioned_negative_evidence_facts,
                    self.gpu_conditioned_evidence_facts,
                    self.gpu_source_conditioned_negative_evidence_facts,
                    self.gpu_source_conditioned_evidence_facts,
                    self.gpu_program_conditioned_negative_evidence_facts,
                    self.gpu_program_conditioned_evidence_facts
                ),
            });
        }

        let operator_fact_count = self
            .gpu_conditioned_know_evidence_facts
            .checked_add(self.gpu_conditioned_possible_evidence_facts)
            .and_then(|sum| sum.checked_add(self.gpu_conditioned_not_known_evidence_facts))
            .and_then(|sum| sum.checked_add(self.gpu_conditioned_not_possible_evidence_facts))
            .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: "conditioned operator evidence fact counters overflowed".to_string(),
            })?;
        if operator_fact_count != self.gpu_conditioned_evidence_facts {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned operator evidence facts must equal total evidence facts, got \
                     operators={} total={}",
                    operator_fact_count, self.gpu_conditioned_evidence_facts
                ),
            });
        }

        let source_operator_fact_count = self
            .gpu_source_conditioned_know_evidence_facts
            .checked_add(self.gpu_source_conditioned_possible_evidence_facts)
            .and_then(|sum| sum.checked_add(self.gpu_source_conditioned_not_known_evidence_facts))
            .and_then(|sum| {
                sum.checked_add(self.gpu_source_conditioned_not_possible_evidence_facts)
            })
            .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: "source conditioned operator evidence fact counters overflowed"
                    .to_string(),
            })?;
        if source_operator_fact_count != self.gpu_source_conditioned_evidence_facts {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "source conditioned operator evidence facts must equal source evidence \
                     facts, got operators={} total={}",
                    source_operator_fact_count, self.gpu_source_conditioned_evidence_facts
                ),
            });
        }

        let program_operator_fact_count = self
            .gpu_program_conditioned_know_evidence_facts
            .checked_add(self.gpu_program_conditioned_possible_evidence_facts)
            .and_then(|sum| sum.checked_add(self.gpu_program_conditioned_not_known_evidence_facts))
            .and_then(|sum| {
                sum.checked_add(self.gpu_program_conditioned_not_possible_evidence_facts)
            })
            .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: "program conditioned operator evidence fact counters overflowed"
                    .to_string(),
            })?;
        if program_operator_fact_count != self.gpu_program_conditioned_evidence_facts {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "program conditioned operator evidence facts must equal program evidence \
                     facts, got operators={} total={}",
                    program_operator_fact_count, self.gpu_program_conditioned_evidence_facts
                ),
            });
        }

        let expected_max_arity = self
            .gpu_source_conditioned_max_evidence_arity
            .max(self.gpu_program_conditioned_max_evidence_arity);
        if self.gpu_conditioned_max_evidence_arity != expected_max_arity {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned max evidence arity must equal max(source, program), got \
                     aggregate={} source={} program={}",
                    self.gpu_conditioned_max_evidence_arity,
                    self.gpu_source_conditioned_max_evidence_arity,
                    self.gpu_program_conditioned_max_evidence_arity
                ),
            });
        }
        if (self.gpu_conditioned_nonzero_arity_evidence_facts == 0)
            != (self.gpu_conditioned_max_evidence_arity == 0)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned max evidence arity must be nonzero exactly when nonzero-arity \
                     facts are present, got nonzero={} max_arity={}",
                    self.gpu_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_conditioned_max_evidence_arity
                ),
            });
        }
        if (self.gpu_source_conditioned_nonzero_arity_evidence_facts == 0)
            != (self.gpu_source_conditioned_max_evidence_arity == 0)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "source conditioned max evidence arity must be nonzero exactly when \
                     source nonzero-arity facts are present, got nonzero={} max_arity={}",
                    self.gpu_source_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_source_conditioned_max_evidence_arity
                ),
            });
        }
        if (self.gpu_program_conditioned_nonzero_arity_evidence_facts == 0)
            != (self.gpu_program_conditioned_max_evidence_arity == 0)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "program conditioned max evidence arity must be nonzero exactly when \
                     program nonzero-arity facts are present, got nonzero={} max_arity={}",
                    self.gpu_program_conditioned_nonzero_arity_evidence_facts,
                    self.gpu_program_conditioned_max_evidence_arity
                ),
            });
        }
        if (self.accepted_evidence_assumptions_consumed != 0
            || self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed != 0
            || self.accepted_gpu_max_evidence_arity_consumed != 0)
            && (self.gpu_conditioned_evidence_facts > self.accepted_evidence_assumptions_consumed
                || self.gpu_conditioned_nonzero_arity_evidence_facts
                    > self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
                || self.gpu_conditioned_max_evidence_arity
                    > self.accepted_gpu_max_evidence_arity_consumed)
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned evidence facts must be bounded by accepted GPU evidence, got \
                     facts={}/{} nonzero={}/{} max_arity={}/{}",
                    self.gpu_conditioned_evidence_facts,
                    self.accepted_evidence_assumptions_consumed,
                    self.gpu_conditioned_nonzero_arity_evidence_facts,
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed,
                    self.gpu_conditioned_max_evidence_arity,
                    self.accepted_gpu_max_evidence_arity_consumed
                ),
            });
        }
        Ok(())
    }

    /// Require that this trace is eligible for production probability metrics.
    ///
    /// This gate proves accepted-evidence admission and fixture containment for
    /// this adapter path. It does not establish that every epistemic execution
    /// route produces a compatible evidence record or that accepted assumptions
    /// conditioned the probabilistic model.
    pub fn require_production_metric_eligibility(&self) -> Result<()> {
        let capabilities = production_capabilities();
        if capabilities.fixture_circuit_allowed {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "bounded EpistemicCircuit fixtures are not allowed for production metrics"
                    .to_string(),
            });
        }
        if capabilities.gpu_exact_provenance != EpistemicProbProductionCapabilityStatus::Available {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "GPU exact/provenance production capability is not available".to_string(),
            });
        }
        if capabilities.gpu_pir_cnf != EpistemicProbProductionCapabilityStatus::Available {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "GPU PIR/CNF production capability is not available".to_string(),
            });
        }
        if capabilities.gpu_knowledge_compilation
            != EpistemicProbProductionCapabilityStatus::Available
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: capabilities.gpu_knowledge_compilation_blocker.to_string(),
            });
        }
        if capabilities.gpu_exact_query_and_gradient
            != EpistemicProbProductionCapabilityStatus::Available
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "GPU exact query/gradient production capability is not available"
                    .to_string(),
            });
        }
        if self.accepted_world_view_evidence_consumed == 0 {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "production probability metrics require accepted world-view evidence"
                    .to_string(),
            });
        }
        let gpu_production_path_events = self.checked_gpu_production_path_events()?;
        if gpu_production_path_events == 0 {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "production probability metrics require an existing GPU exact/provenance/PIR/CNF/knowledge-compilation counter"
                    .to_string(),
            });
        }
        if self.accepted_gpu_production_path_events == 0 {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: "production probability metrics require GPU exact/provenance/PIR/CNF/knowledge-compilation work inside an accepted world-view evidence gate"
                    .to_string(),
            });
        }
        if self.accepted_gpu_production_path_events > gpu_production_path_events {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU probability production events cannot exceed total GPU production events: accepted={} total={}",
                    self.accepted_gpu_production_path_events, gpu_production_path_events
                ),
            });
        }
        if self.accepted_gpu_production_path_events < self.accepted_world_view_evidence_consumed {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production metric gate".to_string(),
                context: format!(
                    "accepted GPU probability production events must cover each accepted \
                     world-view evidence record, got accepted_events={} evidence={}",
                    self.accepted_gpu_production_path_events,
                    self.accepted_world_view_evidence_consumed
                ),
            });
        }
        self.require_accepted_gpu_world_view_evidence_trace()?;
        self.require_accepted_gpu_tuple_evidence_trace()?;
        self.require_conditioned_evidence_trace()?;
        self.require_pir_cnf_accounting()?;
        self.require_gpu_path_accounting()?;
        Ok(())
    }

    fn require_conditioned_evidence_metric_witness(&self) -> Result<()> {
        if self.accepted_conditioned_world_view_evidence_consumed == 0
            || self.gpu_conditioned_evidence_facts == 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context:
                    "production probability metrics require accepted world-view evidence compiled as exact evidence facts"
                        .to_string(),
            });
        }
        Ok(())
    }

    /// Require the stricter metric subset for accepted world-view evidence conditioning.
    ///
    /// General production eligibility proves fixture containment plus GPU exact/PIR/CNF/
    /// knowledge-compilation reuse. This gate additionally proves at least one accepted
    /// world view was compiled into exact evidence facts rather than only used as a
    /// production-path admission gate.
    pub fn require_conditioned_evidence_metric_eligibility(&self) -> Result<()> {
        self.require_production_metric_eligibility()?;
        self.require_conditioned_evidence_metric_witness()?;
        if self.gpu_conditioned_nonzero_arity_evidence_facts > 0
            && self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed == 0
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic conditioned evidence metric gate".to_string(),
                context: format!(
                    "conditioned nonzero-arity evidence facts require accepted GPU nonzero-arity \
                     assumptions, got conditioned_nonzero={} accepted_nonzero={}",
                    self.gpu_conditioned_nonzero_arity_evidence_facts,
                    self.accepted_gpu_nonzero_arity_evidence_assumptions_consumed
                ),
            });
        }
        Ok(())
    }
}

/// Device-side PIR/CNF evidence produced after accepted epistemic gating.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct EpistemicProbPirCnfEvidence {
    /// Number of host provenance PIR nodes uploaded to the GPU PIR layout.
    pub pir_nodes: usize,
    /// Number of roots supplied to GPU CNF encoding.
    pub root_count: usize,
    /// GPU CNF variable capacity emitted by `encode_cnf_gpu`.
    pub cnf_var_cap: u32,
    /// GPU CNF clause capacity emitted by `encode_cnf_gpu`.
    pub cnf_clause_cap: u32,
    /// GPU CNF literal capacity emitted by `encode_cnf_gpu`.
    pub cnf_lit_cap: u32,
}

/// One accepted GPU epistemic execution record used for probabilistic production gating.
#[derive(Clone, Copy)]
pub struct EpistemicProbGpuExecutionEvidence<'a> {
    /// Accepted GPU execution result whose world-view boundary must be validated.
    pub result: &'a EpistemicGpuExecutionResult,
    /// Epistemic assumptions represented by the accepted world view.
    pub assumptions: &'a [EpistemicAssumption],
}

/// Accepted GPU batch execution evidence used for probabilistic production gating.
pub struct EpistemicProbGpuBatchExecutionEvidence<'a> {
    /// Accepted GPU batch execution result whose aggregate trace and timing must be validated.
    pub batch: &'a EpistemicGpuBatchExecutionResult,
    /// Epistemic assumptions represented by each accepted component world view.
    pub assumptions_by_component: &'a [&'a [EpistemicAssumption]],
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EpistemicProbPirCnfPath {
    Source,
    Program,
}

/// Thin adapter from accepted epistemic evidence to the existing GPU exact path.
pub struct EpistemicProbProductionAdapter {
    config: GpuConfig,
    trace: EpistemicProbProductionTrace,
}

/// One accepted-evidence exact circuit whose independent fact weights may change.
///
/// Clones share a single lock covering both host probability metadata and the
/// device-resident weight tables. Evaluation and updates are therefore serialized
/// for the same prepared circuit while unrelated circuits remain independent.
/// If a failed device update cannot be rolled back, the shared circuit is
/// permanently invalidated and every later operation through every clone fails.
#[derive(Clone)]
pub struct PreparedConditionedProgram {
    #[cfg(feature = "host-io")]
    state: Arc<Mutex<PreparedConditionedState>>,
}

/// Authoritative identity and preparation count for one conditioned circuit.
///
/// `circuit_generation` is process-local and opaque. Together with `cache_slot`
/// it identifies the exact state and device-cache handle retained by a prepared
/// program; callers should compare values only within that handle's lifetime.
#[cfg(feature = "host-io")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ConditionedCircuitWitness {
    /// Actual GPU circuit-compiler invocations performed while preparing this handle.
    pub preparation_compiles: u64,
    /// Successful exact-circuit materializations performed while preparing this handle.
    pub materializations: u64,
    /// Verified disk-cache restorations used while preparing this handle.
    pub disk_cache_restores: u64,
    /// GPU circuit-cache hits used while preparing this handle.
    pub gpu_cache_hits: u64,
    /// Opaque process-local generation assigned to the retained exact state.
    pub circuit_generation: u64,
    /// Device-cache slot retained by the exact state.
    pub cache_slot: u32,
}

#[cfg(feature = "host-io")]
fn source_has_only_gpu_count_lift_queries(provenance: &Provenance) -> bool {
    !provenance.queries.is_empty()
        && provenance.evidence.is_empty()
        && provenance.choice_probs.is_empty()
        && provenance.queries.iter().all(|query| {
            provenance.aggregate_lifting.iter().any(|entry| {
                entry.status == AggregateLiftStatus::Fired
                    && entry.operator == "count"
                    && entry.deterministic_rows == 0
                    && entry.predicate == query.predicate
            })
        })
}

#[cfg(feature = "host-io")]
struct PreparedConditionedState {
    exact: ExactDdnnfProgram,
    evaluation_trace: EpistemicProbProductionTrace,
    initial_circuit: ExactCircuitWitness,
    successful_reuses: u64,
}

#[cfg(feature = "host-io")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PreparedCircuitSnapshot {
    circuit: ExactCircuitWitness,
    successful_reuses: u64,
}

#[cfg(feature = "host-io")]
impl PreparedConditionedState {
    fn ensure_usable(&self) -> Result<()> {
        self.exact.ensure_usable()
    }

    fn snapshot(&self) -> Result<PreparedCircuitSnapshot> {
        self.ensure_usable()?;
        let circuit = self.exact.circuit_witness()?;
        if circuit.circuit_generation != self.initial_circuit.circuit_generation
            || circuit.cache_slot != self.initial_circuit.cache_slot
        {
            return Err(XlogError::Compilation(
                "Prepared conditioned circuit identity changed after preparation".to_string(),
            ));
        }
        Ok(PreparedCircuitSnapshot {
            circuit,
            successful_reuses: self.successful_reuses,
        })
    }

    fn record_successful_reuse(&mut self) -> Result<PreparedCircuitSnapshot> {
        let before = self.snapshot()?;
        self.successful_reuses = self.successful_reuses.checked_add(1).ok_or_else(|| {
            XlogError::Compilation(
                "Prepared conditioned circuit reuse counter overflowed".to_string(),
            )
        })?;
        let after = self.snapshot()?;
        debug_assert_eq!(
            before.circuit.circuit_generation,
            after.circuit.circuit_generation
        );
        debug_assert_eq!(before.circuit.cache_slot, after.circuit.cache_slot);
        Ok(after)
    }
}

#[cfg(feature = "host-io")]
impl PreparedConditionedProgram {
    fn new(
        exact: ExactDdnnfProgram,
        preparation_trace: EpistemicProbProductionTrace,
    ) -> Result<Self> {
        let initial_circuit = exact.circuit_witness()?;
        validate_single_circuit_materialization(initial_circuit)?;
        if initial_circuit.compiler_invocations > 1 {
            return Err(XlogError::Compilation(format!(
                "Prepared conditioned circuit invoked the GPU compiler {} times",
                initial_circuit.compiler_invocations
            )));
        }
        Ok(Self {
            state: Arc::new(Mutex::new(PreparedConditionedState {
                exact,
                evaluation_trace: reuse_trace_template(preparation_trace),
                initial_circuit,
                successful_reuses: 0,
            })),
        })
    }

    fn lock_state(&self) -> Result<MutexGuard<'_, PreparedConditionedState>> {
        self.state.lock().map_err(|_| {
            XlogError::Execution(
                "Prepared conditioned state mutex is poisoned and permanently invalid".to_string(),
            )
        })
    }

    fn apply_fact_probability_update(
        &self,
        apply: impl FnOnce(&mut ExactDdnnfProgram) -> Result<()>,
    ) -> Result<()> {
        let mut state = self.lock_state()?;
        let before = state.snapshot()?;
        let update_result = apply(&mut state.exact);
        if let Err(update_error) = update_result {
            if state.ensure_usable().is_err() {
                return Err(update_error);
            }
            let after = state.snapshot()?;
            if before != after {
                return Err(XlogError::Compilation(
                    "Failed fact probability update changed the prepared circuit identity or compile ledger"
                        .to_string(),
                ));
            }
            return Err(update_error);
        }
        let after = state.snapshot()?;
        if before != after {
            return Err(XlogError::Compilation(
                "Fact probability update changed the prepared circuit identity or compile ledger"
                    .to_string(),
            ));
        }
        Ok(())
    }

    /// Evaluate the prepared conditioned circuit without recompiling it.
    #[cfg(feature = "host-io")]
    pub fn evaluate(&self) -> Result<(ExactResult, EpistemicProbProductionTrace)> {
        let mut state = self.lock_state()?;
        let before = state.snapshot()?;
        let result = state.exact.evaluate()?;
        let after = state.record_successful_reuse()?;
        let trace = query_reuse_trace(state.evaluation_trace, before, after)?;
        trace.require_conditioned_evidence_metric_eligibility()?;
        Ok((result, trace))
    }

    /// Evaluate probabilities and gradients without recompiling the circuit.
    #[cfg(feature = "host-io")]
    pub fn evaluate_with_grads(
        &self,
    ) -> Result<(ExactResultWithGrads, EpistemicProbProductionTrace)> {
        let mut state = self.lock_state()?;
        let before = state.snapshot()?;
        let result = state.exact.evaluate_gpu_with_grads()?;
        let after = state.record_successful_reuse()?;
        let trace = gradient_reuse_trace(state.evaluation_trace, before, after)?;
        trace.require_conditioned_evidence_metric_eligibility()?;
        Ok((result, trace))
    }

    /// Return the current CNF-variable metadata, including updated fact priors.
    pub fn prob_var_map(&self) -> Result<Vec<ProbVarInfo>> {
        let state = self.lock_state()?;
        state.snapshot()?;
        state.exact.checked_prob_var_map()
    }

    /// Return the immutable exact-state identity and lifetime preparation count.
    pub fn circuit_witness(&self) -> Result<ConditionedCircuitWitness> {
        let state = self.lock_state()?;
        conditioned_circuit_witness(state.snapshot()?)
    }

    /// Atomically update independent probabilistic fact weights.
    ///
    /// A failed device write is rolled back before this returns. If that rollback
    /// also fails, the shared prepared circuit is permanently invalidated so that
    /// neither this handle nor any clone can observe potentially partial weights.
    #[cfg(feature = "host-io")]
    pub fn set_fact_probabilities(&self, updates: &BTreeMap<u32, f64>) -> Result<()> {
        self.apply_fact_probability_update(|exact| exact.set_fact_probabilities(updates))
    }

    #[cfg(all(test, feature = "host-io"))]
    fn set_fact_probabilities_with_device_failures(
        &self,
        updates: &BTreeMap<u32, f64>,
        fail_after_successful_writes: Option<usize>,
        fail_after_successful_rollback_writes: Option<usize>,
    ) -> Result<()> {
        self.apply_fact_probability_update(|exact| {
            exact.set_fact_probabilities_with_device_failures_for_test(
                updates,
                fail_after_successful_writes,
                fail_after_successful_rollback_writes,
            )
        })
    }
}

#[cfg(feature = "host-io")]
fn conditioned_circuit_witness(
    snapshot: PreparedCircuitSnapshot,
) -> Result<ConditionedCircuitWitness> {
    validate_single_circuit_materialization(snapshot.circuit)?;
    Ok(ConditionedCircuitWitness {
        preparation_compiles: snapshot.circuit.compiler_invocations,
        materializations: snapshot.circuit.materializations,
        disk_cache_restores: snapshot.circuit.disk_cache_restores,
        gpu_cache_hits: snapshot.circuit.gpu_cache_hits,
        circuit_generation: snapshot.circuit.circuit_generation,
        cache_slot: snapshot.circuit.cache_slot,
    })
}

#[cfg(feature = "host-io")]
fn validate_single_circuit_materialization(circuit: ExactCircuitWitness) -> Result<()> {
    let origins = circuit
        .compiler_invocations
        .checked_add(circuit.disk_cache_restores)
        .and_then(|count| count.checked_add(circuit.gpu_cache_hits))
        .ok_or_else(|| {
            XlogError::Compilation(
                "Prepared conditioned circuit materialization ledger overflowed".to_string(),
            )
        })?;
    if circuit.materializations != 1 || origins != 1 {
        return Err(XlogError::Compilation(format!(
            "Prepared conditioned circuit requires one materialization from one compile/cache event, got materializations={} compiler_invocations={} disk_cache_restores={} gpu_cache_hits={}",
            circuit.materializations,
            circuit.compiler_invocations,
            circuit.disk_cache_restores,
            circuit.gpu_cache_hits
        )));
    }
    Ok(())
}

#[cfg(feature = "host-io")]
fn reuse_trace_template(
    mut preparation_trace: EpistemicProbProductionTrace,
) -> EpistemicProbProductionTrace {
    preparation_trace.gpu_exact_query_evaluations = 0;
    preparation_trace.gpu_source_exact_query_evaluations = 0;
    preparation_trace.gpu_program_exact_query_evaluations = 0;
    preparation_trace.gpu_exact_gradient_evaluations = 0;
    preparation_trace.gpu_source_exact_gradient_evaluations = 0;
    preparation_trace.gpu_program_exact_gradient_evaluations = 0;
    preparation_trace.gpu_source_conditioned_gradient_evaluations = 0;
    preparation_trace.gpu_program_conditioned_gradient_evaluations = 0;
    preparation_trace.gpu_pir_graph_uploads = 0;
    preparation_trace.gpu_source_pir_graph_uploads = 0;
    preparation_trace.gpu_program_pir_graph_uploads = 0;
    preparation_trace.gpu_cnf_encodes = 0;
    preparation_trace.gpu_source_cnf_encodes = 0;
    preparation_trace.gpu_program_cnf_encodes = 0;
    preparation_trace.gpu_knowledge_compilation_end_to_end_runs = 0;
    preparation_trace.gpu_source_knowledge_compilation_end_to_end_runs = 0;
    preparation_trace.gpu_program_knowledge_compilation_end_to_end_runs = 0;
    preparation_trace.accepted_gpu_production_path_events = 0;
    preparation_trace
}

#[cfg(feature = "host-io")]
fn query_reuse_trace(
    mut trace: EpistemicProbProductionTrace,
    before: PreparedCircuitSnapshot,
    after: PreparedCircuitSnapshot,
) -> Result<EpistemicProbProductionTrace> {
    apply_authoritative_reuse_witness(&mut trace, before, after)?;
    trace.gpu_exact_query_evaluations = trace.gpu_conditioned_circuit_reuses;
    trace.gpu_source_exact_query_evaluations = trace.gpu_conditioned_circuit_reuses;
    trace.accepted_gpu_production_path_events = trace.checked_gpu_production_path_events()?;
    Ok(trace)
}

#[cfg(feature = "host-io")]
fn gradient_reuse_trace(
    mut trace: EpistemicProbProductionTrace,
    before: PreparedCircuitSnapshot,
    after: PreparedCircuitSnapshot,
) -> Result<EpistemicProbProductionTrace> {
    apply_authoritative_reuse_witness(&mut trace, before, after)?;
    trace.gpu_exact_gradient_evaluations = trace.gpu_conditioned_circuit_reuses;
    trace.gpu_source_exact_gradient_evaluations = trace.gpu_conditioned_circuit_reuses;
    trace.gpu_source_conditioned_gradient_evaluations = trace.gpu_conditioned_circuit_reuses;
    trace.accepted_gpu_production_path_events = trace.checked_gpu_production_path_events()?;
    Ok(trace)
}

#[cfg(feature = "host-io")]
fn apply_authoritative_reuse_witness(
    trace: &mut EpistemicProbProductionTrace,
    before: PreparedCircuitSnapshot,
    after: PreparedCircuitSnapshot,
) -> Result<()> {
    if before.circuit.circuit_generation != after.circuit.circuit_generation
        || before.circuit.cache_slot != after.circuit.cache_slot
    {
        return Err(XlogError::Compilation(
            "Prepared conditioned circuit identity changed during evaluation".to_string(),
        ));
    }
    trace.gpu_exact_source_compiles = after
        .circuit
        .compiler_invocations
        .checked_sub(before.circuit.compiler_invocations)
        .ok_or_else(|| {
            XlogError::Compilation(
                "Prepared conditioned compiler invocation ledger regressed".to_string(),
            )
        })?;
    trace.gpu_exact_program_compiles = 0;
    trace.gpu_conditioned_circuit_reuses = after
        .successful_reuses
        .checked_sub(before.successful_reuses)
        .ok_or_else(|| {
            XlogError::Compilation(
                "Prepared conditioned circuit reuse counter regressed".to_string(),
            )
        })?;
    let witness = conditioned_circuit_witness(after)?;
    trace.gpu_conditioned_circuit_preparation_compiles = witness.preparation_compiles;
    trace.gpu_conditioned_circuit_materializations = witness.materializations;
    trace.gpu_conditioned_circuit_disk_cache_restores = witness.disk_cache_restores;
    trace.gpu_conditioned_circuit_gpu_cache_hits = witness.gpu_cache_hits;
    trace.gpu_conditioned_circuit_generation = witness.circuit_generation;
    trace.gpu_conditioned_circuit_cache_slot = u64::from(witness.cache_slot);
    if trace.gpu_exact_source_compiles != 0
        || trace.gpu_exact_program_compiles != 0
        || trace.gpu_conditioned_circuit_reuses != 1
    {
        return Err(XlogError::Compilation(format!(
            "Prepared conditioned evaluation must not compile and must reuse one circuit, got source_compiles={} program_compiles={} reuses={}",
            trace.gpu_exact_source_compiles,
            trace.gpu_exact_program_compiles,
            trace.gpu_conditioned_circuit_reuses
        )));
    }
    Ok(())
}

#[cfg(feature = "host-io")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EpistemicProbConditionedEvidencePath {
    Source,
    Program,
}

impl EpistemicProbProductionAdapter {
    fn checked_trace_counter_add(current: u64, delta: u64, counter: &str) -> Result<u64> {
        current
            .checked_add(delta)
            .ok_or_else(|| XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production trace accounting".to_string(),
                context: format!(
                    "accepted GPU probability trace counter {counter} overflowed while adding \
                    {delta} to {current}"
                ),
            })
    }

    #[cfg(feature = "host-io")]
    fn record_gpu_exact_query_evaluation(&mut self, program: &ExactDdnnfProgram) -> Result<()> {
        checked_prob_trace_counter_inc!(self, gpu_exact_query_evaluations);
        match program.origin() {
            ExactProgramOrigin::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_exact_query_evaluations);
            }
            ExactProgramOrigin::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_exact_query_evaluations);
            }
        }
        Ok(())
    }

    #[cfg(feature = "host-io")]
    fn record_gpu_exact_gradient_evaluation(&mut self, program: &ExactDdnnfProgram) -> Result<()> {
        self.record_gpu_exact_gradient_evaluation_for_origin(program.origin())
    }

    #[cfg(feature = "host-io")]
    fn record_gpu_exact_gradient_evaluation_for_origin(
        &mut self,
        origin: ExactProgramOrigin,
    ) -> Result<()> {
        checked_prob_trace_counter_inc!(self, gpu_exact_gradient_evaluations);
        match origin {
            ExactProgramOrigin::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_exact_gradient_evaluations);
            }
            ExactProgramOrigin::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_exact_gradient_evaluations);
            }
        }
        Ok(())
    }

    /// Create a production adapter with a GPU exact inference configuration.
    pub fn new(config: GpuConfig) -> Self {
        Self {
            config,
            trace: EpistemicProbProductionTrace::default(),
        }
    }

    /// Return current production-path trace counters.
    pub fn trace(&self) -> EpistemicProbProductionTrace {
        self.trace
    }

    /// Apply accepted world-view evidence to a caller-owned incremental circuit fixture.
    ///
    /// This records accepted evidence after the typed `Gpu`/`RejectUnsupported` plan boundary,
    /// but it is not a production metric event. Production metric eligibility still requires
    /// observed GPU exact/provenance/PIR/CNF/knowledge-compilation path events.
    pub fn apply_accepted_world_view_to_circuit(
        &mut self,
        circuit: &mut EpistemicCircuit,
        evidence: AcceptedWorldViewEvidence,
    ) -> Result<CircuitUpdate> {
        epistemic_prob_trace_transaction!(self, {
            self.consume_accepted_evidence(&evidence)?;
            let update = circuit.apply_accepted_world_view(evidence)?;
            if update.mode == CircuitUpdateMode::IncrementalEvidence {
                self.trace.accepted_incremental_circuit_updates = Self::checked_trace_counter_add(
                    self.trace.accepted_incremental_circuit_updates,
                    1,
                    "accepted_incremental_circuit_updates",
                )?;
            }
            Ok(update)
        })
    }

    /// Apply accepted GPU epistemic execution evidence to a caller-owned incremental circuit.
    pub fn apply_accepted_world_view_to_circuit_with_gpu_execution_result(
        &mut self,
        circuit: &mut EpistemicCircuit,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<CircuitUpdate> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.apply_accepted_world_view_to_circuit(circuit, evidence)
    }

    /// Apply accepted split/batch GPU epistemic execution evidence to an incremental circuit.
    pub fn apply_accepted_world_views_to_circuit_for_gpu_batch_execution_result(
        &mut self,
        circuit: &mut EpistemicCircuit,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<CircuitUpdate>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic incremental circuit batch production",
            )?;

            let mut updates = Vec::with_capacity(accepted.len());
            for evidence in accepted {
                updates.push(self.apply_accepted_world_view_to_circuit(circuit, evidence)?);
            }
            Ok(updates)
        })
    }

    fn accepted_world_views_from_gpu_batch_execution_evidence(
        &mut self,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
        construct: &str,
    ) -> Result<Vec<AcceptedWorldViewEvidence>> {
        if evidence.batch.results.is_empty() {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: construct.to_string(),
                context: "probabilistic batch gating requires at least one accepted GPU component"
                    .to_string(),
            });
        }
        if evidence.assumptions_by_component.len() != evidence.batch.results.len() {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: construct.to_string(),
                context: format!(
                    "assumption group count {} does not match GPU batch component count {}",
                    evidence.assumptions_by_component.len(),
                    evidence.batch.results.len()
                ),
            });
        }
        let batch_trace = evidence.batch.trace;
        if batch_trace.component_count != evidence.batch.results.len()
            || batch_trace.gpu_runtime_component_executions != evidence.batch.results.len()
            || batch_trace.tracked_dtoh_calls != 0
            || batch_trace.tracked_htod_calls != 0
            || batch_trace.tracked_data_plane_htod_calls != 0
            || batch_trace.per_candidate_host_round_trips != 0
            || batch_trace.violated_constraint_relations != 0
            || !batch_trace.aggregate_kernel_timing.is_recorded()
        {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: construct.to_string(),
                context: format!(
                    "accepted GPU batch evidence requires complete GPU component execution and \
                     zero observed hot-path transfers outside bounded launch metadata plus \
                     aggregate CUDA-event timing, got components={}/{}, dtoh_calls={}, \
                     htod_calls={}, data_plane_htod_calls={}, launch_metadata_htod_calls={}, \
                     round_trips={}, constraint_violations={}, aggregate_timing_recorded={}",
                    batch_trace.gpu_runtime_component_executions,
                    batch_trace.component_count,
                    batch_trace.tracked_dtoh_calls,
                    batch_trace.tracked_htod_calls,
                    batch_trace.tracked_data_plane_htod_calls,
                    batch_trace.tracked_launch_metadata_htod_calls,
                    batch_trace.per_candidate_host_round_trips,
                    batch_trace.violated_constraint_relations,
                    batch_trace.aggregate_kernel_timing.is_recorded()
                ),
            });
        }
        evidence.batch.require_trace_matches_components(construct)?;

        let mut accepted = Vec::with_capacity(evidence.batch.results.len());
        for (result, assumptions) in evidence
            .batch
            .results
            .iter()
            .zip(evidence.assumptions_by_component.iter())
        {
            accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                (*assumptions).to_vec(),
            )?);
        }

        self.trace.accepted_gpu_batch_evidence_consumed = Self::checked_trace_counter_add(
            self.trace.accepted_gpu_batch_evidence_consumed,
            1,
            "accepted_gpu_batch_evidence_consumed",
        )?;
        self.trace.accepted_gpu_batch_component_evidence_consumed =
            Self::checked_trace_counter_add(
                self.trace.accepted_gpu_batch_component_evidence_consumed,
                accepted.len() as u64,
                "accepted_gpu_batch_component_evidence_consumed",
            )?;

        Ok(accepted)
    }

    #[cfg(feature = "host-io")]
    fn record_conditioned_evidence_counts(
        &mut self,
        counts: EpistemicProbConditionedEvidenceCounts,
        path: EpistemicProbConditionedEvidencePath,
    ) -> Result<()> {
        macro_rules! add_counter {
            ($field:ident, $delta:expr) => {
                self.trace.$field =
                    Self::checked_trace_counter_add(self.trace.$field, $delta, stringify!($field))?;
            };
        }

        add_counter!(accepted_conditioned_world_view_evidence_consumed, 1);
        add_counter!(gpu_conditioned_evidence_facts, counts.total as u64);
        add_counter!(
            gpu_conditioned_nonzero_arity_evidence_facts,
            counts.nonzero_arity as u64
        );
        self.trace.gpu_conditioned_max_evidence_arity = self
            .trace
            .gpu_conditioned_max_evidence_arity
            .max(counts.max_arity);
        add_counter!(
            gpu_conditioned_negative_evidence_facts,
            counts.negative as u64
        );
        match path {
            EpistemicProbConditionedEvidencePath::Source => {
                add_counter!(accepted_source_conditioned_world_view_evidence_consumed, 1);
                add_counter!(gpu_source_conditioned_evidence_facts, counts.total as u64);
                add_counter!(
                    gpu_source_conditioned_nonzero_arity_evidence_facts,
                    counts.nonzero_arity as u64
                );
                self.trace.gpu_source_conditioned_max_evidence_arity = self
                    .trace
                    .gpu_source_conditioned_max_evidence_arity
                    .max(counts.max_arity);
                add_counter!(
                    gpu_source_conditioned_negative_evidence_facts,
                    counts.negative as u64
                );
                add_counter!(
                    gpu_source_conditioned_know_evidence_facts,
                    counts.know as u64
                );
                add_counter!(
                    gpu_source_conditioned_possible_evidence_facts,
                    counts.possible as u64
                );
                add_counter!(
                    gpu_source_conditioned_not_known_evidence_facts,
                    counts.not_known as u64
                );
                add_counter!(
                    gpu_source_conditioned_not_possible_evidence_facts,
                    counts.not_possible as u64
                );
            }
            EpistemicProbConditionedEvidencePath::Program => {
                add_counter!(accepted_program_conditioned_world_view_evidence_consumed, 1);
                add_counter!(gpu_program_conditioned_evidence_facts, counts.total as u64);
                add_counter!(
                    gpu_program_conditioned_nonzero_arity_evidence_facts,
                    counts.nonzero_arity as u64
                );
                self.trace.gpu_program_conditioned_max_evidence_arity = self
                    .trace
                    .gpu_program_conditioned_max_evidence_arity
                    .max(counts.max_arity);
                add_counter!(
                    gpu_program_conditioned_negative_evidence_facts,
                    counts.negative as u64
                );
                add_counter!(
                    gpu_program_conditioned_know_evidence_facts,
                    counts.know as u64
                );
                add_counter!(
                    gpu_program_conditioned_possible_evidence_facts,
                    counts.possible as u64
                );
                add_counter!(
                    gpu_program_conditioned_not_known_evidence_facts,
                    counts.not_known as u64
                );
                add_counter!(
                    gpu_program_conditioned_not_possible_evidence_facts,
                    counts.not_possible as u64
                );
            }
        }
        add_counter!(gpu_conditioned_know_evidence_facts, counts.know as u64);
        add_counter!(
            gpu_conditioned_possible_evidence_facts,
            counts.possible as u64
        );
        add_counter!(
            gpu_conditioned_not_known_evidence_facts,
            counts.not_known as u64
        );
        add_counter!(
            gpu_conditioned_not_possible_evidence_facts,
            counts.not_possible as u64
        );
        Ok(())
    }

    fn record_accepted_gpu_production_path_events_since(
        &mut self,
        events_before: u64,
    ) -> Result<()> {
        let events_after = self.trace.checked_gpu_production_path_events()?;
        let delta = events_after.checked_sub(events_before).ok_or_else(|| {
            XlogError::UnsupportedEpistemicConstruct {
                construct: "epistemic probabilistic production trace accounting".to_string(),
                context: format!(
                    "accepted GPU probability production events decreased from {events_before} to \
                     {events_after}"
                ),
            }
        })?;
        self.trace.accepted_gpu_production_path_events = Self::checked_trace_counter_add(
            self.trace.accepted_gpu_production_path_events,
            delta,
            "accepted_gpu_production_path_events",
        )?;
        Ok(())
    }

    /// Compile source through the existing GPU-native exact/provenance path.
    pub fn compile_source_with_accepted_world_view(
        &mut self,
        source: &str,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactDdnnfProgram> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            let program = ExactDdnnfProgram::compile_source_with_gpu(source, self.config)?;
            require_gpu_exact_backend(&program, "epistemic probabilistic source exact compile")?;
            checked_prob_trace_counter_inc!(self, gpu_exact_source_compiles);
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(program)
        })
    }

    /// Compile source through the GPU exact path after accepted GPU epistemic execution.
    pub fn compile_source_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactDdnnfProgram> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.compile_source_with_accepted_world_view(source, &evidence)
    }

    /// Compile source once per accepted GPU epistemic execution result.
    pub fn compile_source_for_gpu_execution_results(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactDdnnfProgram>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic source exact compile batch".to_string(),
                    context:
                        "batched source exact compile requires at least one accepted GPU result"
                            .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut programs = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                programs.push(self.compile_source_with_accepted_world_view(source, evidence)?);
            }
            Ok(programs)
        })
    }

    /// Compile source once per accepted split/batch GPU epistemic component.
    pub fn compile_source_for_gpu_batch_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactDdnnfProgram>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic source exact compile batch production",
            )?;

            let mut programs = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                programs.push(self.compile_source_with_accepted_world_view(source, evidence)?);
            }
            Ok(programs)
        })
    }

    /// Compile a parsed program through the existing GPU-native exact/provenance path.
    pub fn compile_program_with_accepted_world_view(
        &mut self,
        program: &Program,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactDdnnfProgram> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            let exact = ExactDdnnfProgram::compile_from_program(program, self.config)?;
            require_gpu_exact_backend(
                &exact,
                "epistemic probabilistic parsed-program exact compile",
            )?;
            checked_prob_trace_counter_inc!(self, gpu_exact_program_compiles);
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(exact)
        })
    }

    /// Compile a parsed program through the GPU exact path after accepted GPU epistemic execution.
    pub fn compile_program_with_gpu_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactDdnnfProgram> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.compile_program_with_accepted_world_view(program, &evidence)
    }

    /// Compile a parsed program once per accepted GPU epistemic execution result.
    pub fn compile_program_for_gpu_execution_results(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactDdnnfProgram>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic parsed-program exact compile batch"
                        .to_string(),
                    context: "batched parsed-program exact compile requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut programs = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                programs.push(self.compile_program_with_accepted_world_view(program, evidence)?);
            }
            Ok(programs)
        })
    }

    /// Compile a parsed program once per accepted split/batch GPU epistemic component.
    pub fn compile_program_for_gpu_batch_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactDdnnfProgram>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic parsed-program exact compile batch production",
            )?;

            let mut programs = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                programs.push(self.compile_program_with_accepted_world_view(program, evidence)?);
            }
            Ok(programs)
        })
    }

    /// Compile source and evaluate queries through the existing GPU exact path after one accepted gate.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_source_with_accepted_world_view(
        &mut self,
        source: &str,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            let program = ExactDdnnfProgram::compile_source_with_gpu(source, self.config)?;
            require_gpu_exact_backend(
                &program,
                "epistemic probabilistic source exact compile/evaluate",
            )?;
            checked_prob_trace_counter_inc!(self, gpu_exact_source_compiles);
            let result = program.evaluate()?;
            checked_prob_trace_counter_inc!(self, gpu_exact_query_evaluations);
            checked_prob_trace_counter_inc!(self, gpu_source_exact_query_evaluations);
            checked_prob_trace_counter_inc!(self, gpu_knowledge_compilation_end_to_end_runs);
            checked_prob_trace_counter_inc!(self, gpu_source_knowledge_compilation_end_to_end_runs);
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(result)
        })
    }

    /// Compile source and evaluate queries after accepted GPU epistemic execution.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_source_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResult> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.compile_and_evaluate_source_with_accepted_world_view(source, &evidence)
    }

    /// Compile and evaluate source once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_source_for_gpu_execution_results(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic production batch".to_string(),
                    context:
                        "batched knowledge compilation requires at least one accepted GPU result"
                            .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(
                    self.compile_and_evaluate_source_with_accepted_world_view(source, evidence)?,
                );
            }
            Ok(results)
        })
    }

    /// Compile and evaluate source once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_source_for_gpu_batch_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(
                    self.compile_and_evaluate_source_with_accepted_world_view(source, evidence)?,
                );
            }
            Ok(results)
        })
    }

    /// Compile source with accepted zero-arity epistemic assumptions as exact evidence.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_accepted_world_view(
        &mut self,
        source: &str,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            self.compile_and_evaluate_conditioned_program_with_path(
                &program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Source,
                "epistemic probabilistic conditioned source exact compile/evaluate",
            )
        })
    }

    /// Compile source with accepted GPU epistemic assumptions as exact evidence.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.compile_and_evaluate_conditioned_program_with_path(
                &program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Source,
                "epistemic probabilistic conditioned source exact compile/evaluate",
            )
        })
    }

    /// Compile accepted epistemic evidence and an exact probabilistic source once.
    ///
    /// The returned handle reuses the conditioned circuit and permits atomic
    /// updates only for independent probabilistic fact variables.
    #[cfg(feature = "host-io")]
    pub fn prepare_conditioned_source_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<PreparedConditionedProgram> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let program = parse_program(source)?;
            if program.directives.prob_engine_or_default() != ProbEngine::ExactDdnnf {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "reusable conditioned circuit".to_string(),
                    context: "reusable conditioned programs require the exact Decision-DNNF engine; Monte Carlo programs cannot expose mutable CNF fact weights"
                        .to_string(),
                });
            }
            let provenance = extract_from_program(&program)?;
            if source_has_only_gpu_count_lift_queries(&provenance) {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "reusable conditioned circuit".to_string(),
                    context:
                        "GPU count-lift exact programs do not expose mutable CNF fact variables"
                            .to_string(),
                });
            }
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            let (conditioned_program, evidence_counts) =
                condition_program_with_accepted_evidence_using_provenance(
                    &program,
                    &provenance,
                    evidence,
                )?;
            let exact = ExactDdnnfProgram::compile_from_program(&conditioned_program, self.config)?;
            if exact.uses_gpu_native_count_lift() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "reusable conditioned circuit".to_string(),
                    context: "GPU count-lift exact programs do not expose mutable CNF fact weights"
                        .to_string(),
                });
            }
            require_gpu_exact_backend(
                &exact,
                "epistemic probabilistic reusable conditioned source exact compile",
            )?;
            checked_prob_trace_counter_inc!(self, gpu_exact_source_compiles);
            self.record_conditioned_evidence_counts(
                evidence_counts,
                EpistemicProbConditionedEvidencePath::Source,
            )?;
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            PreparedConditionedProgram::new(exact, self.trace)
        })
    }

    /// Compile conditioned source once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_for_gpu_execution_results(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic conditioned production batch".to_string(),
                    context: "batched conditioned knowledge compilation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.compile_and_evaluate_conditioned_program_with_path(
                    &program,
                    &provenance,
                    evidence,
                    EpistemicProbConditionedEvidencePath::Source,
                    "epistemic probabilistic conditioned source exact compile/evaluate",
                )?);
            }
            Ok(results)
        })
    }

    /// Compile conditioned source once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_for_gpu_batch_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.compile_and_evaluate_conditioned_program_with_path(
                    &program,
                    &provenance,
                    evidence,
                    EpistemicProbConditionedEvidencePath::Source,
                    "epistemic probabilistic conditioned source exact compile/evaluate",
                )?);
            }
            Ok(results)
        })
    }

    /// Compile source with accepted epistemic assumptions as exact evidence and evaluate gradients.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_grads_with_accepted_world_view(
        &mut self,
        source: &str,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResultWithGrads> {
        epistemic_prob_trace_transaction!(self, {
            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                &program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Source,
                "epistemic probabilistic conditioned source exact gradient",
            )
        })
    }

    /// Compile source with accepted GPU epistemic assumptions as exact evidence and evaluate gradients.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_grads_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResultWithGrads> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                &program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Source,
                "epistemic probabilistic conditioned source exact gradient",
            )
        })
    }

    /// Compile conditioned source gradients once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_grads_for_gpu_execution_results(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic conditioned gradient production batch"
                        .to_string(),
                    context: "batched conditioned gradient compilation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(
                    self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                        &program,
                        &provenance,
                        evidence,
                        EpistemicProbConditionedEvidencePath::Source,
                        "epistemic probabilistic conditioned source exact gradient",
                    )?,
                );
            }
            Ok(results)
        })
    }

    /// Compile conditioned source gradients once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_source_with_grads_for_gpu_batch_execution_result(
        &mut self,
        source: &str,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let program = parse_program(source)?;
            let provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(
                    self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                        &program,
                        &provenance,
                        evidence,
                        EpistemicProbConditionedEvidencePath::Source,
                        "epistemic probabilistic conditioned source exact gradient",
                    )?,
                );
            }
            Ok(results)
        })
    }

    #[cfg(feature = "host-io")]
    fn compile_and_evaluate_conditioned_program_with_path(
        &mut self,
        program: &Program,
        provenance: &Provenance,
        evidence: &AcceptedWorldViewEvidence,
        path: EpistemicProbConditionedEvidencePath,
        backend_context: &'static str,
    ) -> Result<ExactResult> {
        self.require_accepted_evidence(evidence)?;
        let production_events_before = self.trace.checked_gpu_production_path_events()?;
        let (program, evidence_counts) = condition_program_with_accepted_evidence_using_provenance(
            program, provenance, evidence,
        )?;
        let exact = ExactDdnnfProgram::compile_from_program(&program, self.config)?;
        require_gpu_exact_backend(&exact, backend_context)?;
        match path {
            EpistemicProbConditionedEvidencePath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_exact_source_compiles);
            }
            EpistemicProbConditionedEvidencePath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_exact_program_compiles);
            }
        }
        self.record_conditioned_evidence_counts(evidence_counts, path)?;
        let result = exact.evaluate()?;
        checked_prob_trace_counter_inc!(self, gpu_exact_query_evaluations);
        checked_prob_trace_counter_inc!(self, gpu_knowledge_compilation_end_to_end_runs);
        match path {
            EpistemicProbConditionedEvidencePath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_exact_query_evaluations);
                checked_prob_trace_counter_inc!(
                    self,
                    gpu_source_knowledge_compilation_end_to_end_runs
                );
            }
            EpistemicProbConditionedEvidencePath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_exact_query_evaluations);
                checked_prob_trace_counter_inc!(
                    self,
                    gpu_program_knowledge_compilation_end_to_end_runs
                );
            }
        }
        self.record_accepted_gpu_production_path_events_since(production_events_before)?;
        self.record_accepted_evidence(evidence)?;
        Ok(result)
    }

    #[cfg(feature = "host-io")]
    fn compile_and_evaluate_conditioned_program_with_grads_with_path(
        &mut self,
        program: &Program,
        provenance: &Provenance,
        evidence: &AcceptedWorldViewEvidence,
        path: EpistemicProbConditionedEvidencePath,
        backend_context: &'static str,
    ) -> Result<ExactResultWithGrads> {
        self.require_accepted_evidence(evidence)?;
        let production_events_before = self.trace.checked_gpu_production_path_events()?;
        let (program, evidence_counts) = condition_program_with_accepted_evidence_using_provenance(
            program, provenance, evidence,
        )?;
        let exact = ExactDdnnfProgram::compile_from_program(&program, self.config)?;
        require_gpu_exact_backend(&exact, backend_context)?;
        match path {
            EpistemicProbConditionedEvidencePath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_exact_source_compiles);
            }
            EpistemicProbConditionedEvidencePath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_exact_program_compiles);
            }
        }
        self.record_conditioned_evidence_counts(evidence_counts, path)?;
        let result = exact.evaluate_gpu_with_grads()?;
        let origin = match path {
            EpistemicProbConditionedEvidencePath::Source => ExactProgramOrigin::Source,
            EpistemicProbConditionedEvidencePath::Program => ExactProgramOrigin::Program,
        };
        self.record_gpu_exact_gradient_evaluation_for_origin(origin)?;
        checked_prob_trace_counter_inc!(self, gpu_knowledge_compilation_end_to_end_runs);
        match path {
            EpistemicProbConditionedEvidencePath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_conditioned_gradient_evaluations);
                checked_prob_trace_counter_inc!(
                    self,
                    gpu_source_knowledge_compilation_end_to_end_runs
                );
            }
            EpistemicProbConditionedEvidencePath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_conditioned_gradient_evaluations);
                checked_prob_trace_counter_inc!(
                    self,
                    gpu_program_knowledge_compilation_end_to_end_runs
                );
            }
        }
        self.record_accepted_gpu_production_path_events_since(production_events_before)?;
        self.record_accepted_evidence(evidence)?;
        Ok(result)
    }

    /// Compile a parsed program with accepted epistemic assumptions as exact evidence.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_accepted_world_view(
        &mut self,
        program: &Program,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            let provenance = extract_from_program(program)?;
            self.compile_and_evaluate_conditioned_program_with_path(
                program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Program,
                "epistemic probabilistic conditioned parsed-program exact compile/evaluate",
            )
        })
    }

    /// Compile a parsed program with accepted GPU epistemic assumptions as exact evidence.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_gpu_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let provenance = extract_from_program(program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.compile_and_evaluate_conditioned_program_with_path(
                program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Program,
                "epistemic probabilistic conditioned parsed-program exact compile/evaluate",
            )
        })
    }

    /// Compile conditioned parsed program once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_for_gpu_execution_results(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic conditioned production batch".to_string(),
                    context: "batched conditioned program compilation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.compile_and_evaluate_conditioned_program_with_path(
                    program,
                    &provenance,
                    evidence,
                    EpistemicProbConditionedEvidencePath::Program,
                    "epistemic probabilistic conditioned parsed-program exact compile/evaluate",
                )?);
            }
            Ok(results)
        })
    }

    /// Compile conditioned parsed program once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_for_gpu_batch_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.compile_and_evaluate_conditioned_program_with_path(
                    program,
                    &provenance,
                    evidence,
                    EpistemicProbConditionedEvidencePath::Program,
                    "epistemic probabilistic conditioned parsed-program exact compile/evaluate",
                )?);
            }
            Ok(results)
        })
    }

    /// Compile a parsed program with accepted epistemic assumptions as exact evidence and evaluate gradients.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_grads_with_accepted_world_view(
        &mut self,
        program: &Program,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResultWithGrads> {
        epistemic_prob_trace_transaction!(self, {
            let provenance = extract_from_program(program)?;
            self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Program,
                "epistemic probabilistic conditioned parsed-program exact gradient",
            )
        })
    }

    /// Compile a parsed program with accepted GPU epistemic assumptions as exact evidence and evaluate gradients.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_grads_with_gpu_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResultWithGrads> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let provenance = extract_from_program(program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                program,
                &provenance,
                evidence,
                EpistemicProbConditionedEvidencePath::Program,
                "epistemic probabilistic conditioned parsed-program exact gradient",
            )
        })
    }

    /// Compile conditioned parsed-program gradients once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_grads_for_gpu_execution_results(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic conditioned gradient production batch"
                        .to_string(),
                    context: "batched conditioned program gradient compilation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(
                    self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                        program,
                        &provenance,
                        evidence,
                        EpistemicProbConditionedEvidencePath::Program,
                        "epistemic probabilistic conditioned parsed-program exact gradient",
                    )?,
                );
            }
            Ok(results)
        })
    }

    /// Compile conditioned parsed-program gradients once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_conditioned_program_with_grads_for_gpu_batch_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(
                    self.compile_and_evaluate_conditioned_program_with_grads_with_path(
                        program,
                        &provenance,
                        evidence,
                        EpistemicProbConditionedEvidencePath::Program,
                        "epistemic probabilistic conditioned parsed-program exact gradient",
                    )?,
                );
            }
            Ok(results)
        })
    }

    /// Compile a parsed program and evaluate queries through the existing GPU exact path.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_program_with_accepted_world_view(
        &mut self,
        program: &Program,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            let exact = ExactDdnnfProgram::compile_from_program(program, self.config)?;
            require_gpu_exact_backend(
                &exact,
                "epistemic probabilistic parsed-program exact compile/evaluate",
            )?;
            checked_prob_trace_counter_inc!(self, gpu_exact_program_compiles);
            let result = exact.evaluate()?;
            checked_prob_trace_counter_inc!(self, gpu_exact_query_evaluations);
            checked_prob_trace_counter_inc!(self, gpu_program_exact_query_evaluations);
            checked_prob_trace_counter_inc!(self, gpu_knowledge_compilation_end_to_end_runs);
            checked_prob_trace_counter_inc!(
                self,
                gpu_program_knowledge_compilation_end_to_end_runs
            );
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(result)
        })
    }

    /// Compile a parsed program and evaluate queries after accepted GPU epistemic execution.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_program_with_gpu_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResult> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.compile_and_evaluate_program_with_accepted_world_view(program, &evidence)
    }

    /// Compile and evaluate a parsed program once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_program_for_gpu_execution_results(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic parsed-program production batch"
                        .to_string(),
                    context: "batched parsed-program knowledge compilation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(
                    self.compile_and_evaluate_program_with_accepted_world_view(program, evidence)?,
                );
            }
            Ok(results)
        })
    }

    /// Compile and evaluate a parsed program once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn compile_and_evaluate_program_for_gpu_batch_execution_result(
        &mut self,
        program: &Program,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic batch production",
            )?;

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(
                    self.compile_and_evaluate_program_with_accepted_world_view(program, evidence)?,
                );
            }
            Ok(results)
        })
    }

    /// Encode source through the existing GPU PIR and CNF production path.
    pub fn encode_source_pir_cnf_with_accepted_world_view(
        &mut self,
        source: &str,
        provider: &Arc<CudaKernelProvider>,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let program = parse_program(source)?;
            let base_provenance = extract_from_program(&program)?;
            self.encode_program_pir_cnf_with_base_provenance(
                &program,
                &base_provenance,
                provider,
                evidence,
                EpistemicProbPirCnfPath::Source,
            )
        })
    }

    /// Encode source PIR/CNF after accepted GPU epistemic execution.
    pub fn encode_source_pir_cnf_with_gpu_execution_result(
        &mut self,
        source: &str,
        provider: &Arc<CudaKernelProvider>,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let program = parse_program(source)?;
            let base_provenance = extract_from_program(&program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &base_provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.encode_program_pir_cnf_with_base_provenance(
                &program,
                &base_provenance,
                provider,
                evidence,
                EpistemicProbPirCnfPath::Source,
            )
        })
    }

    /// Encode source PIR/CNF once per accepted GPU epistemic execution result.
    pub fn encode_source_pir_cnf_for_gpu_execution_results(
        &mut self,
        source: &str,
        provider: &Arc<CudaKernelProvider>,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<EpistemicProbPirCnfEvidence>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic source PIR/CNF production batch"
                        .to_string(),
                    context:
                        "batched source PIR/CNF encoding requires at least one accepted GPU result"
                            .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let program = parse_program(source)?;
            let base_provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &base_provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.encode_program_pir_cnf_with_base_provenance(
                    &program,
                    &base_provenance,
                    provider,
                    evidence,
                    EpistemicProbPirCnfPath::Source,
                )?);
            }
            Ok(results)
        })
    }

    /// Encode source PIR/CNF once per accepted split/batch GPU epistemic component.
    pub fn encode_source_pir_cnf_for_gpu_batch_execution_result(
        &mut self,
        source: &str,
        provider: &Arc<CudaKernelProvider>,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<EpistemicProbPirCnfEvidence>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider.as_ref(),
                evidence,
                "epistemic probabilistic source PIR/CNF batch production",
            )?;

            let program = parse_program(source)?;
            let base_provenance = extract_from_program(&program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &base_provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.encode_program_pir_cnf_with_base_provenance(
                    &program,
                    &base_provenance,
                    provider,
                    evidence,
                    EpistemicProbPirCnfPath::Source,
                )?);
            }
            Ok(results)
        })
    }

    /// Encode a parsed program through the existing GPU PIR and CNF production path.
    pub fn encode_program_pir_cnf_with_accepted_world_view(
        &mut self,
        program: &Program,
        provider: &Arc<CudaKernelProvider>,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let base_provenance = extract_from_program(program)?;
            self.encode_program_pir_cnf_with_base_provenance(
                program,
                &base_provenance,
                provider,
                evidence,
                EpistemicProbPirCnfPath::Program,
            )
        })
    }

    /// Encode parsed-program PIR/CNF after accepted GPU epistemic execution.
    pub fn encode_program_pir_cnf_with_gpu_execution_result(
        &mut self,
        program: &Program,
        provider: &Arc<CudaKernelProvider>,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived = assumptions.is_empty();
            let evidence = AcceptedWorldViewEvidence::from_gpu_execution_result(
                provider,
                result,
                assumptions,
            )?;
            let base_provenance = extract_from_program(program)?;
            let filtered_evidence;
            let evidence = if auto_derived {
                filtered_evidence =
                    evidence_with_provenance_backed_assumptions(&evidence, &base_provenance)?;
                &filtered_evidence
            } else {
                &evidence
            };
            self.encode_program_pir_cnf_with_base_provenance(
                program,
                &base_provenance,
                provider,
                evidence,
                EpistemicProbPirCnfPath::Program,
            )
        })
    }

    /// Encode parsed-program PIR/CNF once per accepted GPU epistemic execution result.
    pub fn encode_program_pir_cnf_for_gpu_execution_results(
        &mut self,
        program: &Program,
        provider: &Arc<CudaKernelProvider>,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<EpistemicProbPirCnfEvidence>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic parsed-program PIR/CNF production batch"
                        .to_string(),
                    context:
                        "batched parsed-program PIR/CNF encoding requires at least one accepted GPU result"
                            .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let base_provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (record, evidence) in evidence_records.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if record.assumptions.is_empty() {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &base_provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.encode_program_pir_cnf_with_base_provenance(
                    program,
                    &base_provenance,
                    provider,
                    evidence,
                    EpistemicProbPirCnfPath::Program,
                )?);
            }
            Ok(results)
        })
    }

    /// Encode parsed-program PIR/CNF once per accepted split/batch GPU epistemic component.
    pub fn encode_program_pir_cnf_for_gpu_batch_execution_result(
        &mut self,
        program: &Program,
        provider: &Arc<CudaKernelProvider>,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<EpistemicProbPirCnfEvidence>> {
        epistemic_prob_trace_transaction!(self, {
            let auto_derived_by_component = evidence
                .assumptions_by_component
                .iter()
                .map(|assumptions| assumptions.is_empty())
                .collect::<Vec<_>>();
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider.as_ref(),
                evidence,
                "epistemic probabilistic parsed-program PIR/CNF batch production",
            )?;

            let base_provenance = extract_from_program(program)?;
            let mut results = Vec::with_capacity(accepted.len());
            for (auto_derived, evidence) in auto_derived_by_component.iter().zip(accepted.iter()) {
                let filtered_evidence;
                let evidence = if *auto_derived {
                    filtered_evidence =
                        evidence_with_provenance_backed_assumptions(evidence, &base_provenance)?;
                    &filtered_evidence
                } else {
                    evidence
                };
                results.push(self.encode_program_pir_cnf_with_base_provenance(
                    program,
                    &base_provenance,
                    provider,
                    evidence,
                    EpistemicProbPirCnfPath::Program,
                )?);
            }
            Ok(results)
        })
    }

    /// Evaluate GPU exact query probabilities after accepted world-view evidence was consumed.
    #[cfg(feature = "host-io")]
    pub fn evaluate(
        &mut self,
        program: &ExactDdnnfProgram,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResult> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            require_gpu_exact_backend(program, "epistemic probabilistic exact query evaluation")?;
            let result = program.evaluate()?;
            self.record_gpu_exact_query_evaluation(program)?;
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(result)
        })
    }

    /// Evaluate GPU exact query probabilities after accepted GPU epistemic execution.
    #[cfg(feature = "host-io")]
    pub fn evaluate_with_gpu_execution_result(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResult> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.evaluate(program, &evidence)
    }

    /// Evaluate GPU exact query probabilities once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn evaluate_for_gpu_execution_results(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic query evaluation production batch"
                        .to_string(),
                    context: "batched query evaluation requires at least one accepted GPU result"
                        .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(self.evaluate(program, evidence)?);
            }
            Ok(results)
        })
    }

    /// Evaluate GPU exact query probabilities once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn evaluate_for_gpu_batch_execution_result(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResult>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic query evaluation batch production",
            )?;

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(self.evaluate(program, evidence)?);
            }
            Ok(results)
        })
    }

    /// Evaluate GPU exact gradients after accepted world-view evidence was consumed.
    #[cfg(feature = "host-io")]
    pub fn evaluate_gpu_with_grads(
        &mut self,
        program: &ExactDdnnfProgram,
        evidence: &AcceptedWorldViewEvidence,
    ) -> Result<ExactResultWithGrads> {
        epistemic_prob_trace_transaction!(self, {
            self.require_accepted_evidence(evidence)?;
            let production_events_before = self.trace.checked_gpu_production_path_events()?;
            require_gpu_exact_backend(
                program,
                "epistemic probabilistic exact gradient evaluation",
            )?;
            let result = program.evaluate_gpu_with_grads()?;
            self.record_gpu_exact_gradient_evaluation(program)?;
            self.record_accepted_gpu_production_path_events_since(production_events_before)?;
            self.record_accepted_evidence(evidence)?;
            Ok(result)
        })
    }

    /// Evaluate GPU exact gradients after accepted GPU epistemic execution.
    #[cfg(feature = "host-io")]
    pub fn evaluate_gpu_with_grads_with_gpu_execution_result(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        result: &EpistemicGpuExecutionResult,
        assumptions: Vec<EpistemicAssumption>,
    ) -> Result<ExactResultWithGrads> {
        let evidence =
            AcceptedWorldViewEvidence::from_gpu_execution_result(provider, result, assumptions)?;
        self.evaluate_gpu_with_grads(program, &evidence)
    }

    /// Evaluate GPU exact gradients once per accepted GPU epistemic execution result.
    #[cfg(feature = "host-io")]
    pub fn evaluate_gpu_with_grads_for_gpu_execution_results(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        evidence_records: &[EpistemicProbGpuExecutionEvidence<'_>],
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            if evidence_records.is_empty() {
                return Err(XlogError::UnsupportedEpistemicConstruct {
                    construct: "epistemic probabilistic gradient evaluation production batch"
                        .to_string(),
                    context:
                        "batched gradient evaluation requires at least one accepted GPU result"
                            .to_string(),
                });
            }

            let mut accepted = Vec::with_capacity(evidence_records.len());
            for record in evidence_records {
                accepted.push(AcceptedWorldViewEvidence::from_gpu_execution_result(
                    provider,
                    record.result,
                    record.assumptions.to_vec(),
                )?);
            }

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(self.evaluate_gpu_with_grads(program, evidence)?);
            }
            Ok(results)
        })
    }

    /// Evaluate GPU exact gradients once per accepted split/batch GPU epistemic component.
    #[cfg(feature = "host-io")]
    pub fn evaluate_gpu_with_grads_for_gpu_batch_execution_result(
        &mut self,
        program: &ExactDdnnfProgram,
        provider: &CudaKernelProvider,
        evidence: EpistemicProbGpuBatchExecutionEvidence<'_>,
    ) -> Result<Vec<ExactResultWithGrads>> {
        epistemic_prob_trace_transaction!(self, {
            let accepted = self.accepted_world_views_from_gpu_batch_execution_evidence(
                provider,
                evidence,
                "epistemic probabilistic gradient evaluation batch production",
            )?;

            let mut results = Vec::with_capacity(accepted.len());
            for evidence in &accepted {
                results.push(self.evaluate_gpu_with_grads(program, evidence)?);
            }
            Ok(results)
        })
    }

    fn consume_accepted_evidence(&mut self, evidence: &AcceptedWorldViewEvidence) -> Result<()> {
        self.require_accepted_evidence(evidence)?;
        self.record_accepted_evidence(evidence)?;
        Ok(())
    }

    fn require_accepted_evidence(&self, evidence: &AcceptedWorldViewEvidence) -> Result<()> {
        if evidence.world_count() == 0 {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted world-view evidence".to_string(),
                context: "probabilistic production path requires a non-empty accepted world view"
                    .to_string(),
            });
        }
        if evidence.gpu_epistemic_mode().is_none() {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted world-view evidence".to_string(),
                context: "probabilistic production path requires GPU execution evidence; CPU \
                     world-view evidence is oracle-only"
                    .to_string(),
            });
        }
        Ok(())
    }

    fn record_accepted_evidence(&mut self, evidence: &AcceptedWorldViewEvidence) -> Result<()> {
        self.trace.accepted_world_view_evidence_consumed = Self::checked_trace_counter_add(
            self.trace.accepted_world_view_evidence_consumed,
            1,
            "accepted_world_view_evidence_consumed",
        )?;
        match evidence.gpu_epistemic_mode() {
            Some(EirEpistemicMode::G91) => {
                self.trace.accepted_g91_world_view_evidence_consumed =
                    Self::checked_trace_counter_add(
                        self.trace.accepted_g91_world_view_evidence_consumed,
                        1,
                        "accepted_g91_world_view_evidence_consumed",
                    )?;
            }
            Some(EirEpistemicMode::Faeel) => {
                self.trace.accepted_faeel_world_view_evidence_consumed =
                    Self::checked_trace_counter_add(
                        self.trace.accepted_faeel_world_view_evidence_consumed,
                        1,
                        "accepted_faeel_world_view_evidence_consumed",
                    )?;
            }
            None => {}
        }
        self.trace.accepted_evidence_assumptions_consumed = Self::checked_trace_counter_add(
            self.trace.accepted_evidence_assumptions_consumed,
            evidence.assumption_count() as u64,
            "accepted_evidence_assumptions_consumed",
        )?;
        self.trace
            .accepted_gpu_nonzero_arity_evidence_assumptions_consumed =
            Self::checked_trace_counter_add(
                self.trace
                    .accepted_gpu_nonzero_arity_evidence_assumptions_consumed,
                evidence.nonzero_arity_assumption_count() as u64,
                "accepted_gpu_nonzero_arity_evidence_assumptions_consumed",
            )?;
        if evidence.max_assumption_arity() > u32::MAX as usize {
            return Err(XlogError::ResourceExhausted {
                context: "accepted GPU probability evidence max arity".to_string(),
                estimated_bytes: evidence.max_assumption_arity() as u64,
                budget_bytes: u32::MAX as u64,
            });
        }
        self.trace.accepted_gpu_max_evidence_arity_consumed = self
            .trace
            .accepted_gpu_max_evidence_arity_consumed
            .max(evidence.max_assumption_arity() as u32);
        self.trace.accepted_gpu_tuple_key_column_reads_consumed = Self::checked_trace_counter_add(
            self.trace.accepted_gpu_tuple_key_column_reads_consumed,
            evidence.gpu_tuple_key_column_reads() as u64,
            "accepted_gpu_tuple_key_column_reads_consumed",
        )?;
        self.trace.accepted_gpu_final_tuple_row_filters_consumed = Self::checked_trace_counter_add(
            self.trace.accepted_gpu_final_tuple_row_filters_consumed,
            evidence.gpu_final_tuple_row_filters() as u64,
            "accepted_gpu_final_tuple_row_filters_consumed",
        )?;
        self.trace
            .accepted_gpu_final_tuple_negated_row_filters_consumed =
            Self::checked_trace_counter_add(
                self.trace
                    .accepted_gpu_final_tuple_negated_row_filters_consumed,
                evidence.gpu_final_tuple_negated_row_filters() as u64,
                "accepted_gpu_final_tuple_negated_row_filters_consumed",
            )?;
        self.trace
            .accepted_gpu_row_specific_membership_row_capacity_consumed =
            Self::checked_trace_counter_add(
                self.trace
                    .accepted_gpu_row_specific_membership_row_capacity_consumed,
                evidence.gpu_row_specific_membership_row_capacity() as u64,
                "accepted_gpu_row_specific_membership_row_capacity_consumed",
            )?;
        self.trace
            .accepted_gpu_row_filter_fallback_row_capacity_consumed =
            Self::checked_trace_counter_add(
                self.trace
                    .accepted_gpu_row_filter_fallback_row_capacity_consumed,
                evidence.gpu_row_filter_fallback_row_capacity() as u64,
                "accepted_gpu_row_filter_fallback_row_capacity_consumed",
            )?;
        self.trace
            .accepted_gpu_constraint_relations_checked_consumed = Self::checked_trace_counter_add(
            self.trace
                .accepted_gpu_constraint_relations_checked_consumed,
            evidence.gpu_checked_constraint_relations() as u64,
            "accepted_gpu_constraint_relations_checked_consumed",
        )?;
        self.trace
            .accepted_gpu_constraint_row_count_device_reads_consumed =
            Self::checked_trace_counter_add(
                self.trace
                    .accepted_gpu_constraint_row_count_device_reads_consumed,
                evidence.gpu_constraint_row_count_device_reads() as u64,
                "accepted_gpu_constraint_row_count_device_reads_consumed",
            )?;
        Ok(())
    }

    fn encode_program_pir_cnf_with_base_provenance(
        &mut self,
        program: &Program,
        base_provenance: &Provenance,
        provider: &Arc<CudaKernelProvider>,
        evidence: &AcceptedWorldViewEvidence,
        path: EpistemicProbPirCnfPath,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        self.require_accepted_evidence(evidence)?;
        let conditioned_provenance = if evidence.assumptions().is_empty() {
            None
        } else {
            let conditioned_program = condition_program_with_available_evidence_using_provenance(
                program,
                base_provenance,
                evidence,
            )?;
            Some(extract_from_program(&conditioned_program)?)
        };
        let provenance = conditioned_provenance.as_ref().unwrap_or(base_provenance);
        self.encode_provenance_pir_cnf_with_accepted_world_view(
            provenance, provider, evidence, path,
        )
    }

    fn encode_provenance_pir_cnf_with_accepted_world_view(
        &mut self,
        provenance: &Provenance,
        provider: &Arc<CudaKernelProvider>,
        evidence: &AcceptedWorldViewEvidence,
        path: EpistemicProbPirCnfPath,
    ) -> Result<EpistemicProbPirCnfEvidence> {
        self.require_accepted_evidence(evidence)?;
        let production_events_before = self.trace.checked_gpu_production_path_events()?;
        let roots = production_pir_roots(provenance)?;
        if roots.is_empty() {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted probabilistic PIR/CNF production path".to_string(),
                context: "GPU PIR/CNF evidence requires at least one query, evidence, or probabilistic variable root".to_string(),
            });
        }
        let gpu_pir = GpuPirGraph::from_host(&provenance.pir, provider)?;
        checked_prob_trace_counter_inc!(self, gpu_pir_graph_uploads);
        match path {
            EpistemicProbPirCnfPath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_pir_graph_uploads);
            }
            EpistemicProbPirCnfPath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_pir_graph_uploads);
            }
        }
        let gpu_roots = GpuPirRoots::from_host(&roots, provider)?;
        let encoding = encode_cnf_gpu(&gpu_pir, &gpu_roots, provider)?;
        checked_prob_trace_counter_inc!(self, gpu_cnf_encodes);
        match path {
            EpistemicProbPirCnfPath::Source => {
                checked_prob_trace_counter_inc!(self, gpu_source_cnf_encodes);
            }
            EpistemicProbPirCnfPath::Program => {
                checked_prob_trace_counter_inc!(self, gpu_program_cnf_encodes);
            }
        }
        let pir_cnf_evidence = EpistemicProbPirCnfEvidence {
            pir_nodes: provenance.pir.len(),
            root_count: roots.len(),
            cnf_var_cap: encoding.cnf.var_cap,
            cnf_clause_cap: encoding.cnf.clause_cap,
            cnf_lit_cap: encoding.cnf.lit_cap,
        };
        self.record_accepted_gpu_production_path_events_since(production_events_before)?;
        self.record_accepted_evidence(evidence)?;
        Ok(pir_cnf_evidence)
    }
}

fn require_gpu_exact_backend(program: &ExactDdnnfProgram, construct: &'static str) -> Result<()> {
    if !program.uses_gpu_production_backend() {
        return Err(XlogError::UnsupportedEpistemicConstruct {
            construct: construct.to_string(),
            context:
                "production probability metrics require a compiled GPU-native Decision-DNNF exact/provenance/PIR/CNF backend; \
                      empty roots and count-lift-only evaluation cannot satisfy accepted epistemic probability reuse"
                    .to_string(),
        });
    }
    Ok(())
}

#[cfg(feature = "host-io")]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct EpistemicProbConditionedEvidenceCounts {
    total: usize,
    nonzero_arity: usize,
    max_arity: u32,
    negative: usize,
    know: usize,
    possible: usize,
    not_known: usize,
    not_possible: usize,
}

fn condition_program_with_available_evidence_using_provenance(
    program: &Program,
    base_provenance: &Provenance,
    evidence: &AcceptedWorldViewEvidence,
) -> Result<Program> {
    if evidence.assumptions().is_empty() {
        return Ok(program.clone());
    }

    let mut program = program.clone();
    let mut applied = 0usize;
    for assumption in evidence.assumptions() {
        validate_conditioned_assumption_shape(assumption)?;
        let Some(atom) = conditioned_evidence_atom_for_assumption(base_provenance, assumption)
        else {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted probabilistic PIR/CNF evidence conditioning".to_string(),
                context: format!(
                    "accepted {} PIR/CNF evidence for {}/{} has no provenance formula; partial \
                     world-view evidence cannot satisfy production conditioning metrics",
                    assumption.evidence_literal(),
                    assumption.predicate,
                    assumption.arity
                ),
            });
        };
        program.evidence.push(Evidence {
            atom,
            value: assumption.value,
        });
        applied += 1;
    }

    if applied == 0 {
        return Err(XlogError::UnsupportedEpistemicConstruct {
            construct: "accepted probabilistic PIR/CNF evidence conditioning".to_string(),
            context: "PIR/CNF encoding requires at least one accepted epistemic assumption to match existing probabilistic provenance"
                .to_string(),
        });
    }

    Ok(program)
}

#[cfg(feature = "host-io")]
fn condition_program_with_accepted_evidence_using_provenance(
    program: &Program,
    provenance: &Provenance,
    evidence: &AcceptedWorldViewEvidence,
) -> Result<(Program, EpistemicProbConditionedEvidenceCounts)> {
    if evidence.assumptions().is_empty() {
        return Err(XlogError::UnsupportedEpistemicConstruct {
            construct: "accepted probabilistic evidence conditioning".to_string(),
            context: "conditioned exact path requires at least one accepted epistemic assumption"
                .to_string(),
        });
    }

    let mut counts = EpistemicProbConditionedEvidenceCounts::default();
    for assumption in evidence.assumptions() {
        validate_conditioned_assumption_shape(assumption)?;
        record_conditioned_assumption_counts(&mut counts, assumption);
    }
    let mut program = program.clone();
    for assumption in evidence.assumptions() {
        let Some(atom) = conditioned_evidence_atom_for_assumption(provenance, assumption) else {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted probabilistic evidence conditioning".to_string(),
                context: format!(
                    "accepted {} exact evidence for {}/{} has no provenance formula; \
                     vacuous evidence cannot satisfy production conditioning metrics",
                    assumption.evidence_literal(),
                    assumption.predicate,
                    assumption.arity
                ),
            });
        };
        program.evidence.push(Evidence {
            atom,
            value: assumption.value,
        });
    }

    Ok((program, counts))
}

fn validate_conditioned_assumption_shape(assumption: &EpistemicAssumption) -> Result<()> {
    if assumption.arity == 0 {
        if !assumption.terms.is_empty() {
            return Err(XlogError::UnsupportedEpistemicConstruct {
                construct: "accepted probabilistic evidence conditioning".to_string(),
                context: format!(
                    "zero-arity exact evidence must not carry tuple terms, got {}/{}",
                    assumption.predicate, assumption.arity
                ),
            });
        }
    } else if assumption.terms.len() != assumption.arity {
        return Err(XlogError::UnsupportedEpistemicConstruct {
            construct: "accepted probabilistic evidence conditioning".to_string(),
            context: format!(
                "nonzero exact evidence conditioning requires {} concrete tuple terms, got {} for {}/{}",
                assumption.arity,
                assumption.terms.len(),
                assumption.predicate,
                assumption.arity
            ),
        });
    }
    Ok(())
}

#[cfg(feature = "host-io")]
fn record_conditioned_assumption_counts(
    counts: &mut EpistemicProbConditionedEvidenceCounts,
    assumption: &EpistemicAssumption,
) {
    counts.total += 1;
    if assumption.arity > 0 {
        counts.nonzero_arity += 1;
        counts.max_arity = counts.max_arity.max(assumption.arity as u32);
    }
    if !assumption.value {
        counts.negative += 1;
    }
    match (assumption.kind, assumption.value) {
        (EpistemicAssumptionKind::Know, true) => counts.know += 1,
        (EpistemicAssumptionKind::Possible, true) => counts.possible += 1,
        (EpistemicAssumptionKind::Know, false) => counts.not_known += 1,
        (EpistemicAssumptionKind::Possible, false) => counts.not_possible += 1,
    }
}

fn evidence_term_variants(term: &EpistemicEvidenceTerm) -> Vec<(Value, Term)> {
    match term {
        EpistemicEvidenceTerm::Integer(value) => vec![(Value::I64(*value), Term::Integer(*value))],
        EpistemicEvidenceTerm::String(value) => {
            let symbol_id = symbol::intern(value);
            vec![
                (Value::String(value.clone()), Term::String(value.clone())),
                (Value::Symbol(symbol_id), Term::Symbol(symbol_id)),
            ]
        }
        EpistemicEvidenceTerm::Symbol(value) => {
            let string_value = symbol::resolve(*value);
            vec![
                (Value::Symbol(*value), Term::Symbol(*value)),
                (
                    Value::String(string_value.clone()),
                    Term::String(string_value),
                ),
            ]
        }
    }
}

fn conditioned_evidence_atom_for_assumption(
    provenance: &Provenance,
    assumption: &EpistemicAssumption,
) -> Option<Atom> {
    if assumption.terms.is_empty() {
        if provenance
            .query_formula(&assumption.predicate, &[])
            .is_some()
        {
            return Some(Atom {
                predicate: assumption.predicate.clone(),
                terms: Vec::new(),
            });
        }
        return None;
    }

    let variants = assumption
        .terms
        .iter()
        .map(evidence_term_variants)
        .collect::<Vec<_>>();
    let mut values = Vec::with_capacity(variants.len());
    let mut terms = Vec::with_capacity(variants.len());
    conditioned_evidence_atom_from_variants(
        provenance,
        assumption,
        &variants,
        0,
        &mut values,
        &mut terms,
    )
}

fn conditioned_evidence_atom_from_variants(
    provenance: &Provenance,
    assumption: &EpistemicAssumption,
    variants: &[Vec<(Value, Term)>],
    index: usize,
    values: &mut Vec<Value>,
    terms: &mut Vec<Term>,
) -> Option<Atom> {
    if index == variants.len() {
        if provenance
            .query_formula(&assumption.predicate, values)
            .is_some()
        {
            return Some(Atom {
                predicate: assumption.predicate.clone(),
                terms: terms.clone(),
            });
        }
        return None;
    }

    for (value, term) in &variants[index] {
        values.push(value.clone());
        terms.push(term.clone());
        if let Some(atom) = conditioned_evidence_atom_from_variants(
            provenance,
            assumption,
            variants,
            index + 1,
            values,
            terms,
        ) {
            return Some(atom);
        }
        values.pop();
        terms.pop();
    }
    None
}

fn assumption_has_provenance_formula(
    provenance: &Provenance,
    assumption: &EpistemicAssumption,
) -> bool {
    conditioned_evidence_atom_for_assumption(provenance, assumption).is_some()
}

fn evidence_with_provenance_backed_assumptions(
    evidence: &AcceptedWorldViewEvidence,
    provenance: &Provenance,
) -> Result<AcceptedWorldViewEvidence> {
    let assumptions = evidence
        .assumptions()
        .iter()
        .filter(|assumption| assumption_has_provenance_formula(provenance, assumption))
        .cloned()
        .collect::<Vec<_>>();
    if assumptions.is_empty() {
        return Err(XlogError::UnsupportedEpistemicConstruct {
            construct: "accepted probabilistic PIR/CNF evidence conditioning".to_string(),
            context: "PIR/CNF encoding requires at least one accepted epistemic assumption to match existing probabilistic provenance"
                .to_string(),
        });
    }
    Ok(evidence.with_assumptions(assumptions))
}

fn production_pir_roots(provenance: &Provenance) -> Result<Vec<PirNodeId>> {
    let mut roots = BTreeSet::new();

    for (atom, value) in &provenance.evidence {
        if let Some(id) = provenance.query_formula(&atom.predicate, &atom.args) {
            roots.insert(id);
        } else if *value {
            return Err(XlogError::Execution(format!(
                "Exact inference error: evidence atom is never derivable: {}",
                atom.predicate
            )));
        }
    }

    for atom in &provenance.queries {
        if let Some(id) = provenance.query_formula(&atom.predicate, &atom.args) {
            roots.insert(id);
        }
    }

    for (idx, node) in provenance.pir.nodes().iter().enumerate() {
        if matches!(
            node,
            PirNode::Decision { .. } | PirNode::Lit { .. } | PirNode::NegLit { .. }
        ) {
            roots.insert(PirNodeId::from_u32(idx as u32));
        }
    }

    Ok(roots.into_iter().collect())
}

#[cfg(all(test, feature = "host-io"))]
mod prepared_conditioned_poison_tests {
    use super::*;

    #[test]
    fn rollback_failure_permanently_invalidates_prepared_state_and_all_clones() {
        let _gpu_guard = crate::test_gpu_lock::lock();
        match xlog_cuda::CudaDevice::new(0) {
            Ok(_) => {}
            Err(error) if std::env::var("XLOG_REQUIRE_CUDA").as_deref() == Ok("1") => {
                panic!("XLOG_REQUIRE_CUDA=1 but CUDA runtime initialization failed: {error}")
            }
            Err(error) => {
                eprintln!("Skipping test: CUDA runtime unavailable: {error}");
                return;
            }
        }

        let exact = ExactDdnnfProgram::compile_source(
            "0.5::rain().\n\
             query(rain()).\n",
        )
        .expect("compile prepared exact program");
        let prepared =
            PreparedConditionedProgram::new(exact, EpistemicProbProductionTrace::default())
                .expect("prepare exact circuit");
        let rain_var = prepared
            .prob_var_map()
            .expect("read fact map before injected failure")
            .iter()
            .enumerate()
            .find_map(|(var, info)| {
                matches!(info, ProbVarInfo::Fact { atom, .. } if atom.predicate == "rain")
                    .then_some(var as u32)
            })
            .expect("rain fact has a CNF variable");
        let clone_before_failure = prepared.clone();

        let error = prepared
            .set_fact_probabilities_with_device_failures(
                &BTreeMap::from([(rain_var, 0.9)]),
                Some(1),
                Some(1),
            )
            .expect_err("rollback failure must be reported");
        let message = error.to_string();
        assert!(message.contains("device update failed"), "{message}");
        assert!(message.contains("rollback also failed"), "{message}");
        assert!(
            message.contains("injected fact probability device-write failure"),
            "{message}"
        );
        assert!(
            message.contains("injected fact probability rollback failure"),
            "{message}"
        );

        let clone_after_failure = prepared.clone();
        for (operation, result) in [
            ("evaluate", prepared.evaluate().map(|_| ())),
            ("gradient", prepared.evaluate_with_grads().map(|_| ())),
            ("prob_var_map", prepared.prob_var_map().map(|_| ())),
            (
                "set_fact_probabilities",
                prepared.set_fact_probabilities(&BTreeMap::from([(rain_var, 0.4)])),
            ),
            (
                "clone-before evaluate",
                clone_before_failure.evaluate().map(|_| ()),
            ),
            (
                "clone-before prob_var_map",
                clone_before_failure.prob_var_map().map(|_| ()),
            ),
            (
                "clone-after gradient",
                clone_after_failure.evaluate_with_grads().map(|_| ()),
            ),
            (
                "clone-after set",
                clone_after_failure.set_fact_probabilities(&BTreeMap::from([(rain_var, 0.4)])),
            ),
        ] {
            let later = result
                .err()
                .unwrap_or_else(|| panic!("{operation} reused poisoned state"));
            let later_message = later.to_string();
            assert!(
                later_message.contains("permanently invalid"),
                "{operation}: {later_message}"
            );
            assert!(
                later_message.contains("rollback also failed"),
                "{operation}: {later_message}"
            );
        }
    }
}