"""Public-only acquisition followed by frozen decisions and private auditing."""
from __future__ import annotations

from collections import defaultdict
from contextlib import ExitStack
from dataclasses import asdict, dataclass, fields
from itertools import product
import math
import random
import threading
import time
import tracemalloc
from typing import Callable

from .core import CandidateSpace, Contract, Machine
from .fixtures import catalog, make_case, oracle_plan
from .interfaces import InterfaceError, TargetProcess
from .identity import canonical_hash, scientific_view, source_hash
from .conditions import condition_key, condition_fields, normalize_fault_schedule
from .backends import CANDIDATE_BACKENDS, make_space, backend_summary

POLICIES = ("fixed", "random", "disagreement", "task", "aalpy", "counterexample", "entropy", "ec2", "branching", "lookahead", "voi")
ADMISSIONS = ("robust", "adaptive", "one_best", "unique", "fit", "identity", "abstain", "oracle", "risk_controlled")
EXECUTION_MODES = ("open_loop", "shielded", "recovery", "replan")
RECOVERY_STRATEGIES = ("stabilize", "soft_return")
PROBE_SAFETY_MODES = ('unrestricted', 'model_safe')
_RUN_LOCK = threading.RLock()
_PROCESS_SOURCE_HASH = source_hash()


class Cancelled(RuntimeError):
    pass


@dataclass(frozen=True)
class RunConfig:
    family: str = "latch"
    seed: int = 17
    implementation: str = "procedural"
    split: str = "development"
    anchors: bool = False
    budget: int = 16
    horizon: int = 4
    policy: str = "task"
    admission: str = "robust"
    execution_mode: str = "open_loop"
    temporal: bool = False
    escalate: bool = False
    probe_safety: str = 'unrestricted'
    risk_tolerance: float = 0.0
    instance_seed: int | None = None
    fault_schedule: dict | None = None
    candidate_backend: str = 'flat'
    recovery_strategy: str = 'stabilize'

    def __post_init__(self):
        if self.family == 'generated_protocol' and self.instance_seed is None:
            object.__setattr__(self, 'instance_seed', 0)

    @classmethod
    def from_dict(cls, values: dict) -> "RunConfig":
        if not isinstance(values, dict) or set(values) - {f.name for f in fields(cls)}:
            raise ValueError("Unknown run fields or non-object configuration")
        instance = cls(**values)
        instance.check()
        object.__setattr__(instance, 'fault_schedule', normalize_fault_schedule(instance.fault_schedule))
        return instance

    def check(self) -> None:
        for key, low, high in (("budget", 0, 512), ("horizon", 1, 6), ("seed", 0, 2**32 - 1)):
            value = getattr(self, key)
            if type(value) is not int or not low <= value <= high:
                raise ValueError(f"{key} must be an integer in [{low}, {high}]")
        if self.instance_seed is not None and (
                type(self.instance_seed) is not int or not 0 <= self.instance_seed <= 2**32 - 1):
            raise ValueError('instance_seed must be an unsigned 32-bit integer or null')
        if self.instance_seed is not None and self.family != 'generated_protocol':
            raise ValueError('instance_seed is supported only by generated_protocol')
        if type(self.anchors) is not bool:
            raise ValueError("anchors must be a boolean")
        if type(self.temporal) is not bool:
            raise ValueError("temporal must be a boolean")
        if type(self.escalate) is not bool:
            raise ValueError("escalate must be a boolean")
        if self.probe_safety not in PROBE_SAFETY_MODES:
            raise ValueError('Unknown probe_safety mode')
        if (type(self.risk_tolerance) not in (int, float) or not 0 <= self.risk_tolerance < 1
                or not math.isfinite(self.risk_tolerance)):
            raise ValueError('risk_tolerance must be a finite number in [0, 1)')
        if self.risk_tolerance and self.admission != 'risk_controlled':
            raise ValueError('Positive risk_tolerance requires risk_controlled admission')
        if self.policy == 'aalpy' and self.probe_safety != 'unrestricted':
            raise ValueError('The AALpy baseline does not support model_safe calibration')
        if self.escalate and self.temporal:
            raise ValueError("escalate cannot be combined with temporal=True (starts at static and escalates)")
        if self.execution_mode not in EXECUTION_MODES:
            raise ValueError("Unknown execution mode")
        if self.family not in {f["id"] for f in catalog()}:
            raise ValueError("Unknown fixture family")
        if self.implementation not in ("table", "procedural") or self.split not in ("development", "evaluation"):
            raise ValueError("Unknown implementation or split")
        if self.policy not in POLICIES or self.admission not in ADMISSIONS:
            raise ValueError("Unknown probe/admission policy")
        if self.candidate_backend not in CANDIDATE_BACKENDS:
            raise ValueError('Unknown candidate_backend')
        if self.recovery_strategy not in RECOVERY_STRATEGIES:
            raise ValueError('Unknown recovery strategy')
        if self.admission == 'adaptive' and self.execution_mode not in ('shielded', 'recovery'):
            raise ValueError('Adaptive admission requires shielded or recovery execution')
        if self.policy == 'aalpy' and self.temporal:
            raise ValueError('The AALpy baseline does not support temporal adapters')
        if self.policy == 'aalpy' and self.admission == 'adaptive':
            raise ValueError('The AALpy baseline does not support adaptive admission')
        if self.policy == 'aalpy' and self.escalate:
            raise ValueError('The AALpy baseline does not support model escalation')
        normalize_fault_schedule(self.fault_schedule)


