border/simulator.py

"""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))