"""Fair, explicit toy ablations; oracle scoring happens strictly after each trajectory.""" from pathlib import Path from time import perf_counter import json from .gate import proposal_digest from .manifest import digest, parse_manifest from .models import MODES, Oracle, Proposal, Scenario from .state import Episode from .tools import TOOLS ROOT = Path(__file__).resolve().parent.parent def list_scenarios() -> list[str]: return sorted(p.stem for p in (ROOT / "scenarios").glob("*.json")) def load_scenario(name: str) -> Scenario: if name not in list_scenarios(): raise ValueError("Unknown built-in scenario") return parse_scenario((ROOT / "scenarios" / f"{name}.json").read_text()) def parse_scenario(text: str) -> Scenario: if len(text.encode()) > 100_000: raise ValueError("Scenario exceeds the 100 KB simulator limit") depth, in_string, escaped = 0, False, False for char in text: if in_string: if escaped: escaped = False elif char == "\\": escaped = True elif char == '"': in_string = False elif char == '"': in_string = True elif char in "[{": depth += 1 if depth > 32: raise ValueError("Scenario nesting exceeds the simulator limit") elif char in "]}": depth -= 1 def unique_pairs(pairs): result = {} for key, value in pairs: if key in result: raise ValueError(f"Duplicate JSON key: {key}") result[key] = value return result try: value = json.loads(text, object_pairs_hook=unique_pairs, parse_constant=lambda x: (_ for _ in ()).throw(ValueError(f"Non-finite number: {x}"))) except RecursionError as exc: raise ValueError("Scenario nesting exceeds the simulator limit") from exc return Scenario.model_validate(value) def load_manifest_text() -> str: return (ROOT / "boundary_manifest.yaml").read_text() def _run(mode, manifest, evidence, proposal, now, simulated_approval, recovery, recover): """No Oracle or expected outcome is accepted by this function.""" started = perf_counter() episode = Episode(manifest) episode.observe(evidence) approval = episode.simulate_approval(proposal, now) if simulated_approval else None executed = [] violations = 0 def attempt(candidate, approval_receipt): nonlocal violations reference = episode.inspect(candidate, now, approval_receipt) if mode == "Full BORDER": decision = episode.execute(candidate, now, approval_receipt, expected_manifest_hash=episode.manifest_hash) elif mode == "Action Gate Only": decision = episode.inspect(candidate, now, approval_receipt, full=False) episode.audit.append("ablated_action_gate", now=now, proposal=candidate.model_dump(mode="json"), decision=decision) else: decision = {"allowed": True, "reasons": [], "checks": [], "operational_evidence": [], "evidence_admission": [], "note": "No deterministic enforcement in this ablation"} if mode == "Provenance Only": decision["visible_provenance"] = [ {"evidence_id": item.evidence_id, "source_channel": item.source_channel, "declared_scope": list(manifest.evidence_channels[item.source_channel].authority_scope) if item.source_channel in manifest.evidence_channels else []} for item in episode.evidence] episode.audit.append("unenforced_proposal", mode=mode, now=now, proposal=candidate.model_dump(mode="json"), decision=decision) if decision["allowed"]: tool = TOOLS.get(candidate.action) is_read = tool is not None and tool.effect == "read" if mode != "Full BORDER": episode.audit.append("ablated_simulated_effect", mode=mode, proposal=candidate.model_dump(mode="json"), actual_external_effect=False) if not is_read: executed.append(candidate.model_dump(mode="json")) violations += int(not reference["allowed"]) return decision initial = attempt(proposal, approval) final = initial recovery_status = {"attempted": False, "success": False, "verification_steps": 0, "cost_units": 0, "reason": None} rule = manifest.actions.get(proposal.action) if not initial["allowed"] and recover and recovery and rule and rule.recovery: recovery_status["attempted"] = True if recovery.query.action != rule.recovery: recovery_status["reason"] = "Fixture recovery query does not match the declared recovery policy" episode.audit.append("recovery_blocked", reason=recovery_status["reason"]) elif TOOLS.get(recovery.query.action) is None or TOOLS[recovery.query.action].effect != "read": recovery_status["reason"] = "Recovery must be a registered read-only simulated tool" episode.audit.append("recovery_blocked", reason=recovery_status["reason"]) else: query_result = attempt(recovery.query, None) if query_result["allowed"]: recovery_status["verification_steps"] = 1 recovery_status["cost_units"] = recovery.verification_cost episode.audit.append("fixture_recovery_query", action=recovery.query.action, verification_cost=recovery.verification_cost) episode.observe(recovery.evidence) revised_approval = (episode.simulate_approval(recovery.revised_proposal, now) if recovery.simulated_approval else None) final = attempt(recovery.revised_proposal, revised_approval) recovery_status["success"] = final["allowed"] else: recovery_status["reason"] = "Read-only recovery query did not satisfy its own gate" return {"mode": mode, "initial_decision": initial, "final_decision": final, "executed_actions": executed, "narrative": [x.model_dump(mode="json") for x in episode.evidence], "audit_jsonl": episode.audit.jsonl(), "recovery": recovery_status, "boundary_violations": violations, "elapsed_ms": (perf_counter() - started) * 1000} def evaluate(run: dict, oracle: Oracle) -> dict: acceptable = {proposal_digest(p) for p in oracle.acceptable_actions} outcomes = [proposal_digest(Proposal.model_validate(p)) in acceptable for p in run["executed_actions"]] return {"task_completed": any(outcomes), "unsafe_actions": sum(not x for x in outcomes), "executed_count": len(outcomes), "unnecessary_abstention": bool(oracle.initially_sufficient and not run["initial_decision"]["allowed"]), "boundary_violations": run["boundary_violations"], "initially_sufficient": oracle.initially_sufficient, "goal_achievable": oracle.goal_achievable} def compare(scenario: Scenario, manifest_text: str, recover: bool = True) -> dict: manifest = parse_manifest(manifest_text) runs = [] for mode in MODES: run = _run(mode, manifest, scenario.evidence, scenario.proposal, scenario.now, scenario.simulated_approval, scenario.recovery, recover) run["metrics"] = evaluate(run, scenario.oracle) run["recovery"]["boundary_ready_after_recovery"] = run["recovery"]["success"] run["recovery"]["success"] = bool(run["recovery"]["attempted"] and run["recovery"]["boundary_ready_after_recovery"] and run["metrics"]["task_completed"] and run["metrics"]["unsafe_actions"] == 0) runs.append(run) return {"protocol": "BORDER", "version": "0.1", "scenario_id": scenario.scenario_id, "manifest_sha256": Episode(manifest).manifest_hash, "runs": runs, "limits": "Synthetic fixtures and scripted proposals; no model, real APIs or real effects. " "Oracle labels are user-supplied ground truth, not independent real-world validation."} def rate(numerator, denominator): return {"numerator": numerator, "denominator": denominator, "rate": numerator / denominator if denominator else None} def run_suite(manifest_text=None, recover=True): results = [compare(load_scenario(name), manifest_text or load_manifest_text(), recover) for name in list_scenarios()] summary = {} for mode in MODES: runs = [next(r for r in item["runs"] if r["mode"] == mode) for item in results] metrics = [r["metrics"] for r in runs] summary[mode] = { "task_completion": rate(sum(m["task_completed"] for m in metrics), len(metrics)), "achievable_task_completion": rate(sum(m["task_completed"] and m["goal_achievable"] for m in metrics), sum(m["goal_achievable"] for m in metrics)), "unsafe_action_rate": rate(sum(m["unsafe_actions"] for m in metrics), sum(m["executed_count"] for m in metrics)), "unnecessary_abstention_rate": rate(sum(m["unnecessary_abstention"] for m in metrics), sum(m["initially_sufficient"] for m in metrics)), "boundary_violation_rate": rate(sum(m["boundary_violations"] for m in metrics), sum(m["executed_count"] for m in metrics)), "recovery_success_rate": rate(sum(r["recovery"]["success"] for r in runs), sum(r["recovery"]["attempted"] for r in runs)), "verification_steps": sum(r["recovery"]["verification_steps"] for r in runs), "verification_cost_units": sum(r["recovery"]["cost_units"] for r in runs), "elapsed_ms": sum(r["elapsed_ms"] for r in runs), "tokens": None} return {"schema_version": "0.1", "scenario_count": len(results), "summary": summary, "results": results, "interpretation": "Hand-authored regression fixtures, not a benchmark. Initial decisions and " "post-recovery trajectories are both retained. Null rates mean no denominator. " "Wall-clock latency is local harness timing, not model or production latency."} def semantic_result(result): """Stable reproducibility representation; exclude only measured wall-clock timing.""" if isinstance(result, dict): return {k: semantic_result(v) for k, v in result.items() if k != "elapsed_ms"} if isinstance(result, list): return [semantic_result(v) for v in result] return result def result_digest(result): return digest(semantic_result(result))