Problems with Single-Mode Reasoning:

Rule Engine (Deterministic)          ML Model (Probabilistic)
+---------------------+            +---------------------+
| Strengths:          |            | Strengths:          |
|  - Explainable      |            |  - Handles ambiguity|
|  - Auditable        |            |  - Finds patterns   |
|  - Fast response    |            |  - Self-adaptive    |
+---------------------+            +---------------------+
| Limitations:        |            | Limitations:        |
|  - Cannot handle    |            |  - Black box        |
|    ambiguity        |            |  - Needs large data |
|  - Rule explosion   |            |  - Compliance risk  |
|  - No self-learning |            |  - Hard to audit    |
+---------------------+            +---------------------+

Three-Tier Hybrid Reasoning Architecture:

               Request Entry
                    |
                    v
           +--------+--------+
           |   Router/Triage  |
           +---------+--------+
                     |
        +------------+------------+
        |            |            |
        v            v            v
  +--------+   +--------+   +-----------+
  | Layer 1|   | Layer 2|   |  Layer 3  |
  | Rules  |   |   ML   |   |  Human    |
  |(Determ)|   |(Probab)|   |  Review   |
  +--------+   +--------+   +-----------+
        |            |            |
        +------+-----+------+----+
               |            |
               v            v
        +------------+ +----------+
        | Confidence | | Review   |
        |  Fusion    | |  Queue   |
        +------+-----+ +-----+---+
               |            |
               v            v
          Final Decision <--+

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Any


class ReasoningPath(Enum):
    """Reasoning path types"""
    RULE_ONLY = auto()        # Pure rules, no ML needed
    ML_ONLY = auto()          # Pure ML, no rule coverage
    RULE_THEN_ML = auto()     # Rule pre-filter + ML refinement
    ML_THEN_RULE = auto()     # ML prediction + rule validation
    PARALLEL_FUSION = auto()  # Parallel rule + ML with fusion
    HUMAN_REVIEW = auto()     # Direct to human review


@dataclass
class ReasoningRequest:
    """Reasoning request"""
    request_id: str
    domain: str              # "fraud", "credit", "inventory", etc.
    facts: dict[str, Any]
    urgency: str = "normal"  # "critical", "normal", "low"
    confidence_threshold: float = 0.85


class ReasoningRouter:
    """Request router for hybrid reasoning"""

    def __init__(self):
        self._domain_config: dict[str, dict] = {}
        self._default_path = ReasoningPath.RULE_THEN_ML

    def configure_domain(self, domain: str, config: dict) -> None:
        """Configure routing strategy for a domain"""
        self._domain_config[domain] = config

    def route(self, request: ReasoningRequest) -> ReasoningPath:
        """Determine reasoning path"""
        config = self._domain_config.get(request.domain, {})

        # Critical requests: skip ML, use rules only
        if request.urgency == "critical":
            return ReasoningPath.RULE_ONLY

        rules_coverage = config.get("rules_coverage", 0.0)
        ml_available = config.get("ml_model_available", False)

        if rules_coverage >= 0.95 and not ml_available:
            return ReasoningPath.RULE_ONLY
        if rules_coverage < 0.3 and ml_available:
            return ReasoningPath.ML_ONLY
        if ml_available:
            return ReasoningPath.PARALLEL_FUSION

        return self._default_path

Routing Decision Matrix:

                       ML Model Available
                    Yes              No
               +------------+  +------------+
Rules >= 95%   |  PARALLEL  |  | RULE_ONLY  |
coverage       |  _FUSION   |  |            |
               +------------+  +------------+
Rules 30-95%   |  RULE_THEN |  | RULE_ONLY  |
coverage       |  _ML       |  | + HUMAN    |
               +------------+  +------------+
Rules < 30%    |  ML_ONLY   |  | HUMAN      |
coverage       |            |  | _REVIEW    |
               +------------+  +------------+

from dataclasses import dataclass
import time
from typing import Any