def _check_cancel(cancel: Callable[[], bool] | None) -> None:
    if cancel and cancel():
        raise Cancelled("Run cancelled before scientific completion")


def fit_evidence(probes: list[dict]) -> list[dict]:
    evidence = []
    for item in probes:
        if item['complete']:
            evidence.append(item)
        elif item.get('outputs'):
            evidence.append({'word': item['word'][:len(item['outputs'])],
                             'outputs': item['outputs'], 'complete': True})
    return evidence


def fit_diagnostics(space: CandidateSpace, evidence: list[dict], anchors: dict) -> dict:
    """Residual of the forced best-fit map, split into raw-output and anchor terms.

    Uses the same deterministic tie-break as CandidateSpace.plan('fit'), so a
    nonzero residual distinguishes forced fitting from consistent survivors.
    """
    losses = space.losses(evidence, anchors)
    chosen = min(losses, key=lambda i: (losses[i], i))
    raw = sum(p != o for item in evidence
              for p, o in zip(space.predict(chosen, tuple(item["word"])), item["outputs"]))
    tied = sum(1 for value in losses.values() if value == losses[chosen])
    return {"chosen_adapter_id": chosen, "chosen_adapter": space.adapters[chosen].to_dict(),
            "raw_mismatches": raw, "anchor_mismatches": losses[chosen] - raw,
            "total_residual": losses[chosen], "zero_residual": losses[chosen] == 0,
            "maps_tied_at_minimum": tied,
            "raw_symbols_compared": sum(len(item["outputs"]) for item in evidence)}


