diff --git a/src/intelligence/ecdt/__init__.py b/src/intelligence/ecdt/__init__.py index c76fb1c3..9537f02c 100644 --- a/src/intelligence/ecdt/__init__.py +++ b/src/intelligence/ecdt/__init__.py @@ -1,5 +1,8 @@ """AEON MATRIX Enterprise Cognitive Digital Twin.""" +from .decision_proposal import DecisionProposal +from .decision_proposal_builder import DecisionProposalBuilder + from .governed_scenario_integration import ECDTGovernedScenarioIntegration from .scenario_engine import ECDTScenarioEngine, ScenarioEvaluation @@ -10,6 +13,8 @@ ) __all__ = [ + "DecisionProposal", + "DecisionProposalBuilder", "ECDTGovernedScenarioIntegration", "ECDTScenarioEngine", "ScenarioEvaluation", diff --git a/src/intelligence/ecdt/decision_proposal.py b/src/intelligence/ecdt/decision_proposal.py new file mode 100644 index 00000000..090d1f37 --- /dev/null +++ b/src/intelligence/ecdt/decision_proposal.py @@ -0,0 +1,28 @@ +"""Immutable ECDT decision proposal model.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, Mapping, Tuple + + +@dataclass(frozen=True) +class DecisionProposal: + proposal_id: str + status: str + scenario_name: str + action: str + evidence: Mapping[str, Any] + policy_context: Mapping[str, Any] + trace: Tuple[str, ...] + + def to_dict(self) -> Dict[str, Any]: + return { + "proposal_id": self.proposal_id, + "status": self.status, + "scenario_name": self.scenario_name, + "action": self.action, + "evidence": dict(self.evidence), + "policy_context": dict(self.policy_context), + "trace": list(self.trace), + } diff --git a/src/intelligence/ecdt/decision_proposal_builder.py b/src/intelligence/ecdt/decision_proposal_builder.py new file mode 100644 index 00000000..c12acc48 --- /dev/null +++ b/src/intelligence/ecdt/decision_proposal_builder.py @@ -0,0 +1,93 @@ +"""Build governed ECDT decision proposals from recommendations.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any, Mapping + +from .decision_proposal import DecisionProposal + + +class DecisionProposalBuilder: + """Create deterministic proposals without granting authority.""" + + def build( + self, + *, + recommendation: Mapping[str, Any], + evidence: Mapping[str, Any], + policy_context: Mapping[str, Any] | None = None, + ) -> DecisionProposal: + if not isinstance(recommendation, Mapping): + raise TypeError("recommendation must be mapping-compatible") + + scenario_name = str( + recommendation.get("name", "") + ).strip() + if not scenario_name: + raise ValueError("recommendation name is required") + + action = str( + recommendation.get("action", "") + ).strip() + if not action: + raise ValueError("recommendation action is required") + + if not isinstance(evidence, Mapping) or not evidence: + raise ValueError("evidence is required") + + policy = dict(policy_context or {}) + + proposal_id = self._proposal_id( + scenario_name=scenario_name, + action=action, + evidence=evidence, + policy_context=policy, + ) + + trace = ( + f"scenario={scenario_name}", + f"action={action}", + f"proposal_id={proposal_id}", + "status=PROPOSED", + "execution_authorized=false", + ) + + return DecisionProposal( + proposal_id=proposal_id, + status="PROPOSED", + scenario_name=scenario_name, + action=action, + evidence=dict(evidence), + policy_context=policy, + trace=trace, + ) + + @staticmethod + def _proposal_id( + *, + scenario_name: str, + action: str, + evidence: Mapping[str, Any], + policy_context: Mapping[str, Any], + ) -> str: + payload = { + "scenario_name": scenario_name, + "action": action, + "evidence": evidence, + "policy_context": policy_context, + } + + canonical = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + default=str, + ) + + digest = hashlib.sha256( + canonical.encode("utf-8") + ).hexdigest()[:16] + + return f"proposal-{digest}" diff --git a/tests/ecdt/test_decision_proposal.py b/tests/ecdt/test_decision_proposal.py new file mode 100644 index 00000000..765c5204 --- /dev/null +++ b/tests/ecdt/test_decision_proposal.py @@ -0,0 +1,158 @@ +import pytest + +from src.intelligence.ecdt import ( + DecisionProposal, + DecisionProposalBuilder, +) + + +def recommendation(): + return { + "name": "dynamic_labor_scaling", + "action": "optimize_capacity", + } + + +def evidence(): + return { + "capacity": 0.95, + "demand": 1.10, + "source": "WMS", + } + + +def test_proposal_is_created_as_proposed(): + proposal = DecisionProposalBuilder().build( + recommendation=recommendation(), + evidence=evidence(), + policy_context={"allowed": True}, + ) + + assert proposal.status == "PROPOSED" + assert proposal.scenario_name == "dynamic_labor_scaling" + assert proposal.action == "optimize_capacity" + assert proposal.proposal_id.startswith("proposal-") + + +def test_proposal_serialization(): + proposal = DecisionProposalBuilder().build( + recommendation=recommendation(), + evidence=evidence(), + ) + + result = proposal.to_dict() + + assert result["status"] == "PROPOSED" + assert result["scenario_name"] == "dynamic_labor_scaling" + assert result["action"] == "optimize_capacity" + assert result["proposal_id"].startswith("proposal-") + assert result["evidence"]["source"] == "WMS" + + +def test_proposal_id_is_deterministic(): + builder = DecisionProposalBuilder() + + first = builder.build( + recommendation=recommendation(), + evidence=evidence(), + policy_context={"allowed": True}, + ) + + second = builder.build( + recommendation=recommendation(), + evidence=evidence(), + policy_context={"allowed": True}, + ) + + assert first.proposal_id == second.proposal_id + + +def test_policy_context_changes_proposal_identity(): + builder = DecisionProposalBuilder() + + allowed = builder.build( + recommendation=recommendation(), + evidence=evidence(), + policy_context={"allowed": True}, + ) + + blocked = builder.build( + recommendation=recommendation(), + evidence=evidence(), + policy_context={"allowed": False}, + ) + + assert allowed.proposal_id != blocked.proposal_id + + +def test_missing_recommendation_name_raises(): + data = recommendation() + data["name"] = "" + + with pytest.raises(ValueError): + DecisionProposalBuilder().build( + recommendation=data, + evidence=evidence(), + ) + + +def test_missing_action_raises(): + data = recommendation() + data["action"] = "" + + with pytest.raises(ValueError): + DecisionProposalBuilder().build( + recommendation=data, + evidence=evidence(), + ) + + +def test_missing_evidence_raises(): + with pytest.raises(ValueError): + DecisionProposalBuilder().build( + recommendation=recommendation(), + evidence={}, + ) + + +def test_proposal_has_no_execution_authority(): + proposal = DecisionProposalBuilder().build( + recommendation=recommendation(), + evidence=evidence(), + ) + + result = proposal.to_dict() + + for field in ( + "authorized", + "authorization_id", + "executed", + "applied", + "deployed", + "promoted", + ): + assert field not in result + + +def test_builder_has_no_execution_interface(): + builder = DecisionProposalBuilder() + + for name in ( + "execute", + "executor", + "apply", + "deploy", + "promote", + "authorize", + ): + assert not hasattr(builder, name) + + +def test_proposal_is_immutable(): + proposal = DecisionProposalBuilder().build( + recommendation=recommendation(), + evidence=evidence(), + ) + + with pytest.raises(Exception): + proposal.status = "AUTHORIZED"