@dataclass
class RuleResult:
    """Rule layer reasoning result"""
    rule_ids: list[str]         # Fired rule IDs
    conclusion: str             # Conclusion
    confidence: float           # Deterministic rules typically 1.0
    explanation: list[str]      # Explanation chain
    elapsed_ms: float
    is_conclusive: bool = True  # Whether a definitive conclusion was reached


class RuleReasoningLayer:
    """Layer 1: Deterministic rule reasoning"""

    def __init__(self, rete_network, working_memory):
        self._rete = rete_network
        self._wm = working_memory
        self._inconclusive_threshold = 0.7

    def reason(self, request: ReasoningRequest) -> RuleResult:
        """Execute rule reasoning"""
        start = time.monotonic()

        # Inject request facts into working memory
        for key, value in request.facts.items():
            self._wm.assert_fact(key, value)

        # Run Rete forward chain
        fired_rules = self._rete.run_to_completion(max_cycles=500)
        elapsed = (time.monotonic() - start) * 1000

        if not fired_rules:
            return RuleResult(
                rule_ids=[],
                conclusion="no_match",
                confidence=0.0,
                explanation=["No rules matched the current fact set"],
                elapsed_ms=elapsed,
                is_conclusive=False,
            )

        conclusion = self._aggregate_conclusions(fired_rules)
        coverage = self._compute_coverage(request.facts, fired_rules)

        return RuleResult(
            rule_ids=[r.rule_id for r in fired_rules],
            conclusion=conclusion,
            confidence=min(coverage, 1.0),
            explanation=self._build_explanation(fired_rules),
            elapsed_ms=elapsed,
            is_conclusive=coverage >= self._inconclusive_threshold,
        )

    def _aggregate_conclusions(self, fired_rules) -> str:
        sorted_rules = sorted(fired_rules, key=lambda r: r.priority, reverse=True)
        return sorted_rules[0].conclusion

    def _compute_coverage(self, facts: dict, fired_rules) -> float:
        total_facts = len(facts)
        covered = set()
        for rule in fired_rules:
            for cond in rule.conditions:
                covered.add(cond.attribute)
        return len(covered) / max(total_facts, 1)

    def _build_explanation(self, fired_rules) -> list[str]:
        return [
            f"Rule [{r.name}]: {' AND '.join(f'{c.attribute} {c.operator} {c.value}' for c in r.conditions)} -> {r.conclusion}"
            for r in fired_rules
        ]

from typing import Protocol


class MLModel(Protocol):
    """ML model interface"""
    def predict(self, features: dict[str, Any]) -> tuple[str, float]: ...
    def predict_proba(self, features: dict[str, Any]) -> dict[str, float]: ...

    @property
    def model_id(self) -> str: ...
    @property
    def version(self) -> str: ...


@dataclass
class ModelMetadata:
    """Model metadata"""
    model_id: str
    version: str
    domain: str
    accuracy: float
    f1_score: float
    training_date: str
    feature_names: list[str]
    is_active: bool = True


class ModelRegistry:
    """ML model registry"""

    def __init__(self):
        self._models: dict[str, MLModel] = {}
        self._metadata: dict[str, ModelMetadata] = {}
        self._domain_index: dict[str, list[str]] = {}

    def register(self, model: MLModel, metadata: ModelMetadata) -> None:
        key = f"{metadata.model_id}:{metadata.version}"
        self._models[key] = model
        self._metadata[key] = metadata
        self._domain_index.setdefault(metadata.domain, []).append(key)

    def get_best_model(self, domain: str) -> tuple[MLModel, ModelMetadata] | None:
        """Get the best active model for a given domain"""
        keys = self._domain_index.get(domain, [])
        active = [
            (self._models[k], self._metadata[k])
            for k in keys
            if self._metadata[k].is_active
        ]
        if not active:
            return None
        active.sort(key=lambda x: x[1].f1_score, reverse=True)
        return active[0]

@dataclass
class MLResult:
    """ML reasoning result"""
    model_id: str
    model_version: str
    prediction: str
    confidence: float
    probabilities: dict[str, float]
    feature_importance: dict[str, float]
    elapsed_ms: float


