File size: 3,768 Bytes
ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a ecad52f 2618f7a | 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 | """Public supplied-evidence boundary v1.1.0; approvals are never authenticated."""
from datetime import datetime, timezone
from bodulskihal_core import verify_payload, verify_compound_payload
from safe_json import strict_loads, strict_dumps
from matrix_core import parse_instant
BOUNDARY_VERSION = "1.1.0"
MAX_EVIDENCE = 1000
def _times(payload, evidence):
if "evaluation_time" in payload:
payload["evaluation_time"] = parse_instant(payload["evaluation_time"]).isoformat()
for item in evidence:
if isinstance(item, dict) and "observed_at" in item:
item["observed_at"] = parse_instant(item["observed_at"]).isoformat()
def evaluate(payload, batch=False):
"""Bounded local evaluation. Input evidence is supplied, not fetched/authenticated."""
try:
if not isinstance(payload, dict) or type(batch) is not bool:
raise TypeError("payload must be an object; batch must be bool")
# Apply the same finite-number, scalar-Unicode and resource limits to
# Python callers as the public JSON entry point.
clean = strict_loads(strict_dumps(payload))
if batch:
claims = clean.get("claims")
if not isinstance(claims, list) or not 1 <= len(claims) <= 100:
raise ValueError("Provide between 1 and 100 claims")
evidence_map = clean.get("evidence_by_claim", {})
if not isinstance(evidence_map, dict):
raise TypeError("evidence_by_claim must be an object")
if sum(len(v) if isinstance(v, list) else MAX_EVIDENCE + 1 for v in evidence_map.values()) > MAX_EVIDENCE:
raise ValueError("Too many evidence records")
_times(clean, [item for records in evidence_map.values() for item in records])
clean["human_approvals"] = {}
return verify_compound_payload(clean)
evidence = clean.get("evidence", [])
if not isinstance(evidence, list) or len(evidence) > MAX_EVIDENCE:
raise ValueError("Evidence must be a list of at most 1000 records")
_times(clean, evidence)
clean["human_approved"] = False
return verify_payload(clean)
except (ValueError, TypeError, KeyError, OverflowError, RecursionError, UnicodeError) as exc:
return {"classification": "UNKNOWN", "verdict": "BLOCK", "reason": "invalid_input", "error": str(exc)}
def evaluate_json(text, batch=False):
try:
return evaluate(strict_loads(text), batch=batch)
except (ValueError, TypeError, OverflowError, RecursionError, UnicodeError) as exc:
return {"classification": "UNKNOWN", "verdict": "BLOCK", "reason": "invalid_json", "error": str(exc)}
def example(scenario="supported", *, at=None):
"""Synthetic fixture; an explicit aware datetime supports historical replay."""
instant = datetime.now(timezone.utc) if at is None else at
if not isinstance(instant, datetime) or instant.tzinfo is None or instant.utcoffset() is None:
raise ValueError("at must be timezone-aware")
now = instant.astimezone(timezone.utc).isoformat()
payload = {
"claim": {"id": "synthetic-count", "text": "This synthetic record contains 30 questions", "type": "observation", "risk": "low"},
"evidence": [{"source_id": "synthetic-demo", "supports": True, "observed_at": now, "directness": "direct", "origin": "computed_result"}],
"policy": {"version": "3.0", "max_age_seconds": 300},
"evaluation_time": now,
}
if scenario == "unsupported":
payload["evidence"] = []
elif scenario == "contradiction":
payload["evidence"][0]["supports"] = False
elif scenario != "supported":
raise ValueError("Unknown synthetic scenario")
return payload
|