File size: 4,242 Bytes
30f011f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """
ERE gate adapter for the BERT entailment agent.
Wraps every /verify response through the five-gate ERE protocol before
the result is returned to the caller. A verdict that fails any gate is
suppressed — the caller receives an ERE_HALT response instead.
Gates (from sovereign-engine-v2/src/tools/ere.py):
P1 -- No secrets in the output payload
P2 -- No eval / code injection in the output
P3 -- Loop safety
P4 -- No telemetry beacons
P5 -- SHA-256 audit seal (only if P1-P4 pass)
The P5 seal is stored alongside the BLAKE3 attestation in the WORM ledger,
giving each verified claim two independent cryptographic commitments:
- BLAKE3 (daemon layer, over model + threshold + score + source)
- SHA-256 ERE seal (gate layer, over agent_id + intent + verdict JSON)
"""
from __future__ import annotations
import json
import sys
import os
# Allow running from the repo root without installing
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from dataclasses import dataclass
# Import ERE from sovereign-engine-v2 if available; else use bundled copy
try:
# If sovereign-engine-v2 is on the path
from src.tools.ere import EREGate, EREResult, ere_check # type: ignore
except ImportError:
# Bundled minimal implementation (identical logic)
from agent._ere_bundled import EREGate, EREResult, ere_check # type: ignore
@dataclass
class GatedVerdict:
"""Result returned by the ERE gate layer."""
allowed: bool # True iff all five gates passed
verdict: dict | None # Original verdict payload, or None if halted
ere_seal: str | None # P5 SHA-256 seal, or None if gates failed
gate_results: dict # P1..P5 gate verdicts
violations: list[str] # Human-readable violation descriptions
def to_dict(self) -> dict:
if self.allowed:
return {
**self.verdict,
"ere_seal": self.ere_seal,
"ere_gates": self.gate_results,
}
return {
"allowed": False,
"ere_halt": True,
"violations": self.violations,
"ere_gates": self.gate_results,
}
class BERTEREGate:
"""
Wraps a raw BERT /verify response through the ERE five-gate protocol.
Usage:
gate = BERTEREGate()
raw = {"score": 0.98, "verdict": "Entailment", "hash": "a3f8..."}
gated = gate.check(
premise="The Battle of Hastings took place in 1066.",
hypothesis="Hastings occurred in 1066.",
raw_verdict=raw,
)
if not gated.allowed:
raise RuntimeError(f"ERE halt: {gated.violations}")
return gated.to_dict()
"""
def __init__(self) -> None:
self._gate = EREGate()
def check(
self,
premise: str,
hypothesis: str,
raw_verdict: dict,
) -> GatedVerdict:
"""
Run ERE gates over the serialized verdict payload.
The output that ERE inspects is the JSON-serialized verdict — any
secrets, eval patterns, or telemetry injected by a compromised model
or adversarial input would appear here.
"""
output_str = json.dumps(raw_verdict, sort_keys=True)
intent = f"entailment:{premise[:80]}|{hypothesis[:80]}"
result: EREResult = self._gate.check(
agent_id="bert-entailment-agent",
intent=intent,
output=output_str,
)
return GatedVerdict(
allowed=result.passed,
verdict=raw_verdict if result.passed else None,
ere_seal=result.seal,
gate_results=result.gates,
violations=result.violations,
)
# ── Module-level singleton ────────────────────────────────────────────────────
_default = BERTEREGate()
def gate_verdict(premise: str, hypothesis: str, raw_verdict: dict) -> GatedVerdict:
"""Module-level convenience wrapper."""
return _default.check(premise, hypothesis, raw_verdict)
|