core/algebra/backend.py
Shay 2bd70d0a9d
Fix remaining runtime regressions after contract cleanup
- close versor_apply outputs at algebra boundary
- route backend versor_apply through canonical closure semantics
- keep selected ChatResponse surface equal to ArticulationPlan surface
- derive proposition relation from selected slots
- rank proposition slots with pure CGA metric
2026-05-14 19:05:36 -07:00

80 lines
2.4 KiB
Python

"""
Backend dispatch: use Rust extension (core_rs) when available,
fall back to pure Python (algebra/cl41.py etc.) transparently.
This module is the single switch. All algebra modules import from here
for performance-critical ops. Pure Python is always the fallback —
the system is never broken by a missing Rust build.
Usage:
from algebra.backend import geometric_product, versor_apply, cga_inner, vault_recall
"""
import os
import numpy as np
_REQUESTED_BACKEND = os.environ.get("CORE_BACKEND", "").strip().lower()
_ALLOW_RUST = _REQUESTED_BACKEND not in {"numpy", "python", "py"}
try:
import core_rs as _rs
_RUST = _ALLOW_RUST
except ImportError:
_RUST = False
def geometric_product(A: np.ndarray, B: np.ndarray) -> np.ndarray:
if _RUST:
return np.asarray(_rs.geometric_product(A, B), dtype=np.float32)
from algebra.cl41 import geometric_product as _gp
return _gp(A, B)
def versor_apply(V: np.ndarray, F: np.ndarray) -> np.ndarray:
"""Apply a versor through the canonical algebra closure boundary.
The Rust extension's raw sandwich path is intentionally bypassed here
until it enforces the same closure semantics as algebra.versor. Runtime
invariants depend on this operator returning a closed field; generation,
propagation, and vault recall are not allowed to repair it downstream.
"""
from algebra.versor import versor_apply as _va
return _va(V, F)
def versor_condition(F: np.ndarray) -> float:
if _RUST:
return float(_rs.versor_condition(F))
from algebra.versor import versor_condition as _vc
return _vc(F)
def cga_inner(X: np.ndarray, Y: np.ndarray) -> float:
if _RUST:
return float(_rs.cga_inner(X, Y))
from algebra.cga import cga_inner as _ci
return _ci(X, Y)
def vault_recall(versors: list, query: np.ndarray, top_k: int = 5) -> list:
"""
Top-k CGA inner product recall.
Rust path: parallel Rayon scan (releases GIL, true multithreaded).
Python path: sequential list comprehension.
"""
if _RUST:
try:
results = _rs.vault_recall(versors, query, top_k)
return results
except Exception:
pass
q = np.asarray(query)
scores = [(i, float(cga_inner(q, np.asarray(v)))) for i, v in enumerate(versors)]
scores.sort(key=lambda x: -x[1])
return scores[:top_k]
def using_rust() -> bool:
"""Returns True if the Rust extension is loaded."""
return _RUST