ECDT: add decision proposal boundary

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",

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),
+        }

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}"

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"