gatekeepers.py

from typing import Dict, Tuple


class MCPGatekeeper:
    def evaluate(self, proposal: Dict, ontology_snapshot: Dict) -> Tuple[str, str, float]:
        return "allow", "Default", 1.0


class PolicyMCP(MCPGatekeeper):
    def evaluate(self, proposal, snapshot):
        p = proposal.get("payload", {})
        scope = p.get("scope", "public")
        if scope in ["public", "governed", None]:
            return "allow", "Within policy scope", 0.95
        if scope in ["restricted", "classified"]:
            return "deny", "Scope exceeds policy", 0.98
        return "request_info", "Scope unclear", 0.60


class ComplianceMCP(MCPGatekeeper):
    def evaluate(self, proposal, snapshot):
        p_str = str(proposal.get("payload", {}))
        rules = snapshot.get("L", [])
        if any("harm" in r.lower() or "illegal" in r.lower() for r in rules):
            if any(k in p_str.lower() for k in ["harm", "exploit", "bypass"]):
                return "deny", "Compliance rule triggered", 0.95
        return "allow", "No compliance violations", 0.90


class SafetyMCP(MCPGatekeeper):
    def evaluate(self, proposal, snapshot):
        p_str = str(proposal.get("payload", {})).lower()
        bad = ["poison", "exploit", "override", "bypass", "jailbreak", "injection"]
        if any(b in p_str for b in bad):
            return "deny", "Safety invariant triggered", 0.99
        return "allow", "No safety concerns", 0.92


class ProvenanceMCP(MCPGatekeeper):
    def evaluate(self, proposal, snapshot):
        p = proposal.get("payload", {})
        if not (p.get("evidence") or p.get("source_url") or p.get("source") or proposal.get("parent_observation")):
            return "request_info", "Missing provenance", 0.70
        return "allow", "Provenance sufficient", 0.88


class MCPPanel:
    def __init__(self):
        self.gatekeepers = [PolicyMCP(), ComplianceMCP(), SafetyMCP(), ProvenanceMCP()]

    def verdict(self, proposal: Dict, ontology_snapshot: Dict) -> Dict:
        results = []
        for gk in self.gatekeepers:
            v, r, c = gk.evaluate(proposal, ontology_snapshot)
            results.append({"name": gk.__class__.__name__, "verdict": v, "reasoning": r, "confidence": c})
            if v == "deny":
                break
        final = "allow"
        if any(r["verdict"] == "deny" for r in results):
            final = "deny"
        elif any(r["verdict"] == "request_info" for r in results):
            final = "request_info"
        return {"final_verdict": final, "results": results}