1use pyo3::exceptions::{PyRuntimeError, PyValueError};
11use pyo3::prelude::*;
12use pyo3::types::PyDict;
13
14#[cfg(feature = "host-io")]
15use xlog_core::symbol;
16use xlog_core::{ScalarType, Schema};
17use xlog_logic::ast::Term;
18use xlog_prob::exact::ExactDdnnfProgram;
19#[cfg(feature = "host-io")]
20use xlog_prob::exact::{ExactResultWithGrads, QueryProbability};
21use xlog_prob::mc::{McEvalConfig, McEvalOverrides, McProgram, McSamplingMethod};
22use xlog_prob::neural_fast_path::GpuWeightSlots;
23
24use super::neural_registry::NeuralPredicateInfo;
25use super::{
26 dlpack_capsule_from_tensor, enforce_call_memory_limit, pack_query_proof_traces,
27 pack_rule_provenance, provider_memory_stats, types, CompiledProgram, EpochStats, EvalResult,
28 McDeviceEvalResult, TrainingHistory,
29};
30
31pub(crate) struct CachedCircuit {
40 pub(crate) program: ExactDdnnfProgram,
42
43 pub(crate) slots: GpuWeightSlots,
45
46 pub(crate) target_domain: Vec<String>,
48}
49
50#[derive(Debug, Clone)]
51pub(crate) enum InputSource {
52 QueryArg(usize),
53 ImplicitSlot(usize),
54 DomainRow(usize),
58 ConstDummy,
62}
63
64#[derive(Debug, Clone)]
65pub(crate) struct NeuralGroup {
66 pub(crate) info: NeuralPredicateInfo,
67 pub(crate) input_source: InputSource,
68 pub(crate) ground_const: Option<Term>,
73 #[cfg(feature = "host-io")]
74 pub(crate) output_var: Option<String>,
75}
76
77#[derive(Debug, Clone)]
83pub(crate) struct HardFilter {
84 pub(crate) relation: String,
86 pub(crate) arg_head_positions: Vec<usize>,
91}
92
93#[derive(Debug, Clone)]
100pub(crate) struct JoinPlan {
101 pub(crate) relations: Vec<String>,
104 pub(crate) head_domain: Vec<String>,
108}
109
110#[derive(Debug, Clone)]
111pub(crate) enum QuerySignature {
112 Boolean {
113 groups: Vec<NeuralGroup>,
114 hard_filters: Vec<HardFilter>,
115 },
116 Targeted {
117 target_position: usize,
118 groups: Vec<NeuralGroup>,
119 hard_filters: Vec<HardFilter>,
120 join: Option<JoinPlan>,
124 },
125}
126
127impl QuerySignature {
128 pub(crate) fn groups(&self) -> &[NeuralGroup] {
129 match self {
130 QuerySignature::Boolean { groups, .. } | QuerySignature::Targeted { groups, .. } => {
131 groups
132 }
133 }
134 }
135
136 pub(crate) fn hard_filters(&self) -> &[HardFilter] {
137 match self {
138 QuerySignature::Boolean { hard_filters, .. }
139 | QuerySignature::Targeted { hard_filters, .. } => hard_filters,
140 }
141 }
142
143 pub(crate) fn join(&self) -> Option<&JoinPlan> {
145 match self {
146 QuerySignature::Targeted { join, .. } => join.as_ref(),
147 QuerySignature::Boolean { .. } => None,
148 }
149 }
150}
151
152pub(crate) enum CompiledProbProgram {
153 Exact(Box<ExactDdnnfProgram>),
154 Mc(Box<McProgram>),
155}
156
157impl CompiledProbProgram {
158 #[cfg(feature = "host-io")]
159 pub(crate) fn num_vars(&self) -> usize {
160 match self {
161 Self::Exact(p) => p.num_vars(),
162 Self::Mc(p) => p.num_vars(),
163 }
164 }
165}
166
167#[cfg(feature = "host-io")]
172pub(crate) fn atom_to_string(atom: &xlog_prob::provenance::GroundAtom) -> String {
173 use xlog_prob::provenance::Value;
174
175 if atom.args.is_empty() {
176 return format!("{}()", atom.predicate);
177 }
178
179 let mut s = String::new();
180 s.push_str(&atom.predicate);
181 s.push('(');
182 for (i, arg) in atom.args.iter().enumerate() {
183 if i != 0 {
184 s.push_str(", ");
185 }
186 match arg {
187 Value::I64(v) => s.push_str(&v.to_string()),
188 Value::F64(bits) => s.push_str(&f64::from_bits(*bits).to_string()),
189 Value::Symbol(sym) => {
190 s.push_str(&symbol::resolve_checked(*sym).unwrap_or_else(|| format!("sym#{}", sym)))
191 }
192 Value::String(v) => s.push_str(v),
193 }
194 }
195 s.push(')');
196 s
197}
198
199fn resolve_mc_eval_config(
200 directives: &xlog_logic::ast::Directives,
201 samples: Option<usize>,
202 seed: Option<u64>,
203 confidence: Option<f64>,
204 max_nonmonotone_iterations: Option<usize>,
205 sampling_method: Option<McSamplingMethod>,
206 allow_cpu_oracle_fallback: bool,
207) -> xlog_core::Result<McEvalConfig> {
208 let mut cfg = McEvalConfig::from_directives(directives)?;
209 cfg.apply_overrides(McEvalOverrides {
210 samples,
211 seed,
212 confidence,
213 max_nonmonotone_iterations,
214 sampling_method,
215 allow_cpu_oracle_fallback: Some(allow_cpu_oracle_fallback),
216 })?;
217 Ok(cfg)
218}
219
220impl CompiledProgram {
225 pub(crate) fn parse_sampling_method(s: Option<String>) -> PyResult<Option<McSamplingMethod>> {
226 match s.as_deref() {
227 None => Ok(None),
228 Some("rejection") => Ok(Some(McSamplingMethod::Rejection)),
229 Some("evidence_clamping") => Ok(Some(McSamplingMethod::EvidenceClamping)),
230 Some(other) => Err(PyValueError::new_err(format!(
231 "Unknown sampling_method '{}'. Use 'rejection' or 'evidence_clamping'.",
232 other
233 ))),
234 }
235 }
236
237 pub(crate) fn evaluate_query_probability(&self, query: &str) -> PyResult<f64> {
239 let probs = self.evaluate_query_probabilities(&[query.to_string()])?;
240 probs
241 .into_iter()
242 .next()
243 .ok_or_else(|| PyRuntimeError::new_err("Query evaluation returned no results"))
244 }
245
246 pub(crate) fn evaluate_query_probabilities(&self, queries: &[String]) -> PyResult<Vec<f64>> {
248 #[cfg(not(feature = "host-io"))]
249 {
250 let _ = queries;
251 return Err(types::host_io_disabled_pyerr());
252 }
253
254 #[cfg(feature = "host-io")]
255 {
256 let mut source_with_queries = self._source.clone();
258 for query in queries {
259 source_with_queries.push_str(&format!("\nquery({}).", query));
260 }
261
262 let result: Vec<QueryProbability> = match self._prob_engine {
264 xlog_logic::ast::ProbEngine::ExactDdnnf => {
265 let program = ExactDdnnfProgram::compile_source_with_gpu(
266 &source_with_queries,
267 self._gpu_config,
268 )
269 .map_err(|e| types::gpu_err("Query compilation error", e))?;
270
271 program
272 .evaluate()
273 .map_err(|e| types::gpu_err("Query evaluation error", e))?
274 .query_probs
275 }
276 xlog_logic::ast::ProbEngine::Mc => {
277 let program =
278 McProgram::compile_source_with_gpu(&source_with_queries, self._gpu_config)
279 .map_err(|e| types::gpu_err("Query compilation error", e))?;
280
281 let cfg = resolve_mc_eval_config(
282 &self.ast.directives,
283 None,
284 None,
285 None,
286 None,
287 None,
288 false,
289 )
290 .map_err(types::xlog_err)?;
291 program
292 .evaluate(cfg)
293 .map_err(|e| types::gpu_err("Query evaluation error", e))?
294 .query_estimates
295 .into_iter()
296 .map(|e| QueryProbability {
297 atom: e.atom,
298 prob: e.prob,
299 log_prob: e.log_prob,
300 })
301 .collect()
302 }
303 };
304
305 let probs: Vec<f64> = result.iter().map(|qp| qp.prob).collect();
308
309 if probs.len() != queries.len() {
310 return Err(PyRuntimeError::new_err(format!(
311 "Expected {} query results, got {}",
312 queries.len(),
313 probs.len()
314 )));
315 }
316
317 Ok(probs)
318 }
319 }
320
321 #[cfg(feature = "host-io")]
322 fn pack_result_probs(
323 &self,
324 py: Python<'_>,
325 query_probs: Vec<QueryProbability>,
326 log_z_e: f64,
327 ) -> PyResult<EvalResult> {
328 let mut atoms: Vec<String> = Vec::with_capacity(query_probs.len());
329 let mut probs: Vec<f64> = Vec::with_capacity(query_probs.len());
330 let mut log_probs: Vec<f64> = Vec::with_capacity(query_probs.len());
331
332 for q in query_probs {
333 atoms.push(atom_to_string(&q.atom));
334 probs.push(q.prob);
335 log_probs.push(q.log_prob);
336 }
337
338 let schema = Schema::new(vec![("col0".to_string(), ScalarType::F64)]);
339 let prob_buf = self
340 .output_provider
341 .create_buffer_from_slice::<f64>(&probs, schema.clone())
342 .map_err(types::xlog_err)?;
343 let log_prob_buf = self
344 .output_provider
345 .create_buffer_from_slice::<f64>(&log_probs, schema)
346 .map_err(types::xlog_err)?;
347
348 let prob_tensor = self
349 .output_provider
350 .to_dlpack_table(prob_buf)
351 .column(0)
352 .map_err(types::xlog_err)?;
353 let log_prob_tensor = self
354 .output_provider
355 .to_dlpack_table(log_prob_buf)
356 .column(0)
357 .map_err(types::xlog_err)?;
358
359 Ok(EvalResult {
360 atoms,
361 prob: dlpack_capsule_from_tensor(py, prob_tensor)?,
362 log_prob: dlpack_capsule_from_tensor(py, log_prob_tensor)?,
363 num_vars: self.program.num_vars(),
364 log_z_e: Some(log_z_e),
365 grad_true: None,
366 grad_false: None,
367 approx: false,
368 stderr: None,
369 ci_low: None,
370 ci_high: None,
371 samples: None,
372 evidence_samples: None,
373 seed: None,
374 confidence: None,
375 nonmonotone_semantics: None,
376 nonmonotone_sccs: None,
377 nonmonotone_cycles: None,
378 nonmonotone_iteration_limit_hits: None,
379 sampling_method: None,
380 mc_engine: None,
381 })
382 }
383
384 #[cfg(feature = "host-io")]
385 fn pack_result_with_grads(
386 &self,
387 py: Python<'_>,
388 result: ExactResultWithGrads,
389 ) -> PyResult<EvalResult> {
390 let mut atoms: Vec<String> = Vec::with_capacity(result.query_grads.len());
391 let mut probs: Vec<f64> = Vec::with_capacity(result.query_grads.len());
392 let mut log_probs: Vec<f64> = Vec::with_capacity(result.query_grads.len());
393
394 let mut grad_true_caps: Vec<Py<PyAny>> = Vec::with_capacity(result.query_grads.len());
395 let mut grad_false_caps: Vec<Py<PyAny>> = Vec::with_capacity(result.query_grads.len());
396
397 let schema = Schema::new(vec![("col0".to_string(), ScalarType::F64)]);
398
399 let num_vars = self.program.num_vars();
400 let log_z_e = result.log_z_e;
401 for q in result.query_grads {
402 atoms.push(atom_to_string(&q.atom));
403 probs.push(q.prob);
404 log_probs.push(q.log_prob);
405
406 let grad_true_buf = self
407 .output_provider
408 .create_buffer_from_slice::<f64>(&q.grad_true, schema.clone())
409 .map_err(types::xlog_err)?;
410 let grad_false_buf = self
411 .output_provider
412 .create_buffer_from_slice::<f64>(&q.grad_false, schema.clone())
413 .map_err(types::xlog_err)?;
414
415 let grad_true_tensor = self
416 .output_provider
417 .to_dlpack_table(grad_true_buf)
418 .column(0)
419 .map_err(types::xlog_err)?;
420 let grad_false_tensor = self
421 .output_provider
422 .to_dlpack_table(grad_false_buf)
423 .column(0)
424 .map_err(types::xlog_err)?;
425
426 grad_true_caps.push(dlpack_capsule_from_tensor(py, grad_true_tensor)?);
427 grad_false_caps.push(dlpack_capsule_from_tensor(py, grad_false_tensor)?);
428 }
429
430 let prob_buf = self
431 .output_provider
432 .create_buffer_from_slice::<f64>(&probs, schema.clone())
433 .map_err(types::xlog_err)?;
434 let log_prob_buf = self
435 .output_provider
436 .create_buffer_from_slice::<f64>(&log_probs, schema)
437 .map_err(types::xlog_err)?;
438
439 let prob_tensor = self
440 .output_provider
441 .to_dlpack_table(prob_buf)
442 .column(0)
443 .map_err(types::xlog_err)?;
444 let log_prob_tensor = self
445 .output_provider
446 .to_dlpack_table(log_prob_buf)
447 .column(0)
448 .map_err(types::xlog_err)?;
449
450 Ok(EvalResult {
451 atoms,
452 prob: dlpack_capsule_from_tensor(py, prob_tensor)?,
453 log_prob: dlpack_capsule_from_tensor(py, log_prob_tensor)?,
454 num_vars,
455 log_z_e: Some(log_z_e),
456 grad_true: Some(grad_true_caps),
457 grad_false: Some(grad_false_caps),
458 approx: false,
459 stderr: None,
460 ci_low: None,
461 ci_high: None,
462 samples: None,
463 evidence_samples: None,
464 seed: None,
465 confidence: None,
466 nonmonotone_semantics: None,
467 nonmonotone_sccs: None,
468 nonmonotone_cycles: None,
469 nonmonotone_iteration_limit_hits: None,
470 sampling_method: None,
471 mc_engine: None,
472 })
473 }
474
475 #[cfg(feature = "host-io")]
476 fn pack_result_mc(
477 &self,
478 py: Python<'_>,
479 result: xlog_prob::mc::McResult,
480 ) -> PyResult<EvalResult> {
481 let mut atoms: Vec<String> = Vec::with_capacity(result.query_estimates.len());
482 let mut probs: Vec<f64> = Vec::with_capacity(result.query_estimates.len());
483 let mut log_probs: Vec<f64> = Vec::with_capacity(result.query_estimates.len());
484 let mut stderrs: Vec<f64> = Vec::with_capacity(result.query_estimates.len());
485 let mut ci_lows: Vec<f64> = Vec::with_capacity(result.query_estimates.len());
486 let mut ci_highs: Vec<f64> = Vec::with_capacity(result.query_estimates.len());
487
488 for q in &result.query_estimates {
489 atoms.push(atom_to_string(&q.atom));
490 probs.push(q.prob);
491 log_probs.push(q.log_prob);
492 stderrs.push(q.stderr);
493 ci_lows.push(q.ci_low);
494 ci_highs.push(q.ci_high);
495 }
496
497 let schema = Schema::new(vec![("col0".to_string(), ScalarType::F64)]);
498 let prob_buf = self
499 .output_provider
500 .create_buffer_from_slice::<f64>(&probs, schema.clone())
501 .map_err(types::xlog_err)?;
502 let log_prob_buf = self
503 .output_provider
504 .create_buffer_from_slice::<f64>(&log_probs, schema.clone())
505 .map_err(types::xlog_err)?;
506 let stderr_buf = self
507 .output_provider
508 .create_buffer_from_slice::<f64>(&stderrs, schema.clone())
509 .map_err(types::xlog_err)?;
510 let ci_low_buf = self
511 .output_provider
512 .create_buffer_from_slice::<f64>(&ci_lows, schema.clone())
513 .map_err(types::xlog_err)?;
514 let ci_high_buf = self
515 .output_provider
516 .create_buffer_from_slice::<f64>(&ci_highs, schema)
517 .map_err(types::xlog_err)?;
518
519 let prob_tensor = self
520 .output_provider
521 .to_dlpack_table(prob_buf)
522 .column(0)
523 .map_err(types::xlog_err)?;
524 let log_prob_tensor = self
525 .output_provider
526 .to_dlpack_table(log_prob_buf)
527 .column(0)
528 .map_err(types::xlog_err)?;
529 let stderr_tensor = self
530 .output_provider
531 .to_dlpack_table(stderr_buf)
532 .column(0)
533 .map_err(types::xlog_err)?;
534 let ci_low_tensor = self
535 .output_provider
536 .to_dlpack_table(ci_low_buf)
537 .column(0)
538 .map_err(types::xlog_err)?;
539 let ci_high_tensor = self
540 .output_provider
541 .to_dlpack_table(ci_high_buf)
542 .column(0)
543 .map_err(types::xlog_err)?;
544
545 Ok(EvalResult {
546 atoms,
547 prob: dlpack_capsule_from_tensor(py, prob_tensor)?,
548 log_prob: dlpack_capsule_from_tensor(py, log_prob_tensor)?,
549 num_vars: self.program.num_vars(),
550 log_z_e: None,
551 grad_true: None,
552 grad_false: None,
553 approx: true,
554 stderr: Some(dlpack_capsule_from_tensor(py, stderr_tensor)?),
555 ci_low: Some(dlpack_capsule_from_tensor(py, ci_low_tensor)?),
556 ci_high: Some(dlpack_capsule_from_tensor(py, ci_high_tensor)?),
557 samples: Some(result.total_samples),
558 evidence_samples: Some(result.evidence_samples),
559 seed: Some(result.seed),
560 confidence: Some(result.confidence),
561 nonmonotone_semantics: Some(xlog_prob::mc::NONMONOTONE_SEMANTICS.to_string()),
562 nonmonotone_sccs: Some(result.nonmonotone_sccs),
563 nonmonotone_cycles: Some(result.nonmonotone_cycles),
564 nonmonotone_iteration_limit_hits: Some(result.nonmonotone_iteration_limit_hits),
565 sampling_method: Some(match result.sampling_method {
566 McSamplingMethod::Rejection => "rejection".to_string(),
567 McSamplingMethod::EvidenceClamping => "evidence_clamping".to_string(),
568 }),
569 mc_engine: Some(result.engine.as_str().to_string()),
570 })
571 }
572}
573
574#[pymethods]
579impl CompiledProgram {
580 #[pyo3(signature = (return_grads=false, samples=None, seed=None, confidence=None, max_nonmonotone_iterations=None, sampling_method=None, memory_mb=None, allow_cpu_oracle=false))]
581 #[expect(
582 clippy::too_many_arguments,
583 reason = "the Python evaluation API exposes each documented Monte Carlo override by name"
584 )]
585 pub fn evaluate(
586 &self,
587 py: Python<'_>,
588 return_grads: bool,
589 samples: Option<usize>,
590 seed: Option<u64>,
591 confidence: Option<f64>,
592 max_nonmonotone_iterations: Option<usize>,
593 sampling_method: Option<String>,
594 memory_mb: Option<u64>,
595 allow_cpu_oracle: bool,
596 ) -> PyResult<EvalResult> {
597 enforce_call_memory_limit(&self.output_provider, memory_mb)?;
598 match &self.program {
599 CompiledProbProgram::Exact(_program) => {
600 if samples.is_some()
601 || seed.is_some()
602 || confidence.is_some()
603 || max_nonmonotone_iterations.is_some()
604 || sampling_method.is_some()
605 || allow_cpu_oracle
606 {
607 return Err(PyValueError::new_err(
608 "samples, seed, confidence, max_nonmonotone_iterations, \
609 sampling_method, and allow_cpu_oracle are only supported for \
610 prob_engine='mc'",
611 ));
612 }
613 #[cfg(feature = "host-io")]
614 {
615 if return_grads {
616 let result = py
617 .detach(|| _program.evaluate_gpu_with_grads())
618 .map_err(types::xlog_err)?;
619 self.pack_result_with_grads(py, result)
620 } else {
621 let result = py.detach(|| _program.evaluate()).map_err(types::xlog_err)?;
622 self.pack_result_probs(py, result.query_probs, result.log_z_e)
623 }
624 }
625 #[cfg(not(feature = "host-io"))]
626 {
627 let _ = return_grads;
628 Err(types::host_io_disabled_pyerr())
629 }
630 }
631 CompiledProbProgram::Mc(_program) => {
632 if return_grads {
633 return Err(PyValueError::new_err(
634 "MC inference does not support gradients (return_grads must be false)",
635 ));
636 }
637
638 let cfg = resolve_mc_eval_config(
639 &self.ast.directives,
640 samples,
641 seed,
642 confidence,
643 max_nonmonotone_iterations,
644 Self::parse_sampling_method(sampling_method)?,
645 allow_cpu_oracle,
646 )
647 .map_err(types::xlog_err)?;
648 #[cfg(feature = "host-io")]
649 {
650 let result = py
651 .detach(|| _program.evaluate(cfg))
652 .map_err(types::xlog_err)?;
653 self.pack_result_mc(py, result)
654 }
655 #[cfg(not(feature = "host-io"))]
656 {
657 let _ = cfg;
658 Err(types::host_io_disabled_pyerr())
659 }
660 }
661 }
662 }
663
664 pub fn prob_var_map(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
689 #[cfg(feature = "host-io")]
690 {
691 use xlog_prob::exact::ProbVarInfo;
692
693 let entries = match &self.program {
694 CompiledProbProgram::Exact(p) => {
695 if p.uses_gpu_native_count_lift() {
696 return Err(PyValueError::new_err(
697 "prob_var_map is unavailable: this program was compiled through \
698 the GPU count-lift fast path (a count aggregate without evidence \
699 or annotated disjunctions), which evaluates the aggregate with a \
700 dedicated kernel and never builds a CNF encoding. There is no \
701 CNF-variable-to-fact map to report. This does not mean the \
702 program has no probabilistic facts.",
703 ));
704 }
705 p.prob_var_map()
706 }
707 CompiledProbProgram::Mc(_) => {
708 return Err(PyValueError::new_err(
709 "prob_var_map is only available for the exact engine",
710 ))
711 }
712 };
713
714 let mut out: Vec<Py<PyAny>> = Vec::with_capacity(entries.len());
715 for entry in entries {
716 let d = PyDict::new(py);
717 match entry {
718 ProbVarInfo::Fact { atom, prob } => {
719 d.set_item("kind", "fact")?;
720 d.set_item("atom", atom_to_string(&atom))?;
721 d.set_item("prob", prob)?;
722 }
723 ProbVarInfo::Choice {
724 choices,
725 choice_index,
726 prob,
727 } => {
728 d.set_item("kind", "choice")?;
729 d.set_item(
730 "atoms",
731 choices
732 .iter()
733 .map(|(a, _)| atom_to_string(a))
734 .collect::<Vec<_>>(),
735 )?;
736 d.set_item(
740 "probs",
741 choices.iter().map(|(_, p)| *p).collect::<Vec<f64>>(),
742 )?;
743 d.set_item("choice_index", choice_index)?;
744 d.set_item("prob", prob)?;
750 }
751 ProbVarInfo::Other => {
752 d.set_item("kind", "other")?;
753 }
754 }
755 out.push(d.into());
756 }
757 Ok(out)
758 }
759 #[cfg(not(feature = "host-io"))]
760 {
761 let _ = py;
762 Err(types::host_io_disabled_pyerr())
763 }
764 }
765
766 #[pyo3(signature = (samples=None, seed=None, confidence=None, max_nonmonotone_iterations=None, sampling_method=None, memory_mb=None))]
771 #[expect(
772 clippy::too_many_arguments,
773 reason = "the GPU-native Python evaluation API exposes each documented override by name"
774 )]
775 pub fn evaluate_device(
776 &self,
777 py: Python<'_>,
778 samples: Option<usize>,
779 seed: Option<u64>,
780 confidence: Option<f64>,
781 max_nonmonotone_iterations: Option<usize>,
782 sampling_method: Option<String>,
783 memory_mb: Option<u64>,
784 ) -> PyResult<McDeviceEvalResult> {
785 enforce_call_memory_limit(&self.output_provider, memory_mb)?;
786 let (
787 query_counts,
788 evidence_count,
789 total_samples,
790 seed,
791 confidence,
792 nonmonotone_sccs,
793 nonmonotone_cycles,
794 nonmonotone_iteration_limit_hits,
795 sampling_method_val,
796 no_host,
797 ) = match &self.program {
798 CompiledProbProgram::Mc(program) => {
799 let cfg = resolve_mc_eval_config(
800 &self.ast.directives,
801 samples,
802 seed,
803 confidence,
804 max_nonmonotone_iterations,
805 Self::parse_sampling_method(sampling_method)?,
806 false,
807 )
808 .map_err(types::xlog_err)?;
809
810 let provider = self.output_provider.clone();
811 let result = py
812 .detach(|| program.evaluate_gpu_device_with_provider(cfg, provider))
813 .map_err(types::xlog_err)?;
814
815 (
816 result.query_counts,
817 result.evidence_count,
818 result.total_samples,
819 result.seed,
820 result.confidence,
821 result.nonmonotone_sccs,
822 result.nonmonotone_cycles,
823 result.nonmonotone_iteration_limit_hits,
824 result.sampling_method,
825 result.no_host,
826 )
827 }
828 _ => {
829 return Err(PyValueError::new_err(
830 "evaluate_device is only supported for prob_engine='mc'",
831 ))
832 }
833 };
834
835 let schema_i32 = Schema::new(vec![("col0".to_string(), ScalarType::I32)]);
838
839 let make_count_tensor =
840 |counts: xlog_cuda::memory::TrackedCudaSlice<u32>, rows: u64| -> PyResult<Py<PyAny>> {
841 let rows_u32 = u32::try_from(rows).map_err(|_| {
842 PyValueError::new_err(format!("Row count {} exceeds u32::MAX", rows))
843 })?;
844
845 let mut d_num_rows = self
846 .output_provider
847 .memory()
848 .alloc::<u32>(1)
849 .map_err(types::xlog_err)?;
850 self.output_provider
851 .device()
852 .inner()
853 .htod_sync_copy_into(&[rows_u32], &mut d_num_rows)
854 .map_err(types::xlog_err)?;
855
856 let buffer = xlog_cuda::CudaBuffer::from_columns(
857 vec![counts.into_bytes().into()],
858 rows,
859 d_num_rows,
860 schema_i32.clone(),
861 );
862 let tensor = self
863 .output_provider
864 .to_dlpack_table(buffer)
865 .column(0)
866 .map_err(types::xlog_err)?;
867 dlpack_capsule_from_tensor(py, tensor)
868 };
869
870 let query_rows = u64::try_from(query_counts.len())
871 .map_err(|_| PyValueError::new_err("query_counts length overflow"))?;
872 let query_counts_capsule = make_count_tensor(query_counts, query_rows)?;
873 let evidence_count_capsule = make_count_tensor(evidence_count, 1)?;
874 let resident_no_host_certified = no_host.is_no_host();
875
876 Ok(McDeviceEvalResult {
877 query_counts: query_counts_capsule,
878 evidence_count: evidence_count_capsule,
879 total_samples,
880 seed,
881 confidence,
882 nonmonotone_semantics: xlog_prob::mc::NONMONOTONE_SEMANTICS.to_string(),
883 nonmonotone_sccs,
884 nonmonotone_cycles,
885 nonmonotone_iteration_limit_hits,
886 sampling_method: match sampling_method_val {
887 McSamplingMethod::Rejection => "rejection".to_string(),
888 McSamplingMethod::EvidenceClamping => "evidence_clamping".to_string(),
889 },
890 resident_no_host_certified,
891 resident_no_host_policy_result: if resident_no_host_certified {
892 "certified".to_string()
893 } else {
894 "failed".to_string()
895 },
896 resident_no_host_tracked_dtoh_calls: no_host.tracked_dtoh_calls,
897 resident_no_host_tracked_htod_calls: no_host.tracked_htod_calls,
898 resident_no_host_host_loop_iterations: no_host.host_loop_iterations,
899 resident_no_host_per_sample_host_launches: no_host.per_sample_host_launches,
900 resident_no_host_untracked_metadata_reads: no_host.untracked_metadata_reads,
901 resident_no_host_engine_launches: no_host.engine_launches,
902 resident_no_host_host_fixpoint_iterations: no_host.host_fixpoint_iterations,
903 resident_no_host_per_operator_host_allocations: no_host.per_operator_host_allocations,
904 })
905 }
906
907 fn nll_loss(&self, query: &str) -> PyResult<f64> {
924 let prob = self.evaluate_query_probability(query)?;
925 Ok(types::nll_loss_value(prob))
926 }
927
928 fn nll_loss_batch(&self, queries: Vec<String>) -> PyResult<f64> {
941 if queries.is_empty() {
942 return Ok(0.0);
943 }
944
945 let probs = self.evaluate_query_probabilities(&queries)?;
946 Ok(probs.iter().map(|&p| types::nll_loss_value(p)).sum())
947 }
948
949 fn nll_loss_mean(&self, queries: Vec<String>) -> PyResult<f64> {
964 if queries.is_empty() {
965 return Err(PyValueError::new_err(
966 "Cannot compute mean NLL loss for empty query batch",
967 ));
968 }
969
970 let probs = self.evaluate_query_probabilities(&queries)?;
971 let sum: f64 = probs.iter().map(|&p| types::nll_loss_value(p)).sum();
972 Ok(sum / probs.len() as f64)
973 }
974
975 fn nll_loss_tensor(&self, py: Python<'_>, query: &str) -> PyResult<Py<PyAny>> {
986 let loss = self.nll_loss(query)?;
987 types::create_torch_tensor(py, loss)
988 }
989
990 fn nll_loss_batch_tensor(&self, py: Python<'_>, queries: Vec<String>) -> PyResult<Py<PyAny>> {
998 let loss = self.nll_loss_batch(queries)?;
999 types::create_torch_tensor(py, loss)
1000 }
1001
1002 pub fn zero_grad(&self, py: Python<'_>) -> PyResult<()> {
1011 for name in self.network_registry.names() {
1012 if let Some(handle) = self.network_registry.get(name) {
1013 if let Some(optimizer) = handle.optimizer() {
1014 optimizer.call_method0(py, "zero_grad")?;
1015 }
1016 }
1017 }
1018 Ok(())
1019 }
1020
1021 pub fn optimizer_step(&self, py: Python<'_>) -> PyResult<()> {
1026 for name in self.network_registry.names() {
1027 if let Some(handle) = self.network_registry.get(name) {
1028 if let Some(optimizer) = handle.optimizer() {
1029 optimizer.call_method0(py, "step")?;
1030 }
1031 }
1032 }
1033 Ok(())
1034 }
1035
1036 pub fn clip_grad_norms(&self, py: Python<'_>, max_norm: f64) -> PyResult<()> {
1040 let clip_fn = py.import("torch.nn.utils")?.getattr("clip_grad_norm_")?;
1041 for name in self.network_registry.names() {
1042 if let Some(handle) = self.network_registry.get(name) {
1043 if let Some(module) = handle.module() {
1044 let params = module.call_method0(py, "parameters")?;
1045 clip_fn.call1((params, max_norm))?;
1046 }
1047 }
1048 }
1049 Ok(())
1050 }
1051
1052 #[pyo3(signature = (network_name=None))]
1061 fn scheduler_step(&self, py: Python<'_>, network_name: Option<&str>) -> PyResult<()> {
1062 match network_name {
1063 Some(name) => {
1064 let handle = self.network_registry.get(name).ok_or_else(|| {
1065 pyo3::exceptions::PyValueError::new_err(format!(
1066 "No network registered with name '{name}'"
1067 ))
1068 })?;
1069 if let Some(scheduler) = handle.scheduler() {
1070 scheduler.call_method0(py, "step")?;
1071 }
1072 }
1073 None => {
1074 for name in self.network_registry.names() {
1075 if let Some(handle) = self.network_registry.get(name) {
1076 if let Some(scheduler) = handle.scheduler() {
1077 scheduler.call_method0(py, "step")?;
1078 }
1079 }
1080 }
1081 }
1082 }
1083 Ok(())
1084 }
1085
1086 fn get_lr(&self, py: Python<'_>, network_name: &str) -> PyResult<f64> {
1093 let handle = self.network_registry.get(network_name).ok_or_else(|| {
1094 pyo3::exceptions::PyValueError::new_err(format!(
1095 "No network registered with name '{network_name}'"
1096 ))
1097 })?;
1098 let optimizer = handle.optimizer().ok_or_else(|| {
1099 pyo3::exceptions::PyValueError::new_err(format!(
1100 "Network '{network_name}' has no optimizer"
1101 ))
1102 })?;
1103 let param_groups = optimizer.getattr(py, "param_groups")?;
1104 let group0 = param_groups.call_method1(py, "__getitem__", (0i32,))?;
1105 let lr = group0.call_method1(py, "__getitem__", ("lr",))?;
1106 lr.extract(py)
1107 }
1108
1109 fn set_lr(&self, py: Python<'_>, network_name: &str, lr: f64) -> PyResult<()> {
1117 let handle = self.network_registry.get(network_name).ok_or_else(|| {
1118 pyo3::exceptions::PyValueError::new_err(format!(
1119 "No network registered with name '{network_name}'"
1120 ))
1121 })?;
1122 let optimizer = handle.optimizer().ok_or_else(|| {
1123 pyo3::exceptions::PyValueError::new_err(format!(
1124 "Network '{network_name}' has no optimizer"
1125 ))
1126 })?;
1127 let param_groups = optimizer.getattr(py, "param_groups")?;
1128 let num_groups: usize = param_groups.call_method0(py, "__len__")?.extract(py)?;
1129 for i in 0..num_groups {
1130 let group = param_groups.call_method1(py, "__getitem__", (i as i32,))?;
1131 group.call_method(py, "__setitem__", ("lr", lr), None)?;
1132 }
1133 Ok(())
1134 }
1135
1136 #[pyo3(signature = (queries, batch_size=32, max_grad_norm=None))]
1154 fn train_epoch(
1155 &mut self,
1156 py: Python<'_>,
1157 queries: Vec<String>,
1158 batch_size: usize,
1159 max_grad_norm: Option<f64>,
1160 ) -> PyResult<EpochStats> {
1161 let mut history = TrainingHistory::new();
1162 self.train_epoch_internal(
1163 py,
1164 &queries,
1165 batch_size,
1166 usize::MAX,
1167 max_grad_norm,
1168 &mut history,
1169 )
1170 }
1171
1172 pub fn evaluate_loss(&self, queries: Vec<String>) -> PyResult<f64> {
1182 if queries.is_empty() {
1183 return Ok(0.0);
1184 }
1185
1186 let probs = self.evaluate_query_probabilities(&queries)?;
1187 let total_loss: f64 = probs.iter().map(|&p| types::nll_loss_value(p)).sum();
1188 Ok(total_loss / queries.len() as f64)
1189 }
1190
1191 #[pyo3(signature = (queries, batch_size=32, max_grad_norm=None))]
1193 fn train_epoch_tensor(
1194 &mut self,
1195 py: Python<'_>,
1196 queries: Vec<String>,
1197 batch_size: usize,
1198 max_grad_norm: Option<f64>,
1199 ) -> PyResult<EpochStats> {
1200 let mut history = TrainingHistory::new();
1201 self.train_epoch_tensor_internal(
1202 py,
1203 &queries,
1204 batch_size,
1205 usize::MAX,
1206 max_grad_norm,
1207 &mut history,
1208 )
1209 }
1210
1211 fn warmup_breakdown(&self, py: Python<'_>) -> PyResult<Option<Py<PyAny>>> {
1218 let ptx_profile = self.output_provider.ptx_load_profile();
1219 let circuit_profile = self
1226 .last_compile_profile
1227 .as_ref()
1228 .or_else(|| match &self.program {
1229 CompiledProbProgram::Exact(p) => p.last_compile_profile(),
1230 CompiledProbProgram::Mc(_) => None,
1231 });
1232
1233 if ptx_profile.is_none() && circuit_profile.is_none() {
1235 return Ok(None);
1236 }
1237
1238 let result = PyDict::new(py);
1239
1240 if let Some(ptx) = ptx_profile {
1241 let ptx_dict = PyDict::new(py);
1242 ptx_dict.set_item("total_sec", ptx.total_sec)?;
1243 ptx_dict.set_item("cubin_loaded", ptx.cubin_loaded)?;
1244 ptx_dict.set_item("ptx_fallback", ptx.ptx_fallback)?;
1245 let per_module = PyDict::new(py);
1246 for (name, sec) in &ptx.per_module_sec {
1247 per_module.set_item(name, *sec)?;
1248 }
1249 ptx_dict.set_item("per_module_sec", per_module)?;
1250 result.set_item("ptx", ptx_dict)?;
1251 }
1252
1253 if let Some(circuit) = circuit_profile {
1254 let circuit_dict = PyDict::new(py);
1255 circuit_dict.set_item("gpu_cache_hit", circuit.gpu_cache_hit)?;
1256 circuit_dict.set_item("disk_cache_hit", circuit.disk_cache_hit)?;
1257 circuit_dict.set_item("d4_compile_sec", circuit.d4_compile_sec)?;
1258 circuit_dict.set_item("verify_sec", circuit.verify_sec)?;
1259 circuit_dict.set_item("smooth_sec", circuit.smooth_sec)?;
1260 circuit_dict.set_item("cache_store_sec", circuit.cache_store_sec)?;
1261 circuit_dict.set_item("free_var_mask_sec", circuit.free_var_mask_sec)?;
1262 circuit_dict.set_item("cnf_hash_sec", circuit.cnf_hash_sec)?;
1263 result.set_item("circuit", circuit_dict)?;
1264 }
1265
1266 Ok(Some(result.into()))
1267 }
1268
1269 fn clear_circuit_cache(&mut self) {
1272 self.circuit_cache.clear();
1273 }
1274
1275 pub fn memory_stats(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1277 provider_memory_stats(py, &self.output_provider)
1278 }
1279
1280 pub fn rule_provenance(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1281 let provenance = xlog_logic::rule_provenance(&self.ast, None);
1282 pack_rule_provenance(py, &provenance)
1283 }
1284
1285 pub fn proof_traces(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1286 let provenance = xlog_logic::rule_provenance(&self.ast, None);
1287 let traces = xlog_logic::query_proof_traces(&self.ast, &provenance);
1288 pack_query_proof_traces(py, &traces)
1289 }
1290
1291 pub fn host_transfer_stats(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1292 let stats = self.output_provider.host_transfer_stats();
1293 let dict = PyDict::new(py);
1294 dict.set_item("dtoh_bytes", stats.dtoh_bytes)?;
1295 dict.set_item("htod_bytes", stats.htod_bytes)?;
1296 dict.set_item("dtoh_calls", stats.dtoh_calls)?;
1297 dict.set_item("htod_calls", stats.htod_calls)?;
1298 Ok(dict.into())
1299 }
1300
1301 pub fn reset_host_transfer_stats(&self) {
1302 self.output_provider.reset_host_transfer_stats()
1303 }
1304
1305 pub fn neural_hot_loop_diagnostics(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1306 let transfers = self.output_provider.host_transfer_stats();
1307 let dict = PyDict::new(py);
1308 dict.set_item("post_load_dtoh_bytes", transfers.dtoh_bytes)?;
1309 dict.set_item("post_load_htod_bytes", transfers.htod_bytes)?;
1310 dict.set_item("post_load_dtoh_calls", transfers.dtoh_calls)?;
1311 dict.set_item("post_load_htod_calls", transfers.htod_calls)?;
1312 dict.set_item("control_plane_bytes_per_iteration", py.None())?;
1313 dict.set_item(
1314 "control_plane_status",
1315 "unavailable: per-iteration control-plane byte counter is not registered",
1316 )?;
1317 dict.set_item("scalar_sync_checks", py.None())?;
1318 dict.set_item(
1319 "scalar_sync_status",
1320 "unavailable: scalar synchronization counter is not registered",
1321 )?;
1322
1323 let cuda_graph = PyDict::new(py);
1324 cuda_graph.set_item(
1325 "csm_cuda_graph_captures",
1326 self.output_provider.csm_cuda_graph_captures(),
1327 )?;
1328 cuda_graph.set_item(
1329 "csm_cuda_graph_launches",
1330 self.output_provider.csm_cuda_graph_launches(),
1331 )?;
1332 cuda_graph.set_item(
1333 "csm_cuda_graph_fallbacks",
1334 self.output_provider.csm_cuda_graph_fallbacks(),
1335 )?;
1336 cuda_graph.set_item(
1337 "csm_cuda_graph_cache_hits",
1338 self.output_provider.csm_cuda_graph_cache_hits(),
1339 )?;
1340 dict.set_item("cuda_graph", cuda_graph)?;
1341
1342 let circuit_cache = PyDict::new(py);
1343 circuit_cache.set_item("circuit_cache_size", self.circuit_cache.len())?;
1344 circuit_cache.set_item("circuit_cache_hits", self.circuit_cache_hits)?;
1345 circuit_cache.set_item("circuit_cache_misses", self.circuit_cache_misses)?;
1346 circuit_cache.set_item("template_compile_count", self.template_compile_count)?;
1347 circuit_cache.set_item(
1348 "query_signature_cache_size",
1349 self.query_signature_cache.len(),
1350 )?;
1351 dict.set_item("circuit_cache", circuit_cache)?;
1352
1353 Ok(dict.into())
1354 }
1355
1356 pub fn cuda_graph_stats(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
1357 let dict = PyDict::new(py);
1358 dict.set_item(
1359 "csm_cuda_graph_captures",
1360 self.output_provider.csm_cuda_graph_captures(),
1361 )?;
1362 dict.set_item(
1363 "csm_cuda_graph_launches",
1364 self.output_provider.csm_cuda_graph_launches(),
1365 )?;
1366 dict.set_item(
1367 "csm_cuda_graph_fallbacks",
1368 self.output_provider.csm_cuda_graph_fallbacks(),
1369 )?;
1370 dict.set_item(
1371 "csm_cuda_graph_cache_hits",
1372 self.output_provider.csm_cuda_graph_cache_hits(),
1373 )?;
1374 Ok(dict.into())
1375 }
1376}
1377
1378#[cfg(test)]
1379mod tests {
1380 use super::*;
1381 use xlog_logic::ast::{Directives, ProbMethod};
1382
1383 #[test]
1384 fn mc_config_inherits_directives_and_only_explicit_python_args_override_them() {
1385 let directives = Directives {
1386 prob_samples: Some(321),
1387 prob_seed: Some(17),
1388 prob_confidence: Some(0.8),
1389 prob_method: Some(ProbMethod::EvidenceClamping),
1390 prob_max_nonmonotone_iterations: Some(23),
1391 ..Directives::default()
1392 };
1393
1394 let inherited = resolve_mc_eval_config(&directives, None, None, None, None, None, false)
1395 .expect("valid source directives");
1396 assert_eq!(inherited.samples, 321);
1397 assert_eq!(inherited.seed, 17);
1398 assert_eq!(inherited.confidence, 0.8);
1399 assert_eq!(
1400 inherited.sampling_method,
1401 Some(McSamplingMethod::EvidenceClamping)
1402 );
1403 assert_eq!(inherited.max_nonmonotone_iterations, 23);
1404 assert!(!inherited.allow_cpu_oracle_fallback);
1405
1406 let overridden = resolve_mc_eval_config(
1407 &directives,
1408 Some(654),
1409 Some(29),
1410 Some(0.9),
1411 Some(31),
1412 Some(McSamplingMethod::Rejection),
1413 true,
1414 )
1415 .expect("valid explicit overrides");
1416 assert_eq!(overridden.samples, 654);
1417 assert_eq!(overridden.seed, 29);
1418 assert_eq!(overridden.confidence, 0.9);
1419 assert_eq!(
1420 overridden.sampling_method,
1421 Some(McSamplingMethod::Rejection)
1422 );
1423 assert_eq!(overridden.max_nonmonotone_iterations, 31);
1424 assert!(overridden.allow_cpu_oracle_fallback);
1425 }
1426}