Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | """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 | |