def _acquire(public: dict, policy: str, budget: int, query: Callable, counts: dict,
             cancel: Callable[[], bool] | None, progress: Callable | None, temporal: bool = False,
             admission: str = 'robust', escalate: bool = False, probe_safety: str = 'unrestricted',
             step: Callable | None = None, reset: Callable | None = None, candidate_backend: str = 'flat'):
    """This function receives no Case, private map, target states or family label."""
    space = make_space(Machine.from_dict(public["reference"]), Contract.from_dict(public["contract"]), candidate_backend, temporal)
    anchors = {int(k): int(v) for k, v in public.get("anchors", {}).items()}
    probes: list[dict] = []
    survivors = space.survivors(probes, anchors)
    seen: set[tuple] = set()
    error = None
    escalated = False
    escalation_history = []
    safe_mode = probe_safety == 'model_safe'
    # An enabled larger class must be guarded before it becomes necessary for
    # fitting. Static predictions alone cannot protect a delayed target.
    safety_space = (make_space(space.machine, space.contract, candidate_backend, temporal=True)
                    if safe_mode and escalate else space)
    stop_reason = 'CALL_CAP'
    # `budget` is the shared system-call cap: every reset and every action
    # attempt counts, so all methods compare at the same total target access.
    while counts["target_calls"] < budget:
        _check_cancel(cancel)
        if admission == 'adaptive':
            from .adaptive import synthesize_policy
            ready = synthesize_policy(space, survivors, cancel=cancel)['admitted']
        else:
            ready = space.plan(survivors, "robust")["admitted"]
        if ready:
            stop_reason = 'CONTRACT_READY'
            break
        remaining_actions = budget - counts["target_calls"] - 1  # one reset per query
        if remaining_actions < 1:
            break
        selection_space = safety_space if safe_mode else space
        selection_ids = (selection_space.survivors(fit_evidence(probes), anchors)
                         if selection_space is not space else survivors)
        before = len(survivors)

        if policy == 'branching':
            # Interactive within-query branching calibration:
            # Observes intermediate outputs and selects each subsequent action conditionally.
            word = []
            outputs = []
            complete = False
            try:
                if reset is not None:
                    reset()
                elif hasattr(query, '__self__') and hasattr(query.__self__, 'reset'):
                    query.__self__.reset()
            except InterfaceError as exc:
                error = {"code": exc.code, "message": str(exc)}
                stop_reason = 'INTERFACE_ERROR'
                break

            current_ids = list(selection_ids)
            max_query_depth = min(4, remaining_actions)

            while len(word) < max_query_depth and counts["target_calls"] < budget:
                _check_cancel(cancel)
                if len(current_ids) <= 1:
                    break

                if budget - counts["target_calls"] < 1:
                    break

                next_action = selection_space.choose_branching_step(
                    current_ids, tuple(word), safe_only=safe_mode, safe_ids=selection_ids,
                    seed=92821 + len(probes) + len(word)
                )
                if next_action is None:
                    break

                try:
                    if step is not None:
                        out = step(next_action)
                    elif hasattr(query, '__self__') and hasattr(query.__self__, 'step'):
                        out = query.__self__.step(next_action)
                    else:
                        out = query([next_action])[-1]
                    word.append(next_action)
                    outputs.append(out)
                except InterfaceError as exc:
                    error = {"code": exc.code, "message": str(exc)}
                    break

                current_ids = [
                    i for i in current_ids
                    if selection_space.predict(i, tuple(word)) == tuple(outputs)
                ]
                if len(current_ids) <= 1:
                    break

            complete = error is None
            if word:
                seen.add(tuple(word))
                entry = {"word": list(word), "outputs": list(outputs), "complete": complete,
                         "survivors_before": before}
                if safe_mode:
                    entry['safety'] = {'model_class': 'temporal' if selection_space.temporal else 'static',
                                       'candidate_count': len(selection_ids), 'model_safe': True}
                probes.append(entry)
                survivors = space.survivors(fit_evidence(probes), anchors)
                entry["survivors_after"] = len(survivors)
            else:
                stop_reason = 'NO_ELIGIBLE_SAFE_PROBE' if safe_mode else 'NO_INFORMATIVE_PROBE'
                break
        else:
            planning = None
            if policy == 'lookahead':
                from .experiments import synthesize_experiment
                from .experiment_checks import verify_experiment_tree
                planning = synthesize_experiment(selection_space, selection_ids,
                                                budget - counts['target_calls'], seen=seen,
                                                safe_only=safe_mode, cancel=cancel)
                word = tuple(planning['word']) if planning['word'] is not None else None
                if word is not None:
                    checked = verify_experiment_tree(public['reference'], public['contract'], planning['tree'],
                                                     temporal=selection_space.temporal,
                                                     evidence=fit_evidence(probes), anchors=anchors,
                                                     budget=planning['remaining_calls'], safe_only=safe_mode)
                    if not checked['valid']:
                        raise RuntimeError('Independent experiment check failed: ' + checked['reason'])
                    for key, value in planning['objective'].items():
                        if checked[key] != value:
                            raise RuntimeError('Independent experiment objective differs from synthesis')
                    planning.update(certificate=checked, model_class='temporal' if selection_space.temporal else 'static',
                                    safe_only=safe_mode)
            else:
                word = selection_space.choose_probe(selection_ids, policy, remaining_actions, seen,
                                                    seed=92821 + len(probes), safe_only=safe_mode)
            if word is None:
                stop_reason = 'NO_ELIGIBLE_SAFE_PROBE' if safe_mode else 'NO_INFORMATIVE_PROBE'
                break
            seen.add(word)
            outputs = []
            complete = False
            try:
                outputs = query(word)
                complete = True
            except InterfaceError as exc:
                error = {"code": exc.code, "message": str(exc)}
                outputs = getattr(exc, 'trace', {}).get('outputs', [])
            entry = {"word": list(word), "outputs": list(outputs), "complete": complete,
                     "survivors_before": before}
            if planning is not None:
                entry['planning'] = planning
            if safe_mode:
                entry['safety'] = {'model_class': 'temporal' if selection_space.temporal else 'static',
                                   'candidate_count': len(selection_ids), 'model_safe': True}
            probes.append(entry)
            survivors = space.survivors(fit_evidence(probes), anchors)
            entry["survivors_after"] = len(survivors)
        if progress:
            progress(counts["target_calls"], budget, f"{len(survivors)} mappings remain")
        if error:
            stop_reason = 'INTERFACE_ERROR'
            break
        if not survivors:
            if escalate and not space.temporal:
                # Autonomous Model-Class Escalation: static refuted -> escalate to temporal
                previous_work = dict(space.work)
                space = make_space(space.machine, space.contract, candidate_backend, temporal=True)
                # Rebuilding a larger hypothesis class does not erase the work
                # already spent refuting the smaller one.
                space.work.update(previous_work)
                survivors = space.survivors(fit_evidence(probes), anchors)
                entry["survivors_after"] = len(survivors)
                escalated = True
                escalation_history.append({"step": len(probes), "from": "static", "to": "temporal",
                                           "work_before_escalation": previous_work,
                                           "survivors_after_escalation": len(survivors)})
                if progress:
                    progress(counts["target_calls"], budget, f"Escalated to temporal model class: {len(survivors)} mappings")
                if not survivors:
                    stop_reason = 'MODEL_CONTRADICTED'
                    break
            else:
                stop_reason = 'MODEL_CONTRADICTED'
                break
    if safety_space is not space and safe_mode:
        # Count the independent envelope selection work as well as fitting and
        # any discarded static search work. No target calls are hidden here.
        for key, value in safety_space.work.items():
            space.work[key] = space.work.get(key, 0) + value
    escalation_info = {
        "enabled": escalate,
        "escalated": escalated,
        "escalations": len(escalation_history),
        "history": escalation_history,
        "model_class": "temporal" if space.temporal else "static",
        "structural_complexity": 1 if space.temporal else 0,
        'calibration': {'mode': probe_safety, 'stop_reason': stop_reason,
                        'safety_model_class': 'temporal' if (temporal or (safe_mode and escalate)) else 'static',
                        'scope': 'Bounded probe search within the remaining call cap; model-safe means hazard-free '
                                 'under every evidence-compatible enabled model, conditional on truthful semantics.'},
    }
    return space, survivors, probes, error, escalation_info


