ECDT: integrate scenario engine with governed runtime

src/intelligence/ecdt/__init__.py

@@ -1,5 +1,7 @@
 """AEON MATRIX Enterprise Cognitive Digital Twin."""
 
+from .governed_scenario_integration import ECDTGovernedScenarioIntegration
+
 from .scenario_engine import ECDTScenarioEngine, ScenarioEvaluation
 
 from .runtime import (
@@ -8,6 +10,7 @@
 )
 
 __all__ = [
+    "ECDTGovernedScenarioIntegration",
     "ECDTScenarioEngine",
     "ScenarioEvaluation",
     "ECDTExecutionMode",

src/intelligence/ecdt/governed_scenario_integration.py

@@ -0,0 +1,103 @@
+"""Governed integration between ECDT scenario evaluation and runtime."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, Mapping, Sequence
+
+from .runtime import ECDTRuntime
+from .scenario_engine import ECDTScenarioEngine
+
+
+class ECDTGovernedScenarioIntegration:
+    """Route a scenario recommendation into ECDTRuntime governance.
+
+    A recommendation is not execution authority.
+    All execution decisions remain owned by ECDTRuntime.
+    """
+
+    def __init__(
+        self,
+        *,
+        scenario_engine: ECDTScenarioEngine | None = None,
+        runtime: ECDTRuntime | None = None,
+    ) -> None:
+        self.scenario_engine = scenario_engine or ECDTScenarioEngine()
+        self.runtime = runtime or ECDTRuntime()
+
+    def run(
+        self,
+        *,
+        signals: Iterable[Dict[str, Any]],
+        risk: Dict[str, Any],
+        observed_state: Mapping[str, Any],
+        scenarios: Sequence[Mapping[str, Any]],
+        policy: Mapping[str, Any] | None = None,
+        human_approved: bool = False,
+    ) -> Dict[str, Any]:
+        """Evaluate scenarios, then route the recommendation to governance."""
+
+        evaluation = self.scenario_engine.evaluate(
+            observed_state=observed_state,
+            scenarios=scenarios,
+            policy=policy,
+        )
+
+        trace: Dict[str, Any] = {
+            "scenario_evaluation": evaluation,
+            "recommendation_is_authority": False,
+            "executed": False,
+        }
+
+        recommendation = evaluation.get("recommended")
+
+        if recommendation is None:
+            trace["status"] = "NO_RECOMMENDATION"
+            return trace
+
+        selected = self._find_selected_scenario(
+            scenarios=scenarios,
+            name=str(recommendation["name"]),
+        )
+
+        action = str(selected.get("action", "")).strip()
+        if not action:
+            trace["status"] = "ACTION_REQUIRED"
+            trace["selected_scenario"] = dict(selected)
+            return trace
+
+        runtime_scenario = selected.get("runtime_scenario")
+
+        if runtime_scenario is None:
+            runtime_scenario = {
+                "name": recommendation["name"],
+            }
+
+        if not isinstance(runtime_scenario, Mapping):
+            raise TypeError("runtime_scenario must be mapping-compatible")
+
+        runtime_result = self.runtime.run(
+            signals=signals,
+            risk=risk,
+            scenario=dict(runtime_scenario),
+            action=action,
+            human_approved=human_approved,
+        )
+
+        trace["selected_scenario"] = dict(selected)
+        trace["runtime"] = runtime_result
+        trace["status"] = runtime_result["status"]
+        trace["executed"] = bool(runtime_result.get("executed", False))
+
+        return trace
+
+    @staticmethod
+    def _find_selected_scenario(
+        *,
+        scenarios: Sequence[Mapping[str, Any]],
+        name: str,
+    ) -> Mapping[str, Any]:
+        for scenario in scenarios:
+            if str(scenario.get("name", "")).strip() == name:
+                return scenario
+
+        raise RuntimeError("recommended scenario not found in source candidates")

tests/ecdt/test_governed_scenario_integration.py

@@ -0,0 +1,165 @@
+from src.intelligence.ecdt import (
+    ECDTExecutionMode,
+    ECDTGovernedScenarioIntegration,
+    ECDTRuntime,
+)
+
+
+def signals():
+    return [
+        {
+            "source": "WMS",
+            "metric": "capacity",
+            "value": 0.95,
+        }
+    ]
+
+
+def risk():
+    return {
+        "type": "capacity_shortage",
+    }
+
+
+def observed_state():
+    return {
+        "capacity": 0.95,
+        "demand": 1.10,
+    }
+
+
+def candidates(action="optimize_capacity"):
+    return [
+        {
+            "name": "baseline",
+            "impact_score": 0.30,
+            "risk_score": 0.10,
+            "confidence": 0.90,
+            "action": "optimize_capacity",
+        },
+        {
+            "name": "dynamic_labor_scaling",
+            "impact_score": 0.90,
+            "risk_score": 0.10,
+            "confidence": 0.95,
+            "action": action,
+            "runtime_scenario": {
+                "name": "dynamic_labor_scaling",
+            },
+        },
+    ]
+
+
+def test_recommendation_routes_to_runtime_dry_run():
+    integration = ECDTGovernedScenarioIntegration()
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=candidates(),
+    )
+
+    assert result["scenario_evaluation"]["recommended"]["name"] == (
+        "dynamic_labor_scaling"
+    )
+    assert result["recommendation_is_authority"] is False
+    assert result["status"] == "DRY_RUN"
+    assert result["executed"] is False
+    assert result["runtime"]["executed"] is False
+
+
+def test_policy_blocked_scenarios_never_reach_runtime():
+    integration = ECDTGovernedScenarioIntegration()
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=candidates(),
+        policy={"allowed": False},
+    )
+
+    assert result["status"] == "NO_RECOMMENDATION"
+    assert result["executed"] is False
+    assert "runtime" not in result
+
+
+def test_restricted_action_still_requires_human():
+    integration = ECDTGovernedScenarioIntegration()
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=candidates(action="delete_inventory"),
+    )
+
+    assert result["status"] == "HUMAN_REQUIRED"
+    assert result["executed"] is False
+    assert result["runtime"]["governance"]["approval"]["human_required"] is True
+
+
+def test_guard_block_cannot_be_bypassed_by_recommendation():
+    integration = ECDTGovernedScenarioIntegration()
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=candidates(action="shutdown_system"),
+        human_approved=True,
+    )
+
+    assert result["status"] == "BLOCKED"
+    assert result["executed"] is False
+
+
+def test_execute_mode_still_requires_executor():
+    runtime = ECDTRuntime(
+        execution_mode=ECDTExecutionMode.EXECUTE,
+    )
+
+    integration = ECDTGovernedScenarioIntegration(
+        runtime=runtime,
+    )
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=candidates(),
+    )
+
+    assert result["status"] == "EXECUTOR_REQUIRED"
+    assert result["executed"] is False
+
+
+def test_missing_action_does_not_execute():
+    data = candidates()
+    data[1].pop("action")
+
+    integration = ECDTGovernedScenarioIntegration()
+
+    result = integration.run(
+        signals=signals(),
+        risk=risk(),
+        observed_state=observed_state(),
+        scenarios=data,
+    )
+
+    assert result["status"] == "ACTION_REQUIRED"
+    assert result["executed"] is False
+    assert "runtime" not in result
+
+
+def test_integration_has_no_direct_execution_interface():
+    integration = ECDTGovernedScenarioIntegration()
+
+    for name in (
+        "execute",
+        "apply",
+        "deploy",
+        "promote",
+    ):
+        assert not hasattr(integration, name)