69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""Deterministic Decision Logic Tree Engine: store and evaluate DLTs; fail-closed."""
|
|
|
|
from typing import Any
|
|
|
|
from fusionagi.maa.schemas.dlt import DLTContract, DLTNode
|
|
|
|
|
|
class DLTEngine:
|
|
"""Store and evaluate Deterministic Decision Logic Trees; immutable, versioned contracts."""
|
|
|
|
def __init__(self) -> None:
|
|
self._contracts: dict[str, DLTContract] = {}
|
|
|
|
def register(self, contract: DLTContract) -> None:
|
|
"""Register an immutable DLT contract (by contract_id)."""
|
|
key = f"{contract.contract_id}@v{contract.version}"
|
|
self._contracts[key] = contract
|
|
|
|
def get(self, contract_id: str, version: int | None = None) -> DLTContract | None:
|
|
"""Return contract by id; optional version (latest if omitted)."""
|
|
if version is not None:
|
|
return self._contracts.get(f"{contract_id}@v{version}")
|
|
best: DLTContract | None = None
|
|
for k, c in self._contracts.items():
|
|
if c.contract_id == contract_id and (best is None or c.version > best.version):
|
|
best = c
|
|
return best
|
|
|
|
def evaluate(
|
|
self,
|
|
contract_id: str,
|
|
context: dict[str, Any],
|
|
version: int | None = None,
|
|
) -> tuple[bool, str]:
|
|
"""Evaluate DLT from root; deterministic, fail-closed. Return (True, "") or (False, root_cause)."""
|
|
contract = self.get(contract_id, version)
|
|
if not contract:
|
|
return False, f"DLT contract not found: {contract_id}"
|
|
return self._evaluate_node(contract, contract.root_id, context)
|
|
|
|
def _evaluate_node(
|
|
self,
|
|
contract: DLTContract,
|
|
node_id: str,
|
|
context: dict[str, Any],
|
|
) -> tuple[bool, str]:
|
|
node = contract.nodes.get(node_id)
|
|
if not node:
|
|
return False, f"DLT node not found: {node_id}"
|
|
passed = self._check_condition(node, context)
|
|
if not passed:
|
|
if node.fail_closed:
|
|
return False, f"DLT node failed (fail-closed): {node_id} condition={node.condition}"
|
|
for child_id in node.children:
|
|
ok, cause = self._evaluate_node(contract, child_id, context)
|
|
if not ok:
|
|
return False, cause
|
|
return True, ""
|
|
|
|
def _check_condition(self, node: DLTNode, context: dict[str, Any]) -> bool:
|
|
"""Evaluate condition; unknown conditions are fail-closed (False)."""
|
|
if node.condition.startswith("required:"):
|
|
key = node.condition.split(":", 1)[1].strip()
|
|
return key in context and context[key] is not None
|
|
if node.condition == "always":
|
|
return True
|
|
# Unknown condition: fail-closed
|
|
return False
|