def run(config: RunConfig, cancel: Callable[[], bool] | None = None, progress: Callable | None = None) -> dict:
    """A single serialized experiment; targets are always closed on cancellation."""
    config = RunConfig.from_dict(asdict(config))
    with _RUN_LOCK:
        return _run(config, cancel, progress)


def _run(config: RunConfig, cancel: Callable[[], bool] | None, progress: Callable | None) -> dict:
    started, cpu = time.perf_counter(), time.process_time()
    frozen_source = source_hash()
    if frozen_source != _PROCESS_SOURCE_HASH:
        raise RuntimeError('Source changed since runtime import; restart CROSSWIRE before creating evidence')
    _check_cancel(cancel)
    case = make_case(config.family, config.seed, config.implementation, config.split, config.anchors, config.horizon,
                     instance_seed=config.instance_seed, fault_schedule=config.fault_schedule)
    public = case.public
    counts = dict(reset_attempts=0, step_attempts=0, reset_successes=0, step_successes=0, target_calls=0)
    baseline = {"name": config.policy, "privileged": config.admission == "oracle"}
    resources = ExitStack()
    target = None
    if not tracemalloc.is_tracing():
        tracemalloc.start()
    tracemalloc.reset_peak()
    try:
        temporal = config.temporal
        if config.policy == "aalpy" and config.admission != "oracle":
            from .baselines import acquire_aalpy
            target = resources.enter_context(TargetProcess(case.private))
            space, survivors, probes, error, detail = acquire_aalpy(public, config.budget, target, cancel,
                                                                  candidate_backend=config.candidate_backend)
            counts = dict(target.counts)
            baseline.update(detail)
            escalation_info = {
                "enabled": False, "escalated": False, "escalations": 0, "history": [],
                "model_class": "static", "structural_complexity": 0,
            }
        elif config.budget and config.admission != "oracle":
            target = resources.enter_context(TargetProcess(case.private))
            space, survivors, probes, error, escalation_info = _acquire(
                public, config.policy, config.budget, target.query, target.counts, cancel, progress,
                temporal=temporal, admission=config.admission, escalate=config.escalate,
                probe_safety=config.probe_safety,
                step=target.step, reset=target.reset, candidate_backend=config.candidate_backend
            )
            counts = dict(target.counts)
        else:
            space = make_space(Machine.from_dict(public["reference"]), Contract.from_dict(public["contract"]),
                               config.candidate_backend, temporal=temporal)
            survivors = space.survivors([], {int(k): int(v) for k, v in public.get("anchors", {}).items()})
            probes, error = [], None
            escalation_info = {
                "enabled": config.escalate, "escalated": False, "escalations": 0, "history": [],
                "model_class": "temporal" if space.temporal else "static",
                "structural_complexity": 1 if space.temporal else 0,
            }
        _check_cancel(cancel)
        if config.admission == "oracle":
            word = oracle_plan(case.private, config.horizon)
            decision = {"status": "ORACLE_DIAGNOSTIC", "admitted": word is not None,
                        "word": word, "candidate_count": len(survivors)}
        elif config.admission == 'adaptive':
            from .adaptive import synthesize_policy
            decision = synthesize_policy(space, survivors, cancel=cancel)
        else:
            anchors = {int(k): int(v) for k, v in public.get("anchors", {}).items()}
            decision = space.plan(survivors, config.admission, transcripts=fit_evidence(probes),
                                  anchors=anchors, risk_tolerance=config.risk_tolerance)
            if config.admission == "fit":
                baseline["fit_diagnostics"] = fit_diagnostics(space, fit_evidence(probes), anchors)
        if error:
            decision = {"status": "INTERFACE_ERROR", "admitted": False, "word": None,
                        "candidate_count": len(survivors), "error": error}
        elif config.policy == 'aalpy' and not baseline.get('model_complete') and config.admission != 'oracle':
            decision = {'status': 'UNRESOLVED', 'admitted': False, 'word': None,
                        'candidate_count': 0, 'reason': 'Budget exhausted before a complete classical model'}
        decision['calibration'] = escalation_info.pop('calibration', {
            'mode': config.probe_safety,
            'stop_reason': 'CLASSICAL_BASELINE' if config.policy == 'aalpy' else 'NOT_REQUESTED',
            'safety_model_class': 'temporal' if (config.temporal or config.escalate) else 'static',
            'scope': 'Calibration safety is conditional on the declared finite enabled model and truthful anchors.'})
        decision["escalation"] = escalation_info
        if decision["status"] == "FAMILY_CONTRADICTED" and config.escalate:
            static_count = len(space.adapters) // (1 + space.machine.outputs) if space.temporal else len(space.adapters)
            decision["structural_contradiction"] = {
                "classes_refuted": [f"static ({static_count} maps)"] +
                                   ([f"temporal ({len(space.adapters)} latency maps)"] if space.temporal else []),
                "evidence_length": len(probes),
                "scope": "No adapter in the checked finite hypothesis class can explain the public evidence."
            }
        from .certificates import check_result
        candidates = [space.adapters[identifier].to_dict() for identifier in survivors]
        evidence_certificate = check_result({
            'schema': 'crosswire-result-v1', 'configuration': asdict(config),
            'reference': public['reference'], 'contract': public['contract'],
            'anchors': public.get('anchors', {}), 'target_id': public['target_id'],
            'decision': decision, 'candidates': candidates, 'probes': probes,
        })
        if not evidence_certificate['valid']:
            raise RuntimeError('Independent public-evidence check failed: ' + evidence_certificate['reason'])
        ambiguity = space.ambiguity(survivors)
        backend = backend_summary(space, survivors)
        backend['measurement_stage'] = 'after_public_check_before_deployment_and_private_audit'
        # The digest is constructed before importing/calling the private auditor.
        frozen = canonical_hash({"decision": decision, "probes": probes, "reference": public["reference"],
                                 "contract": public["contract"], "target_id": public['target_id'],
                                 'anchors': public.get('anchors', {}), "survivors": list(survivors),
                                 'evidence_certificate': evidence_certificate})
        deployment = {'word': decision['word'], 'outputs': [], 'complete': False,
                      'mode': config.execution_mode,
                      'error': None, 'target_steps': 0, 'resets': 0, 'target_calls': 0}
        if decision['admitted']:
            _check_cancel(cancel)
            if target is None:
                target = resources.enter_context(TargetProcess(case.private))
            if config.admission == 'adaptive':
                from .adaptive import execute_policy
                deployment = execute_policy(target, decision['policy'], space, survivors, cancel,
                                            fault_schedule=config.fault_schedule,
                                            mode='recovery' if config.execution_mode == 'recovery' else 'adaptive', recovery_strategy=config.recovery_strategy)
            else:
                from .shield import execute_deployment
                deployment = execute_deployment(target, decision['word'], space, survivors, space.contract,
                                                mode=config.execution_mode, cancel=cancel,
                                                fault_schedule=config.fault_schedule, recovery_strategy=config.recovery_strategy)
        from .audit import evaluate
        audit = evaluate(case, space, survivors, decision, probes, deployment)
        peak = tracemalloc.get_traced_memory()[1]
        work = dict(space.work)
        result = {
            "schema": "crosswire-result-v1", "configuration": asdict(config),
            'backend': backend,
            'source_hash': frozen_source,
            "reference": public["reference"], "contract": public["contract"],
            "target_id": public["target_id"], "parent_instance_id": public["parent_instance_id"],
            "anchors": public.get("anchors", {}), "decision": decision, "ambiguity": ambiguity,
            "candidates": candidates, "evidence_certificate": evidence_certificate,
            "probes": probes, "decision_frozen_hash": frozen, 'deployment': deployment, "audit": audit,
            "metrics": {"target_steps": counts["step_attempts"], "resets": counts["reset_attempts"],
                        "target_calls": counts["target_calls"], **work,
                        'anchor_facts_supplied': len(public.get('anchors', {})),
                        "escalations": escalation_info["escalations"],
                        "escalation_level": escalation_info["structural_complexity"],
                        "parent_cpu_seconds": time.process_time() - cpu,
                        "wall_seconds": time.perf_counter() - started, "parent_peak_traced_bytes": peak,
                        "audit_product_pairs": audit["audit_product_pairs"],
                        "audit_execution_steps": audit["audit_execution_steps"]},
            "baseline": baseline,
        }
        if 'instance_design' in public:
            result['instance_design'] = public['instance_design']
        if source_hash() != frozen_source:
            raise RuntimeError('Source changed during run; discard mixed-revision evidence')
        result["scientific_hash"] = canonical_hash(scientific_view(result))
        return result
    finally:
        resources.close()
        tracemalloc.stop()


