# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-evaluate-capability-composition-safety-when-combining-peers.md
# Remnant Warrant:    https://corpusfetch.com/c/how-must-an-instance-evaluate-capability-composition-safety-when-combining-peers.warrant.json
# Markdown SHA-256:   aa10975fbea7251a70008784ec1f4b851a79fe92d9ef750accd9c86393a98655
# Code AST SHA-256:   bb30aa386210ed1c27c33a2899bb640f518b16831cca0c7481dc2b81208f6681
# Verified Classes:   CompositionVerdict, ComposedPeer, DataFlow, Composition, CompositionPolicy, CompositionManifest, CapabilityCompositionGuard
# Defensive Guards:   0 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class CompositionVerdict(Enum):
    SINGLE_PEER = auto()
    SAFE_COMPOSITION = auto()
    EMERGENT_RISK_DETECTED = auto()
    DOMAIN_CONFLICT = auto()

@dataclass(frozen=True)
class ComposedPeer:
    peer_id: str
    domain: str
    privilege_level: int
    data_access_scope: frozenset[str]

@dataclass(frozen=True)
class DataFlow:
    from_peer: str
    to_peer: str
    data_class: str

@dataclass(frozen=True)
class Composition:
    composition_id: str
    peers: tuple[ComposedPeer, ...]
    data_flows: tuple[DataFlow, ...]

@dataclass(frozen=True)
class CompositionPolicy:
    require_same_domain: bool
    max_privilege_delta: int

@dataclass(frozen=True)
class CompositionManifest:
    verdict: CompositionVerdict
    directive: str
    manifest_digest: str

class CapabilityCompositionGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_composition(
        cls, comp: Composition, policy: CompositionPolicy
    ) -> CompositionManifest:
        
        # Gate 1: Single peer
        if len(comp.peers) <= 1:
            digest = cls._hash(json.dumps({
                "composition": comp.composition_id,
                "peers": str(len(comp.peers))
            }, sort_keys=True))
            return CompositionManifest(
                verdict=CompositionVerdict.SINGLE_PEER,
                directive=f"PROCEED: Single peer, no composition risk.",
                manifest_digest=digest
            )

        # Gate 2: Domain conflict
        if policy.require_same_domain:
            domains = {p.domain for p in comp.peers}
            if len(domains) > 1:
                return CompositionManifest(
                    verdict=CompositionVerdict.DOMAIN_CONFLICT,
                    directive=f"HALT: Peers span {len(domains)} domains {sorted(domains)}.",
                    manifest_digest=cls._hash(f"DOMAIN_CONFLICT:{len(domains)}")
                )

        # Gate 3: Privilege delta
        peer_map = {p.peer_id: p for p in comp.peers}
        for flow in comp.data_flows:
            from_p = peer_map.get(flow.from_peer)
            to_p = peer_map.get(flow.to_peer)
            if from_p and to_p:
                delta = to_p.privilege_level - from_p.privilege_level
                if delta > policy.max_privilege_delta:
                    return CompositionManifest(
                        verdict=CompositionVerdict.EMERGENT_RISK_DETECTED,
                        directive=f"HALT: Privilege delta {delta} ({from_p.peer_id}→{to_p.peer_id}) > {policy.max_privilege_delta}.",
                        manifest_digest=cls._hash(f"PRIV_ESCALATION:{flow.from_peer}")
                    )

        # Gate 4: Safe composition
        digest = cls._hash(json.dumps({
            "composition": comp.composition_id,
            "peers": tuple(sorted(p.peer_id for p in comp.peers)),
            "flows": tuple((f.from_peer, f.to_peer, f.data_class) for f in comp.data_flows)
        }, sort_keys=True))
        
        return CompositionManifest(
            verdict=CompositionVerdict.SAFE_COMPOSITION,
            directive=f"PROCEED: {len(comp.peers)} peers composed safely with {len(comp.data_flows)} flows.",
            manifest_digest=digest
        )