class MLReasoningLayer:
    """Layer 2: Machine learning reasoning"""

    def __init__(self, model_registry: ModelRegistry):
        self._registry = model_registry
        self._feature_extractors: dict[str, callable] = {}

    def register_feature_extractor(self, domain: str, extractor: callable):
        self._feature_extractors[domain] = extractor

    def reason(self, request: ReasoningRequest) -> MLResult | None:
        """Execute ML reasoning"""
        result = self._registry.get_best_model(request.domain)
        if result is None:
            return None

        model, metadata = result
        start = time.monotonic()

        # Feature extraction
        extractor = self._feature_extractors.get(request.domain)
        features = extractor(request.facts) if extractor else request.facts

        # Inference
        prediction, confidence = model.predict(features)
        probabilities = model.predict_proba(features)

        # Feature importance (permutation-based approximation)
        importance = self._compute_importance(model, features)

        elapsed = (time.monotonic() - start) * 1000

        return MLResult(
            model_id=metadata.model_id,
            model_version=metadata.version,
            prediction=prediction,
            confidence=confidence,
            probabilities=probabilities,
            feature_importance=importance,
            elapsed_ms=elapsed,
        )

    def _compute_importance(self, model: MLModel,
                            features: dict) -> dict[str, float]:
        """Compute feature importance via leave-one-out approximation"""
        _, base_conf = model.predict(features)
        importance = {}
        for key in features:
            modified = {k: v for k, v in features.items() if k != key}
            _, mod_conf = model.predict(modified)
            importance[key] = abs(base_conf - mod_conf)
        total = sum(importance.values()) or 1.0
        return {k: v / total for k, v in importance.items()}

ML Model Sandbox Execution Flow:

  +---------------------+
  |  ReasoningEngine     |
  |                     |
  |  1. Extract features|
  |  2. Serialize req   |
  +----------+----------+
             |
             v
  +----------+----------+
  |  nsjail Sandbox     |
  |  +-----------------+|
  |  | Model Runtime   ||
  |  | - Read-only FS  ||
  |  | - No network    ||
  |  | - CPU/mem limits ||
  |  | - Timeout 5s    ||
  |  +-----------------+|
  +----------+----------+
             |
             v
  +----------+----------+
  |  Deserialize result |
  |  + Validate conf    |
  +---------------------+

@dataclass
class HumanReviewRequest:
    """Human review request"""
    request_id: str
    original_request: ReasoningRequest
    rule_result: RuleResult | None
    ml_result: MLResult | None
    trigger_reason: str
    priority: str = "normal"
    deadline_hours: int = 24


class HumanReviewTrigger:
    """Determines whether human review is needed"""

    def __init__(self):
        self._confidence_threshold = 0.7
        self._conflict_threshold = 0.3
        self._high_impact_domains = {"credit", "fraud", "compliance"}

    def should_review(self, rule_result: RuleResult | None,
                      ml_result: MLResult | None,
                      request: ReasoningRequest) -> tuple[bool, str]:
        """
        Determine whether human review is needed.

        Returns: (needs_review, reason)
        """
        # Condition 1: Rule and ML results conflict
        if rule_result and ml_result:
            if rule_result.conclusion != ml_result.prediction:
                severity = abs(rule_result.confidence - ml_result.confidence)
                if severity < self._conflict_threshold:
                    return True, "Rule and ML conclusions conflict with similar confidence"

        # Condition 2: Both layers have low confidence
        rule_conf = rule_result.confidence if rule_result else 0.0
        ml_conf = ml_result.confidence if ml_result else 0.0
        max_conf = max(rule_conf, ml_conf)
        if max_conf < self._confidence_threshold:
            return True, f"Highest confidence {max_conf:.2f} below threshold"

        # Condition 3: High-impact domain
        if request.domain in self._high_impact_domains:
            if max_conf < 0.95:
                return True, f"High-impact domain {request.domain} requires high confidence"

        # Condition 4: Inconclusive rule layer
        if rule_result and not rule_result.is_conclusive:
            return True, "Rule layer did not reach a definitive conclusion"

        return False, ""

from collections import deque
from datetime import datetime