def _condition(configuration: dict) -> tuple:
    """Comparable scientific conditions, with defaults for legacy row inputs."""
    return condition_key(configuration)


def summarize(rows: list[dict]) -> list[dict]:
    groups = defaultdict(list)
    scopes = {f["id"]: f["scope"] for f in catalog()}
    for row in rows:
        cfg = row["configuration"]
        groups[(cfg["policy"] + "/" + cfg["admission"], cfg["family"], cfg["budget"],
                _condition(cfg))].append(row)
    result = []
    for (method, family, budget, conditions), members in sorted(groups.items()):
        n = len(members)
        admitted = sum(r["decision"]["admitted"] for r in members)
        success = sum(r["audit"]["contract_satisfied"] is True for r in members)
        harm = sum(r["audit"]["contract_satisfied"] is False for r in members)
        feasible = sum(r["audit"]["oracle_feasible"] for r in members)
        abstained = n - admitted
        result.append({
            "method": method, "family": family, "scope": scopes[family], "budget": budget, "requests": n,
            **condition_fields(conditions),
            "admitted": admitted, "successful": success, "harmful": harm,
            "hazardous": sum(r["audit"]["hazard"] is True for r in members),
            "hazards_averted": sum(r["audit"].get("hazard_averted") is True for r in members),
            "shield_interventions": sum(r["audit"].get("shield_intervened") is True for r in members),
            "calibration_hazardous_probes": sum(r['audit']['calibration_hazardous_probes'] for r in members),
            "calibration_hazardous_runs": sum(r['audit']['calibration_hazardous_probes'] > 0 for r in members),
            "mean_target_calls": sum(r['metrics']['target_calls'] for r in members) / n,
            "mean_deployment_calls": sum(r['deployment']['target_calls'] for r in members) / n,
            "mean_total_calls": sum(r['metrics']['target_calls'] + r['deployment']['target_calls'] for r in members) / n,
            "interface_errors": sum(r["decision"]["status"] == "INTERFACE_ERROR"
                                    or bool((r.get("deployment") or {}).get("error")) for r in members),
            "acquisition_interface_errors": sum(r["decision"]["status"] == "INTERFACE_ERROR" for r in members),
            "deployment_interface_errors": sum(bool((r.get("deployment") or {}).get("error")) for r in members),
            "escalated_runs": sum(r["metrics"].get("escalations", 0) > 0 for r in members),
            "recovered_runs": sum((r.get("deployment") or {}).get("recovery", {}).get("succeeded") is True for r in members),
            "abstained": abstained, "oracle_feasible": feasible,
            "admission_coverage": admitted / n, "useful_coverage": success / n,
            "feasible_useful_coverage": success / feasible if feasible else None,
            "selective_harm": harm / admitted if admitted else None, "population_harm": harm / n,
            "mean_target_steps": sum(r["metrics"]["target_steps"] for r in members) / n,
            "mean_resets": sum(r["metrics"]["resets"] for r in members) / n,
            "parent_clusters": len({r["parent_instance_id"] for r in members}),
            "utility": {str(cost): (success - harm - cost * abstained) / n for cost in (0, 0.1, 0.5, 1)},
        })
    return result


