diff --git a/intelligence/ecdt/__init__.py b/intelligence/ecdt/__init__.py new file mode 100644 index 00000000..d78d1647 --- /dev/null +++ b/intelligence/ecdt/__init__.py @@ -0,0 +1,17 @@ +"""Enterprise Cognitive Decision Trust (ECDT).""" + +from .application_validation import ( + ApplicationEvidence, + ApplicationValidation, + ValidationResult, + ValidationStatus, +) +from .application_validation_gate import ApplicationValidationGate + +__all__ = [ + "ApplicationEvidence", + "ApplicationValidation", + "ApplicationValidationGate", + "ValidationResult", + "ValidationStatus", +] diff --git a/intelligence/ecdt/application_validation.py b/intelligence/ecdt/application_validation.py new file mode 100644 index 00000000..8e4bc169 --- /dev/null +++ b/intelligence/ecdt/application_validation.py @@ -0,0 +1,94 @@ +"""Deterministic application validation primitives for AEON MATRIX ECDT.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Mapping + + +class ValidationStatus(str, Enum): + APPROVED = "APPROVED" + REJECTED = "REJECTED" + INSUFFICIENT_EVIDENCE = "INSUFFICIENT_EVIDENCE" + + +@dataclass(frozen=True) +class ApplicationEvidence: + application_id: str + source: str + facts: Mapping[str, object] = field(default_factory=dict) + verified: bool = False + + +@dataclass(frozen=True) +class ValidationResult: + application_id: str + status: ValidationStatus + reasons: tuple[str, ...] + evidence_count: int + verified_evidence_count: int + + @property + def approved(self) -> bool: + return self.status is ValidationStatus.APPROVED + + +class ApplicationValidation: + """Evaluates evidence without executing operational actions.""" + + def __init__(self, minimum_verified_evidence: int = 1) -> None: + if minimum_verified_evidence < 1: + raise ValueError("minimum_verified_evidence must be >= 1") + + self.minimum_verified_evidence = minimum_verified_evidence + + def validate( + self, + application_id: str, + evidence: list[ApplicationEvidence], + ) -> ValidationResult: + if not application_id.strip(): + raise ValueError("application_id must not be empty") + + relevant = tuple( + item for item in evidence + if item.application_id == application_id + ) + + if not relevant: + return ValidationResult( + application_id=application_id, + status=ValidationStatus.INSUFFICIENT_EVIDENCE, + reasons=("NO_EVIDENCE",), + evidence_count=0, + verified_evidence_count=0, + ) + + verified = tuple(item for item in relevant if item.verified) + + if len(verified) < self.minimum_verified_evidence: + return ValidationResult( + application_id=application_id, + status=ValidationStatus.INSUFFICIENT_EVIDENCE, + reasons=("VERIFIED_EVIDENCE_BELOW_THRESHOLD",), + evidence_count=len(relevant), + verified_evidence_count=len(verified), + ) + + if any(not item.source.strip() for item in verified): + return ValidationResult( + application_id=application_id, + status=ValidationStatus.REJECTED, + reasons=("INVALID_EVIDENCE_SOURCE",), + evidence_count=len(relevant), + verified_evidence_count=len(verified), + ) + + return ValidationResult( + application_id=application_id, + status=ValidationStatus.APPROVED, + reasons=("EVIDENCE_THRESHOLD_SATISFIED",), + evidence_count=len(relevant), + verified_evidence_count=len(verified), + ) diff --git a/intelligence/ecdt/application_validation_gate.py b/intelligence/ecdt/application_validation_gate.py new file mode 100644 index 00000000..bea93b3c --- /dev/null +++ b/intelligence/ecdt/application_validation_gate.py @@ -0,0 +1,36 @@ +"""Decision gate for ECDT application validation.""" + +from __future__ import annotations + +from .application_validation import ( + ApplicationEvidence, + ApplicationValidation, + ValidationResult, +) + + +class ApplicationValidationGate: + """ + Fail-closed gate. + + The gate validates evidence only. It does not execute the requested + business action and therefore preserves separation between intelligence, + governance, and operational execution. + """ + + def __init__(self, validator: ApplicationValidation | None = None) -> None: + self.validator = validator or ApplicationValidation() + + def evaluate( + self, + application_id: str, + evidence: list[ApplicationEvidence], + ) -> ValidationResult: + return self.validator.validate(application_id, evidence) + + def allows( + self, + application_id: str, + evidence: list[ApplicationEvidence], + ) -> bool: + return self.evaluate(application_id, evidence).approved diff --git a/services/guardian/execution/authorization.py b/services/guardian/execution/authorization.py new file mode 100644 index 00000000..4dbf8beb --- /dev/null +++ b/services/guardian/execution/authorization.py @@ -0,0 +1,93 @@ +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + + +@dataclass(frozen=True) +class ExecutionAuthorization: + """ + Guardian-issued execution authority. + + The authority is bound to one decision and one action. + V1 provides architectural authority semantics; it is not + a cryptographically signed security token. + """ + + authorization_id: str + decision_id: str + action: str + issued_by: str + policy_status: str + approval_status: str + issued_at: datetime + expires_at: datetime + + +class AuthorizationIssuer: + ISSUER = "GUARDIAN" + + def issue( + self, + *, + decision_id, + action, + policy_status, + approval_status, + ttl_seconds=300, + ): + if policy_status != "APPROVED": + raise ValueError("POLICY_NOT_APPROVED") + + if approval_status != "APPROVED": + raise ValueError("APPROVAL_NOT_APPROVED") + + if not decision_id: + raise ValueError("DECISION_ID_REQUIRED") + + if not action: + raise ValueError("ACTION_REQUIRED") + + issued_at = datetime.now(timezone.utc) + + return ExecutionAuthorization( + authorization_id=str(uuid4()), + decision_id=decision_id, + action=action, + issued_by=self.ISSUER, + policy_status=policy_status, + approval_status=approval_status, + issued_at=issued_at, + expires_at=issued_at + timedelta(seconds=ttl_seconds), + ) + + def verify( + self, + authority, + *, + decision_id, + action, + ): + if not isinstance(authority, ExecutionAuthorization): + return False + + if authority.issued_by != self.ISSUER: + return False + + if authority.policy_status != "APPROVED": + return False + + if authority.approval_status != "APPROVED": + return False + + if authority.decision_id != decision_id: + return False + + if authority.action != action: + return False + + now = datetime.now(timezone.utc) + + if authority.expires_at <= now: + return False + + return True diff --git a/services/guardian/execution/execution_layer.py b/services/guardian/execution/execution_layer.py index 2852d126..b464f225 100644 --- a/services/guardian/execution/execution_layer.py +++ b/services/guardian/execution/execution_layer.py @@ -1,35 +1,94 @@ from services.guardian.execution.action_executor import ActionExecutor -from services.guardian.execution.approval_gate import ApprovalGate from services.guardian.execution.workflow_engine import WorkflowEngine from services.guardian.execution.audit_trail import AuditTrail +from services.guardian.execution.authorization import AuthorizationIssuer +from services.guardian.governance.policy_engine import PolicyEngine +from services.guardian.governance.approval_gate import ApprovalGate class AutonomousExecutionLayer: + """ + Fail-closed execution boundary. + + Execution requires all of the following: + + 1. Guardian policy approval. + 2. Guardian approval gate approval. + 3. A valid Guardian-issued ExecutionAuthorization. + 4. Authority bound to the exact decision and action. + + Confidence is evidence, not execution authority. + Boolean authorization flags are not trusted. + """ def __init__(self): self.executor = ActionExecutor() + self.policy = PolicyEngine() self.approval = ApprovalGate() + self.authority = AuthorizationIssuer() self.workflow = WorkflowEngine() self.audit = AuditTrail() def run(self, decision): - - approval = self.approval.approve( - decision - ) + policy = self.policy.evaluate(decision) + approval = self.approval.request(policy) if not approval["approved"]: return { - "status": "WAITING_HUMAN_APPROVAL", - "approval": approval + "status": ( + "WAITING_HUMAN_APPROVAL" + if approval["human_required"] + else "BLOCKED" + ), + "policy": policy, + "approval": approval, + } + + decision_id = decision.get("decision_id") + action = decision.get("action") + execution_authority = decision.get("execution_authority") + + if not decision_id: + return { + "status": "BLOCKED", + "reason": "DECISION_ID_REQUIRED", + "policy": policy, + "approval": approval, + } + + if not action: + return { + "status": "BLOCKED", + "reason": "ACTION_REQUIRED", + "policy": policy, + "approval": approval, + } + + if not self.authority.verify( + execution_authority, + decision_id=decision_id, + action=action, + ): + return { + "status": "BLOCKED", + "reason": "EXECUTION_AUTHORIZATION_REQUIRED", + "policy": policy, + "approval": approval, } - result = self.executor.execute( - decision["action"] - ) + result = self.executor.execute(action) return { + "status": "EXECUTED", + "policy": policy, + "approval": approval, + "authorization": { + "authorization_id": execution_authority.authorization_id, + "issued_by": execution_authority.issued_by, + "decision_id": execution_authority.decision_id, + "action": execution_authority.action, + }, "workflow": self.workflow.build(decision), "execution": result, - "audit": self.audit.record(result) + "audit": self.audit.record(result), } diff --git a/services/guardian/tests/execution/test_execution_authority_integration.py b/services/guardian/tests/execution/test_execution_authority_integration.py new file mode 100644 index 00000000..57f25d1b --- /dev/null +++ b/services/guardian/tests/execution/test_execution_authority_integration.py @@ -0,0 +1,120 @@ +from services.guardian.execution.authorization import AuthorizationIssuer +from services.guardian.execution.execution_layer import AutonomousExecutionLayer + + +def _issue_authority(decision_id, action): + return AuthorizationIssuer().issue( + decision_id=decision_id, + action=action, + policy_status="APPROVED", + approval_status="APPROVED", + ) + + +def test_boolean_authorization_must_not_execute(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "decision_id": "decision-001", + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.10, + "execution_authorized": True, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_valid_authority_allows_execution(): + engine = AutonomousExecutionLayer() + + decision_id = "decision-002" + action = "ALLOCATE_STOCK" + + authority = _issue_authority(decision_id, action) + + result = engine.run({ + "decision_id": decision_id, + "action": action, + "confidence": 0.99, + "risk_score": 0.10, + "execution_authority": authority, + }) + + assert result["status"] == "EXECUTED" + assert result["execution"]["status"] == "EXECUTED" + + +def test_authority_bound_to_different_action_must_fail_closed(): + engine = AutonomousExecutionLayer() + + authority = _issue_authority( + "decision-003", + "ALLOCATE_STOCK", + ) + + result = engine.run({ + "decision_id": "decision-003", + "action": "DELETE_ORDER", + "confidence": 0.99, + "risk_score": 0.10, + "execution_authority": authority, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_authority_bound_to_different_decision_must_fail_closed(): + engine = AutonomousExecutionLayer() + + authority = _issue_authority( + "decision-original", + "ALLOCATE_STOCK", + ) + + result = engine.run({ + "decision_id": "decision-substituted", + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.10, + "execution_authority": authority, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_missing_authority_must_fail_closed(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "decision_id": "decision-004", + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.10, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_authority_cannot_override_high_risk_policy(): + engine = AutonomousExecutionLayer() + + decision_id = "decision-005" + action = "ALLOCATE_STOCK" + + authority = _issue_authority(decision_id, action) + + result = engine.run({ + "decision_id": decision_id, + "action": action, + "confidence": 0.99, + "risk_score": 0.95, + "execution_authority": authority, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" diff --git a/services/guardian/tests/execution/test_execution_authority_v1.py b/services/guardian/tests/execution/test_execution_authority_v1.py new file mode 100644 index 00000000..255a58cb --- /dev/null +++ b/services/guardian/tests/execution/test_execution_authority_v1.py @@ -0,0 +1,137 @@ +from datetime import datetime, timedelta, timezone + +import pytest + + +def _authority_module(): + from services.guardian.execution.authorization import ( + ExecutionAuthorization, + AuthorizationIssuer, + ) + return ExecutionAuthorization, AuthorizationIssuer + + +def test_issuer_creates_authority_bound_to_decision_and_action(): + ExecutionAuthorization, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + authority = issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="APPROVED", + approval_status="APPROVED", + ) + + assert isinstance(authority, ExecutionAuthorization) + assert authority.authorization_id + assert authority.decision_id == "decision-001" + assert authority.action == "ALLOCATE_STOCK" + assert authority.issued_by == "GUARDIAN" + assert authority.policy_status == "APPROVED" + assert authority.approval_status == "APPROVED" + + +def test_authority_for_different_action_is_invalid(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + authority = issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="APPROVED", + approval_status="APPROVED", + ) + + assert not issuer.verify( + authority, + decision_id="decision-001", + action="CHANGE_ROUTE", + ) + + +def test_authority_for_different_decision_is_invalid(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + authority = issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="APPROVED", + approval_status="APPROVED", + ) + + assert not issuer.verify( + authority, + decision_id="decision-999", + action="ALLOCATE_STOCK", + ) + + +def test_expired_authority_is_invalid(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + authority = issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="APPROVED", + approval_status="APPROVED", + ttl_seconds=-1, + ) + + assert not issuer.verify( + authority, + decision_id="decision-001", + action="ALLOCATE_STOCK", + ) + + +def test_non_authority_object_is_invalid(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + forged = { + "authorization_id": "fake", + "decision_id": "decision-001", + "action": "ALLOCATE_STOCK", + "issued_by": "GUARDIAN", + } + + assert not issuer.verify( + forged, + decision_id="decision-001", + action="ALLOCATE_STOCK", + ) + + +def test_issuer_refuses_nonapproved_policy(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + with pytest.raises(ValueError): + issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="BLOCK", + approval_status="APPROVED", + ) + + +def test_issuer_refuses_nonapproved_approval(): + _, AuthorizationIssuer = _authority_module() + + issuer = AuthorizationIssuer() + + with pytest.raises(ValueError): + issuer.issue( + decision_id="decision-001", + action="ALLOCATE_STOCK", + policy_status="APPROVED", + approval_status="DENIED", + ) diff --git a/services/guardian/tests/execution/test_execution_authorization.py b/services/guardian/tests/execution/test_execution_authorization.py new file mode 100644 index 00000000..cf5c9e7c --- /dev/null +++ b/services/guardian/tests/execution/test_execution_authorization.py @@ -0,0 +1,67 @@ +from services.guardian.execution.execution_layer import AutonomousExecutionLayer + + +def test_high_confidence_alone_must_not_authorize_execution(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_missing_authorization_must_fail_closed(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.10, + }) + + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_high_risk_must_not_execute(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.95, + }) + + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_legacy_boolean_authorization_must_not_execute(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "decision_id": "legacy-boolean-001", + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.10, + "execution_authorized": True, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" + + +def test_legacy_boolean_cannot_override_high_risk_block(): + engine = AutonomousExecutionLayer() + + result = engine.run({ + "decision_id": "legacy-boolean-high-risk-001", + "action": "ALLOCATE_STOCK", + "confidence": 0.99, + "risk_score": 0.95, + "execution_authorized": True, + }) + + assert result.get("status") != "EXECUTED" + assert result.get("execution", {}).get("status") != "EXECUTED" diff --git a/services/guardian/tests/execution/test_execution_layer.py b/services/guardian/tests/execution/test_execution_layer.py index 1c490e7a..7aeb2a37 100644 --- a/services/guardian/tests/execution/test_execution_layer.py +++ b/services/guardian/tests/execution/test_execution_layer.py @@ -1,15 +1,33 @@ +from services.guardian.execution.authorization import AuthorizationIssuer from services.guardian.execution.execution_layer import AutonomousExecutionLayer def test_execution_layer(): - engine = AutonomousExecutionLayer() - result = engine.run( - { - "action": "ALLOCATE_STOCK", - "confidence": 0.94 - } + decision_id = "decision-execution-layer-001" + action = "ALLOCATE_STOCK" + + authority = AuthorizationIssuer().issue( + decision_id=decision_id, + action=action, + policy_status="APPROVED", + approval_status="APPROVED", ) + result = engine.run({ + "decision_id": decision_id, + "action": action, + "confidence": 0.94, + "risk_score": 0.10, + "execution_authority": authority, + }) + + assert result["status"] == "EXECUTED" assert result["execution"]["status"] == "EXECUTED" + + assert result["authorization"]["authorization_id"] == ( + authority.authorization_id + ) + assert result["authorization"]["decision_id"] == decision_id + assert result["authorization"]["action"] == action diff --git a/tests/ecdt/test_application_validation.py b/tests/ecdt/test_application_validation.py new file mode 100644 index 00000000..263d0823 --- /dev/null +++ b/tests/ecdt/test_application_validation.py @@ -0,0 +1,114 @@ +import pytest + +from intelligence.ecdt import ( + ApplicationEvidence, + ApplicationValidation, + ApplicationValidationGate, + ValidationStatus, +) + + +def test_no_evidence_fails_closed(): + validator = ApplicationValidation() + + result = validator.validate("APP-001", []) + + assert result.status is ValidationStatus.INSUFFICIENT_EVIDENCE + assert result.approved is False + assert result.reasons == ("NO_EVIDENCE",) + + +def test_unverified_evidence_is_insufficient(): + validator = ApplicationValidation() + + evidence = [ + ApplicationEvidence( + application_id="APP-001", + source="wms", + facts={"quantity": 12}, + verified=False, + ) + ] + + result = validator.validate("APP-001", evidence) + + assert result.status is ValidationStatus.INSUFFICIENT_EVIDENCE + assert result.verified_evidence_count == 0 + + +def test_verified_evidence_can_be_approved(): + validator = ApplicationValidation() + + evidence = [ + ApplicationEvidence( + application_id="APP-001", + source="wms", + facts={"quantity": 12}, + verified=True, + ) + ] + + result = validator.validate("APP-001", evidence) + + assert result.status is ValidationStatus.APPROVED + assert result.approved is True + + +def test_evidence_for_other_application_is_not_counted(): + validator = ApplicationValidation() + + evidence = [ + ApplicationEvidence( + application_id="APP-OTHER", + source="wms", + verified=True, + ) + ] + + result = validator.validate("APP-001", evidence) + + assert result.status is ValidationStatus.INSUFFICIENT_EVIDENCE + assert result.evidence_count == 0 + + +def test_configurable_verified_evidence_threshold(): + validator = ApplicationValidation(minimum_verified_evidence=2) + + evidence = [ + ApplicationEvidence( + application_id="APP-001", + source="wms", + verified=True, + ) + ] + + result = validator.validate("APP-001", evidence) + + assert result.status is ValidationStatus.INSUFFICIENT_EVIDENCE + + +def test_gate_allows_only_approved_result(): + gate = ApplicationValidationGate() + + evidence = [ + ApplicationEvidence( + application_id="APP-001", + source="digital-twin", + verified=True, + ) + ] + + assert gate.allows("APP-001", evidence) is True + assert gate.allows("APP-002", evidence) is False + + +def test_empty_application_id_is_rejected(): + validator = ApplicationValidation() + + with pytest.raises(ValueError): + validator.validate("", []) + + +def test_invalid_threshold_is_rejected(): + with pytest.raises(ValueError): + ApplicationValidation(minimum_verified_evidence=0)