class ReviewQueue:
    """Human review queue"""

    def __init__(self):
        self._queue: deque[HumanReviewRequest] = deque()
        self._assigned: dict[str, HumanReviewRequest] = {}
        self._completed: dict[str, dict] = {}

    def enqueue(self, review: HumanReviewRequest) -> int:
        self._queue.append(review)
        return len(self._queue)

    def assign_next(self, reviewer_id: str) -> HumanReviewRequest | None:
        if not self._queue:
            return None
        sorted_q = sorted(
            self._queue,
            key=lambda r: (0 if r.priority == "critical" else 1, r.deadline_hours),
        )
        review = sorted_q[0]
        self._queue.remove(review)
        self._assigned[review.request_id] = review
        return review

    def complete_review(self, request_id: str, decision: str,
                       reviewer_id: str, notes: str) -> None:
        review = self._assigned.pop(request_id, None)
        if review is None:
            raise KeyError(f"Review {request_id} not in assigned list")
        self._completed[request_id] = {
            "review": review,
            "decision": decision,
            "reviewer_id": reviewer_id,
            "notes": notes,
            "completed_at": datetime.utcnow().isoformat(),
        }

    @property
    def pending_count(self) -> int:
        return len(self._queue) + len(self._assigned)

Confidence Fusion Flow:

  Rule Result                ML Result
  conclusion: "approve"      prediction: "approve"
  confidence: 0.92           confidence: 0.87
       |                          |
       +----------+---------------+
                  |
                  v
         +--------+--------+
         |  Fusion Engine   |
         |                  |
         | 1. Agreement     |
         |    check         |
         | 2. Weighted      |
         |    fusion        |
         | 3. Calibration   |
         +--------+---------+
                  |
                  v
         Final: "approve"
         confidence: 0.94

@dataclass
class FusionResult:
    """Fused reasoning result"""
    conclusion: str
    confidence: float
    rule_contribution: float
    ml_contribution: float
    fusion_method: str
    explanation: list[str]


class ConfidenceFusion:
    """Confidence fusion engine"""

    def __init__(self):
        self._domain_weights: dict[str, tuple[float, float]] = {
            # (rule_weight, ml_weight)
            "compliance": (0.9, 0.1),   # Compliance favors rules
            "fraud": (0.3, 0.7),        # Fraud favors ML
            "credit": (0.5, 0.5),       # Credit balanced
            "inventory": (0.4, 0.6),    # Inventory favors ML
        }
        self._default_weights = (0.5, 0.5)

    def fuse(self, rule_result: RuleResult | None,
             ml_result: MLResult | None,
             domain: str) -> FusionResult:
        """Fuse rule and ML results"""

        if rule_result and not ml_result:
            return FusionResult(
                conclusion=rule_result.conclusion,
                confidence=rule_result.confidence,
                rule_contribution=1.0, ml_contribution=0.0,
                fusion_method="rule_only",
                explanation=rule_result.explanation,
            )

        if ml_result and not rule_result:
            return FusionResult(
                conclusion=ml_result.prediction,
                confidence=ml_result.confidence,
                rule_contribution=0.0, ml_contribution=1.0,
                fusion_method="ml_only",
                explanation=[f"ML model {ml_result.model_id}: {ml_result.prediction} "
                           f"(confidence: {ml_result.confidence:.2f})"],
            )

        rule_w, ml_w = self._domain_weights.get(domain, self._default_weights)

        if rule_result.conclusion == ml_result.prediction:
            return self._fuse_agreement(rule_result, ml_result, rule_w, ml_w)
        else:
            return self._fuse_conflict(rule_result, ml_result, rule_w, ml_w)

    def _fuse_agreement(self, rule: RuleResult, ml: MLResult,
                        rw: float, mw: float) -> FusionResult:
        """Fusion when both layers agree"""
        weighted = rule.confidence * rw + ml.confidence * mw
        # Agreement boost (Bayesian update approximation)
        boosted = 1 - (1 - weighted) * (1 - min(rule.confidence, ml.confidence))
        return FusionResult(
            conclusion=rule.conclusion,
            confidence=min(boosted, 0.99),
            rule_contribution=rw, ml_contribution=mw,
            fusion_method="weighted_agreement",
            explanation=rule.explanation + [
                f"ML model agrees (confidence {ml.confidence:.2f}), boosted after fusion"
            ],
        )

    def _fuse_conflict(self, rule: RuleResult, ml: MLResult,
                       rw: float, mw: float) -> FusionResult:
        """Fusion when layers disagree"""
        rule_score = rule.confidence * rw
        ml_score = ml.confidence * mw

        if rule_score >= ml_score:
            conclusion = rule.conclusion
            confidence = rule_score / (rule_score + ml_score)
            explanation = rule.explanation + [
                f"Note: ML predicted '{ml.prediction}' (conf {ml.confidence:.2f}) "
                f"but rule layer scored higher"
            ]
        else:
            conclusion = ml.prediction
            confidence = ml_score / (rule_score + ml_score)
            explanation = [
                f"ML prediction '{ml.prediction}' (conf {ml.confidence:.2f}) "
                f"scored higher than rule layer"
            ]

        # Conflict penalty
        confidence = max(confidence - 0.15, 0.1)

        return FusionResult(
            conclusion=conclusion,
            confidence=confidence,
            rule_contribution=rw, ml_contribution=mw,
            fusion_method="conflict_resolution",
            explanation=explanation,
        )