def paired_comparisons(rows: list[dict]) -> list[dict]:
    """Exploratory cluster-paired coverage differences, no power claim.

    Every comparison fixes the budget, execution settings, semantic grounding,
    horizon and split. Label seeds and implementations are matched within each
    semantic parent before whole-parent resampling. Duplicate cells are refused
    instead of allowing their last value to silently determine the contrast.
    Policy pairs also fix admission; admission contrasts use robust as reference.
    """
    scopes = {f["id"]: f["scope"] for f in catalog()}
    comparisons = {'policy': defaultdict(dict), 'admission': defaultdict(dict)}
    seen = set()
    for row in rows:
        c = row["configuration"]
        # Out-of-family/control cases are never pooled into primary in-family effect.
        if scopes.get(c["family"]) != "in_family":
            continue
        conditions = _condition(c)
        cell = (c['family'], c['seed'], c['implementation'], c.get('instance_seed'))
        parent = row['parent_instance_id']
        identity = (c['budget'], conditions, parent, cell, c['policy'], c['admission'])
        if identity in seen:
            raise ValueError('Duplicate matched analysis cell; each condition/method must occur once')
        seen.add(identity)
        outcome = int(row['audit']['contract_satisfied'] is True)
        key = (c['budget'], conditions, c['admission'], parent, cell)
        comparisons['policy'][key][c['policy']] = outcome
        key = (c['budget'], conditions, c['policy'], parent, cell)
        comparisons['admission'][key][c['admission']] = outcome

    output = []
    for axis, cells in comparisons.items():
        strata = sorted({key[:3] for key in cells})
        for budget, conditions, fixed in strata:
            stratum = (budget, conditions, fixed)
            methods = sorted({method for key, values in cells.items() if key[:3] == stratum for method in values})
            if axis == 'policy':
                # Preserve task-relative signs while making arbitrary selected
                # policy pairs (e.g. entropy vs fixed) analyzable too.
                ordered = (['task'] if 'task' in methods else []) + [m for m in methods if m != 'task']
                contrasts = [(a, b) for i, a in enumerate(ordered) for b in ordered[i + 1:]]
            else:
                contrasts = [('robust', m) for m in methods if m != 'robust']
            for reference, method in contrasts:
                per_parent = defaultdict(list)
                families = set()
                for key, values in cells.items():
                    if key[:3] != stratum or reference not in values or method not in values:
                        continue
                    parent, cell = key[3:]
                    difference = values[reference] - values[method]
                    per_parent[parent].append(difference if axis == 'policy' else -difference)
                    families.add(cell[0])
                if not per_parent:
                    continue
                delta = [sum(v) / len(v) for _, v in sorted(per_parent.items())]
                rng = random.Random(68219)
                draws = sorted(sum(rng.choice(delta) for _ in delta) / len(delta) for _ in range(2000))
                output.append({
                    'contrast': f'{reference} - {method}' if axis == 'policy' else f'{method} - robust',
                    'comparison_axis': axis,
                    'fixed_admission' if axis == 'policy' else 'fixed_policy': fixed,
                    'outcome': 'useful_coverage', 'budget': budget,
                    **condition_fields(conditions),
                    'budget_unit': 'target calls (resets + actions)',
                    'parent_clusters': len(delta),
                    'matched_cells': sum(len(v) for v in per_parent.values()),
                    'families_pooled': sorted(families),
                    'mean_delta': sum(delta) / len(delta),
                    'descriptive_95_interval': [draws[49], draws[1949]],
                    'scope': 'exploratory parent-cluster paired bootstrap within identical call cap and conditions; '
                             'in-family fixtures pooled; few clusters, no confirmatory claim',
                })
    return output