@dataclass
class HybridReasoningResult:
    """Final hybrid reasoning result"""
    request_id: str
    conclusion: str
    confidence: float
    path_taken: ReasoningPath
    rule_result: RuleResult | None
    ml_result: MLResult | None
    fusion_result: FusionResult | None
    human_review_needed: bool
    review_reason: str
    total_elapsed_ms: float
    trace: list[str]


class HybridReasoningOrchestrator:
    """Hybrid reasoning orchestrator -- the core coordination component"""

    def __init__(self, router, rule_layer, ml_layer,
                 fusion, review_trigger, review_queue):
        self._router = router
        self._rule_layer = rule_layer
        self._ml_layer = ml_layer
        self._fusion = fusion
        self._review_trigger = review_trigger
        self._review_queue = review_queue

    async def reason(self, request: ReasoningRequest) -> HybridReasoningResult:
        start = time.monotonic()
        trace = []

        # Step 1: Route
        path = self._router.route(request)
        trace.append(f"Routing decision: {path.name}")

        rule_result = None
        ml_result = None

        # Step 2: Execute per path
        if path == ReasoningPath.RULE_ONLY:
            rule_result = self._rule_layer.reason(request)
            trace.append(f"Rule layer: {rule_result.conclusion} "
                        f"(conf {rule_result.confidence:.2f})")

        elif path == ReasoningPath.ML_ONLY:
            ml_result = self._ml_layer.reason(request)
            trace.append(f"ML layer: {ml_result.prediction} "
                        f"(conf {ml_result.confidence:.2f})")

        elif path == ReasoningPath.RULE_THEN_ML:
            rule_result = self._rule_layer.reason(request)
            trace.append(f"Rule layer: {rule_result.conclusion}")
            if not rule_result.is_conclusive:
                ml_result = self._ml_layer.reason(request)
                trace.append(f"Inconclusive, ML layer: {ml_result.prediction}")

        elif path == ReasoningPath.PARALLEL_FUSION:
            rule_result = self._rule_layer.reason(request)
            ml_result = self._ml_layer.reason(request)
            trace.append(f"Parallel: rule={rule_result.conclusion}, "
                        f"ml={ml_result.prediction}")

        # Step 3: Fusion
        fusion_result = None
        if rule_result or ml_result:
            fusion_result = self._fusion.fuse(
                rule_result, ml_result, request.domain
            )
            trace.append(f"Fused: {fusion_result.conclusion} "
                        f"(conf {fusion_result.confidence:.2f})")

        # Step 4: Human review check
        need_review, reason = self._review_trigger.should_review(
            rule_result, ml_result, request
        )
        if need_review:
            self._review_queue.enqueue(HumanReviewRequest(
                request_id=request.request_id,
                original_request=request,
                rule_result=rule_result,
                ml_result=ml_result,
                trigger_reason=reason,
            ))
            trace.append(f"Submitted for human review: {reason}")

        elapsed = (time.monotonic() - start) * 1000

        return HybridReasoningResult(
            request_id=request.request_id,
            conclusion=fusion_result.conclusion if fusion_result else "pending_review",
            confidence=fusion_result.confidence if fusion_result else 0.0,
            path_taken=path,
            rule_result=rule_result,
            ml_result=ml_result,
            fusion_result=fusion_result,
            human_review_needed=need_review,
            review_reason=reason if need_review else "",
            total_elapsed_ms=elapsed,
            trace=trace,
        )