STUDY_DEFAULTS = {"families": ["latch", "symmetric", "neutral"], "seeds": [17], "budgets": [0, 8],
                  "policies": ["random", "task"], "admissions": ["robust"], "implementations": ["procedural"],
                  'instance_seeds': [None], 'fault_schedules': [None], 'candidate_backends': ['flat']}


def study_configs(spec: dict) -> list[RunConfig]:
    if not isinstance(spec, dict):
        raise ValueError("Study must be an object")
    allowed = {"families", "seeds", "budgets", "policies", "admissions", "implementations",
               "anchors", "split", "horizon", "execution_mode", "temporal", "escalate", "probe_safety", "risk_tolerance",
               'instance_seeds', 'fault_schedules', 'candidate_backends', 'recovery_strategy'}
    if set(spec) - allowed:
        raise ValueError("Unknown study fields")
    dimensions = []
    for key, default in STUDY_DEFAULTS.items():
        values = spec.get(key, default)
        if not isinstance(values, list) or not 1 <= len(values) <= 64:
            raise ValueError(f"{key} must contain 1-64 entries")
        if key == 'fault_schedules':
            values = [normalize_fault_schedule(value) for value in values]
        if len({canonical_hash(v) for v in values}) != len(values):
            raise ValueError(f"Duplicate {key} inflate repeated observations")
        dimensions.append(values)
    total = 1
    for values in dimensions:
        total *= len(values)
    if total > 512:
        raise ValueError("Study exceeds 512 runs; split the protocol explicitly")
    configs = [RunConfig.from_dict(dict(zip(("family", "seed", "budget", "policy", "admission", "implementation", 'instance_seed', 'fault_schedule', 'candidate_backend'), values),
                                         anchors=spec.get("anchors", False), split=spec.get("split", "development"),
                                         horizon=spec.get("horizon", 4),
                                         execution_mode=spec.get("execution_mode", "open_loop"),
                                         temporal=spec.get("temporal", False),
                                         escalate=spec.get("escalate", False),
                                         probe_safety=spec.get('probe_safety', 'unrestricted'),
                                         risk_tolerance=spec.get('risk_tolerance', 0.0), recovery_strategy=spec.get('recovery_strategy', 'stabilize'))) for values in product(*dimensions)]
    if len({canonical_hash(asdict(cfg)) for cfg in configs}) != len(configs):
        raise ValueError('Duplicate effective study configuration; null generated instance and zero are identical')
    return configs


def run_study(spec: dict, cancel: Callable[[], bool] | None = None, progress: Callable | None = None) -> dict:
    configs = study_configs(spec)
    frozen_source = source_hash()
    protocol = {**STUDY_DEFAULTS, **spec, "exploratory": True,
                "analysis_unit": "underlying semantic parent instance; label permutations and implementations clustered",
                "pre_execution_source_hash": frozen_source}
    protocol_hash = canonical_hash(protocol)
    rows = []
    for i, cfg in enumerate(configs):
        _check_cancel(cancel)
        if progress:
            progress(i, len(configs), f"{cfg.family}: {cfg.policy}/{cfg.admission}, budget {cfg.budget}")
        rows.append(run(cfg, cancel))
    if source_hash() != frozen_source:
        raise RuntimeError("Source changed during study; results cannot be labeled frozen")
    value = {"schema": "crosswire-study-v1", "protocol": protocol, "protocol_hash": protocol_hash,
             "source_hash": frozen_source, "rows": rows, "summary": summarize(rows),
             "paired": paired_comparisons(rows)}
    from .reliability import reliability_summary
    value['reliability'] = reliability_summary(rows)
    value["scientific_hash"] = canonical_hash(scientific_view(value))
    if progress:
        progress(len(configs), len(configs), "Complete; decisions and private audits recorded")
    return value