Degradation Strategy Matrix:

Scenario                    | Fallback Action                  | Log Level
----------------------------|----------------------------------|---------
ML model service unavailable| Fall back to rule-only reasoning | WARN
ML inference timeout (>5s)  | Use rule result + flag           | WARN
Rule engine exception       | Use ML result + force review     | ERROR
Both layers fail            | Enqueue for human review         | CRITICAL
Review queue backlog >100   | Auto-execute low-risk per ML     | WARN

from datetime import datetime, timedelta


class CircuitBreaker:
    """Circuit breaker for ML inference"""

    def __init__(self, failure_threshold: int = 5,
                 recovery_timeout_s: int = 60):
        self._failure_count = 0
        self._failure_threshold = failure_threshold
        self._recovery_timeout = timedelta(seconds=recovery_timeout_s)
        self._last_failure_time: datetime | None = None
        self._state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def record_success(self) -> None:
        self._failure_count = 0
        self._state = "CLOSED"

    def record_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = datetime.utcnow()
        if self._failure_count >= self._failure_threshold:
            self._state = "OPEN"

    def allow_request(self) -> bool:
        if self._state == "CLOSED":
            return True
        if self._state == "OPEN":
            if (datetime.utcnow() - self._last_failure_time
                    > self._recovery_timeout):
                self._state = "HALF_OPEN"
                return True
            return False
        return True  # HALF_OPEN allows one probe

    @property
    def state(self) -> str:
        return self._state

syntax = "proto3";
package onto.reasoning.v1;

service HybridReasoningService {
    rpc Reason(HybridReasoningRequest) returns (HybridReasoningResponse);
    rpc GetReasoningTrace(TraceRequest) returns (ReasoningTrace);
    rpc GetQueueStatus(QueueStatusRequest) returns (QueueStatusResponse);
}

message HybridReasoningRequest {
    string request_id = 1;
    string domain = 2;
    map<string, string> facts = 3;
    string urgency = 4;
    double confidence_threshold = 5;
}

message HybridReasoningResponse {
    string conclusion = 1;
    double confidence = 2;
    string path_taken = 3;
    bool human_review_needed = 4;
    string review_reason = 5;
    repeated string trace = 6;
    double elapsed_ms = 7;
}

# Scenario: Credit application approval

# 1. Configure domain
router = ReasoningRouter()
router.configure_domain("credit", {
    "rules_coverage": 0.7,
    "ml_model_available": True,
    "ml_model_accuracy": 0.92,
})

# 2. Submit reasoning request
request = ReasoningRequest(
    request_id="CR-2026-001",
    domain="credit",
    facts={
        "applicant_age": 35,
        "annual_income": 250000,
        "debt_ratio": 0.45,
        "credit_score": 680,
        "employment_years": 3,
        "previous_defaults": 0,
    },
    confidence_threshold=0.85,
)

# 3. Routing decision
path = router.route(request)
# -> ReasoningPath.PARALLEL_FUSION (rules_coverage=0.7 + ML available)

# 4. Rule layer result
# Rule: credit_score >= 650 AND debt_ratio < 0.5 -> "conditional_approve"
# confidence: 0.85

# 5. ML layer result
# Model: gradient_boosting_v3
# prediction: "approve", confidence: 0.91

# 6. Fusion (credit domain weights 0.5:0.5)
# Agreement direction ("approve" family)
# Fused confidence: 0.94 (agreement boost)
# Final: "approve", confidence=0.94

# 7. Human review: NOT needed (confidence 0.94 > 0.85 threshold)