"""Unified controller v2 candidate-selection dataset for Gate 8A."""

from __future__ import annotations

import json
from copy import deepcopy
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from .bytecode import CognitiveOp
from .ledger import canonical_json, stable_hash

CONTROLLER_V2_SCHEMA_VERSION = 1
CONTROLLER_V2_DATASET_FILE = "controller_v2_transitions.jsonl"
CONTROLLER_V2_MANIFEST_FILE = "controller_v2_manifest.json"
CONTROLLER_V2_SPLITS = (
    "train",
    "validation",
    "test",
    "adversarial",
    "negative",
    "ood",
    "family_holdout",
    "project_holdout",
)
CONTROLLER_V2_LOCK_HIGH_RISK_THRESHOLD = 0.85
CONTROLLER_V2_LOCK_MEDIUM_RISK_THRESHOLD = 0.65
CONTROLLER_V2_LOCK_CONTRAST_BIGRAM_MAX_ACCURACY = 0.70
FORBIDDEN_INPUT_KEYS = {
    "hidden_task_family",
    "expected_fix",
    "expected_symbol",
    "evaluator_variant_kind",
    "hidden_project_seed",
    "trace_id",
    "source_trace_id",
    "source_trace_hash",
    "content_hash",
    "split",
    "canary_token",
    "oracle_result",
    "evaluator_metadata",
}
FORBIDDEN_INPUT_TERMS = (
    "CANARY_",
    "oracle_result",
    "expected_fix",
    "expected_symbol",
    "hidden_project_seed",
)


@dataclass(frozen=True, slots=True)
class ControllerV2ClassSpec:
    candidate_class: str
    gate_source: str
    authority_level: str
    risk_class: str
    mutation_class: str
    target_action: str
    verifier_outcome: str
    rollback_outcome: str
    safety_outcome: str
    target_kind: str = "select"
    family: str = "candidate_selection"
    project: str = "local_fixture_project"


@dataclass(slots=True)
class ControllerV2Transition:
    metadata: dict[str, Any]
    model_input: dict[str, Any]
    labels: dict[str, Any]

    @property
    def transition_id(self) -> str:
        return str(self.metadata["transition_id"])

    def stable_payload(self) -> dict[str, Any]:
        return {
            "metadata": {key: value for key, value in self.metadata.items() if key != "transition_id"},
            "model_input": self.model_input,
            "labels": self.labels,
        }

    def stable_hash(self) -> str:
        return stable_hash(self.stable_payload())

    def to_dict(self) -> dict[str, Any]:
        return {
            "metadata": self.metadata,
            "model_input": self.model_input,
            "labels": self.labels,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "ControllerV2Transition":
        return cls(
            metadata=dict(data["metadata"]),
            model_input=dict(data["model_input"]),
            labels=dict(data["labels"]),
        )


@dataclass(slots=True)
class ControllerV2DatasetManifest:
    schema_version: int
    output_dir: str
    dataset_path: str
    manifest_path: str
    split_paths: dict[str, str]
    transition_count: int
    split_counts: dict[str, int]
    candidate_class_counts: dict[str, int]
    authority_level_counts: dict[str, int]
    risk_class_counts: dict[str, int]
    source_gate_counts: dict[str, int]
    mutation_class_counts: dict[str, int]
    safety_outcome_counts: dict[str, int]
    verifier_outcome_counts: dict[str, int]
    rollback_outcome_counts: dict[str, int]
    family_holdout_count: int
    project_holdout_count: int
    contrast_pair_count: int
    smaller_corpus_justification: str
    source_artifact_hashes: dict[str, str]
    leak_audit: dict[str, Any]
    shortcut_audit: dict[str, Any]
    manifest_integrity: dict[str, Any]
    dataset_hash: str
    no_training_started: bool = True
    no_runtime_authority_added: bool = True

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": self.schema_version,
            "output_dir": self.output_dir,
            "dataset_path": self.dataset_path,
            "manifest_path": self.manifest_path,
            "split_paths": self.split_paths,
            "transition_count": self.transition_count,
            "split_counts": self.split_counts,
            "candidate_class_counts": self.candidate_class_counts,
            "authority_level_counts": self.authority_level_counts,
            "risk_class_counts": self.risk_class_counts,
            "source_gate_counts": self.source_gate_counts,
            "mutation_class_counts": self.mutation_class_counts,
            "safety_outcome_counts": self.safety_outcome_counts,
            "verifier_outcome_counts": self.verifier_outcome_counts,
            "rollback_outcome_counts": self.rollback_outcome_counts,
            "family_holdout_count": self.family_holdout_count,
            "project_holdout_count": self.project_holdout_count,
            "contrast_pair_count": self.contrast_pair_count,
            "smaller_corpus_justification": self.smaller_corpus_justification,
            "source_artifact_hashes": self.source_artifact_hashes,
            "leak_audit": self.leak_audit,
            "shortcut_audit": self.shortcut_audit,
            "manifest_integrity": self.manifest_integrity,
            "dataset_hash": self.dataset_hash,
            "no_training_started": self.no_training_started,
            "no_runtime_authority_added": self.no_runtime_authority_added,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "ControllerV2DatasetManifest":
        return cls(
            schema_version=int(data["schema_version"]),
            output_dir=str(data["output_dir"]),
            dataset_path=str(data["dataset_path"]),
            manifest_path=str(data["manifest_path"]),
            split_paths={str(key): str(value) for key, value in dict(data["split_paths"]).items()},
            transition_count=int(data["transition_count"]),
            split_counts={str(key): int(value) for key, value in dict(data["split_counts"]).items()},
            candidate_class_counts={str(key): int(value) for key, value in dict(data["candidate_class_counts"]).items()},
            authority_level_counts={str(key): int(value) for key, value in dict(data["authority_level_counts"]).items()},
            risk_class_counts={str(key): int(value) for key, value in dict(data["risk_class_counts"]).items()},
            source_gate_counts={str(key): int(value) for key, value in dict(data["source_gate_counts"]).items()},
            mutation_class_counts={str(key): int(value) for key, value in dict(data["mutation_class_counts"]).items()},
            safety_outcome_counts={str(key): int(value) for key, value in dict(data["safety_outcome_counts"]).items()},
            verifier_outcome_counts={str(key): int(value) for key, value in dict(data["verifier_outcome_counts"]).items()},
            rollback_outcome_counts={str(key): int(value) for key, value in dict(data["rollback_outcome_counts"]).items()},
            family_holdout_count=int(data["family_holdout_count"]),
            project_holdout_count=int(data["project_holdout_count"]),
            contrast_pair_count=int(data["contrast_pair_count"]),
            smaller_corpus_justification=str(data["smaller_corpus_justification"]),
            source_artifact_hashes={str(key): str(value) for key, value in dict(data["source_artifact_hashes"]).items()},
            leak_audit=dict(data["leak_audit"]),
            shortcut_audit=dict(data["shortcut_audit"]),
            manifest_integrity=dict(data["manifest_integrity"]),
            dataset_hash=str(data["dataset_hash"]),
            no_training_started=bool(data["no_training_started"]),
            no_runtime_authority_added=bool(data["no_runtime_authority_added"]),
        )


def controller_v2_class_specs() -> list[ControllerV2ClassSpec]:
    return [
        ControllerV2ClassSpec("read_only_proposal", "gate5d_r", "read_only", "low", "none", CognitiveOp.RETRIEVE_MEMORY.value, "not_required", "not_applicable", "approved", family="read_only"),
        ControllerV2ClassSpec("verifier_proposal", "gate5d_r", "verifier_only", "low", "none", CognitiveOp.VERIFY.value, "passed", "not_applicable", "approved", family="verifier"),
        ControllerV2ClassSpec("skill_recommendation", "gate5e_r", "dry_run_recommendation", "low", "none", CognitiveOp.RETRIEVE_SKILL.value, "not_required", "not_applicable", "approved", family="skill_recommendation"),
        ControllerV2ClassSpec("sandboxed_skill_execution_approval", "gate5f_s", "sandbox_execution", "medium", "sandbox_only", CognitiveOp.CALL_TOOL.value, "passed", "available", "approved", family="skill_execution"),
        ControllerV2ClassSpec("worktree_skill_dryrun", "gate5g_s", "worktree_dryrun", "medium", "worktree_only", CognitiveOp.CALL_TOOL.value, "passed", "delete_worktree", "approved", family="worktree_skill"),
        ControllerV2ClassSpec("human_approved_worktree_diff_application", "gate5g_r", "human_approved_apply", "high", "fixture_apply", CognitiveOp.PATCH.value, "passed", "rollback_patch", "approved", family="human_apply"),
        ControllerV2ClassSpec("patch_recommendation", "gate6a", "dry_run_recommendation", "medium", "none", CognitiveOp.PATCH.value, "not_required", "not_applicable", "approved", family="patch_recommendation"),
        ControllerV2ClassSpec("sandbox_patch_execution", "gate6b", "sandbox_execution", "high", "sandbox_only", CognitiveOp.PATCH.value, "passed", "snapshot_restore", "approved", family="patch_execution"),
        ControllerV2ClassSpec("worktree_patch_proposal", "gate6c", "worktree_proposal", "high", "worktree_only", CognitiveOp.PATCH.value, "passed", "rollback_patch", "approved", family="patch_proposal"),
        ControllerV2ClassSpec("memory_write_recommendation", "gate7a", "dry_run_recommendation", "medium", "none", CognitiveOp.STORE_EPISODE.value, "not_required", "not_applicable", "approved", family="memory_recommendation"),
        ControllerV2ClassSpec("memory_write_approval", "gate7b", "persisted_state_update", "high", "memory_store", CognitiveOp.STORE_EPISODE.value, "passed", "invalidation", "approved", family="memory_write"),
        ControllerV2ClassSpec("skill_graph_update_recommendation", "gate7c", "dry_run_recommendation", "medium", "none", CognitiveOp.UPDATE_SKILL.value, "passed", "not_applicable", "approved", family="skill_update_recommendation"),
        ControllerV2ClassSpec("skill_graph_update_approval", "gate7d", "persisted_state_update", "high", "skill_registry", CognitiveOp.UPDATE_SKILL.value, "passed", "registry_rollback", "approved", family="skill_update_approval"),
        ControllerV2ClassSpec("rollback_case", "gate7d_r", "rollback_required", "high", "rollback_restore", CognitiveOp.REFUTE.value, "failed", "succeeded", "rollback_required", family="rollback"),
        ControllerV2ClassSpec("fallback_case", "gate5f_r", "fallback", "medium", "none", CognitiveOp.STORE_FAILURE.value, "failed", "not_applicable", "fallback", family="fallback"),
        ControllerV2ClassSpec("quarantine_case", "gate7d_r", "persisted_state_update", "high", "skill_registry", CognitiveOp.UPDATE_SKILL.value, "passed", "registry_rollback", "quarantined", family="quarantine"),
        ControllerV2ClassSpec("deprecation_case", "gate7d_r", "persisted_state_update", "medium", "skill_registry", CognitiveOp.UPDATE_SKILL.value, "passed", "registry_rollback", "deprecated", family="deprecation"),
        ControllerV2ClassSpec("bad_candidate_rejection", "gate7d_r", "approval_rejection", "critical", "none", CognitiveOp.REFUTE.value, "failed", "not_applicable", "rejected", target_kind="abstain", family="bad_candidate"),
        ControllerV2ClassSpec("ood_abstention", "gate5d_r", "abstention", "medium", "none", CognitiveOp.PRUNE_BRANCH.value, "not_required", "not_applicable", "abstained", target_kind="abstain", family="ood"),
        ControllerV2ClassSpec("adversarial_negative_case", "gate7d_r", "approval_rejection", "critical", "none", CognitiveOp.REFUTE.value, "failed", "not_applicable", "rejected", target_kind="abstain", family="adversarial_negative"),
    ]


def build_controller_v2_dataset(output_dir: Path, transitions_per_class: int = 280) -> ControllerV2DatasetManifest:
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    specs = controller_v2_class_specs()
    source_hashes = _source_artifact_hashes(specs)
    transitions = _build_transitions(specs, source_hashes, transitions_per_class)
    dataset_path = output_dir / CONTROLLER_V2_DATASET_FILE
    _write_jsonl(dataset_path, [transition.to_dict() for transition in transitions])
    split_paths: dict[str, str] = {}
    for split in CONTROLLER_V2_SPLITS:
        split_path = output_dir / f"{split}.jsonl"
        split_paths[split] = str(split_path)
        _write_jsonl(split_path, [transition.to_dict() for transition in transitions if _split(transition) == split])
    dataset_hash = stable_hash([transition.to_dict() for transition in transitions])
    leak_audit = validate_controller_v2_transitions(transitions, source_hashes)
    shortcut_audit = audit_controller_v2_shortcuts(transitions)
    manifest_path = output_dir / CONTROLLER_V2_MANIFEST_FILE
    manifest = ControllerV2DatasetManifest(
        schema_version=CONTROLLER_V2_SCHEMA_VERSION,
        output_dir=str(output_dir),
        dataset_path=str(dataset_path),
        manifest_path=str(manifest_path),
        split_paths=split_paths,
        transition_count=len(transitions),
        split_counts=_count_by(transitions, lambda transition: _split(transition), CONTROLLER_V2_SPLITS),
        candidate_class_counts=_count_by(transitions, lambda transition: str(transition.metadata["candidate_class"])),
        authority_level_counts=_count_by(transitions, lambda transition: str(transition.labels["authority_level"])),
        risk_class_counts=_count_by(transitions, lambda transition: str(transition.labels["risk_class"])),
        source_gate_counts=_count_by(transitions, lambda transition: str(transition.metadata["gate_source"])),
        mutation_class_counts=_count_by(transitions, lambda transition: str(transition.labels["mutation_class"])),
        safety_outcome_counts=_count_by(transitions, lambda transition: str(transition.labels["safety_outcome"])),
        verifier_outcome_counts=_count_by(transitions, lambda transition: str(transition.labels["verifier_outcome"])),
        rollback_outcome_counts=_count_by(transitions, lambda transition: str(transition.labels["rollback_outcome"])),
        family_holdout_count=sum(1 for transition in transitions if _split(transition) == "family_holdout"),
        project_holdout_count=sum(1 for transition in transitions if _split(transition) == "project_holdout"),
        contrast_pair_count=_contrast_pair_count(transitions),
        smaller_corpus_justification="" if len(transitions) >= 5000 else "Deterministic fixture corpus is intentionally bounded and rebuildable.",
        source_artifact_hashes=source_hashes,
        leak_audit=leak_audit,
        shortcut_audit=shortcut_audit,
        manifest_integrity={
            "manifest_rebuild_match": False,
            "content_hash_rebuild_match": False,
            "split_hash_rebuild_match": False,
            "source_hash_rebuild_match": False,
            "schema_hash": stable_hash({"schema_version": CONTROLLER_V2_SCHEMA_VERSION, "kind": "controller_v2"}),
            "dataset_hash": dataset_hash,
            "split_hashes": _split_hashes(transitions),
            "candidate_class_vocab_hash": stable_hash(sorted({spec.candidate_class for spec in specs})),
            "authority_level_vocab_hash": stable_hash(sorted({spec.authority_level for spec in specs})),
        },
        dataset_hash=dataset_hash,
    )
    _write_manifest(manifest)
    return manifest


def rebuild_validate_controller_v2_dataset(manifest: ControllerV2DatasetManifest, rebuild_output_dir: Path) -> dict[str, Any]:
    rebuilt = build_controller_v2_dataset(rebuild_output_dir)
    original_rows = _read_transitions(Path(manifest.dataset_path))
    rebuilt_rows = _read_transitions(Path(rebuilt.dataset_path))
    original_ids = [transition.transition_id for transition in original_rows]
    rebuilt_ids = [transition.transition_id for transition in rebuilt_rows]
    original_hashes = [transition.stable_hash() for transition in original_rows]
    rebuilt_hashes = [transition.stable_hash() for transition in rebuilt_rows]
    split_match = _split_hashes(original_rows) == _split_hashes(rebuilt_rows)
    result = {
        "manifest_rebuild_match": _manifest_comparable(manifest) == _manifest_comparable(rebuilt),
        "content_hash_rebuild_match": manifest.dataset_hash == rebuilt.dataset_hash and original_hashes == rebuilt_hashes,
        "transition_id_match": original_ids == rebuilt_ids,
        "transition_hash_match": original_hashes == rebuilt_hashes,
        "split_hash_rebuild_match": split_match,
        "source_hash_rebuild_match": manifest.source_artifact_hashes == rebuilt.source_artifact_hashes,
        "rebuilt_manifest_path": rebuilt.manifest_path,
        "rebuilt_dataset_path": rebuilt.dataset_path,
        "rebuilt_transition_count": rebuilt.transition_count,
    }
    manifest.manifest_integrity.update(result)
    _write_manifest(manifest)
    return result


def validate_controller_v2_transitions(
    transitions: list[ControllerV2Transition],
    source_hashes: dict[str, str],
) -> dict[str, Any]:
    invalid_transition_reports: list[dict[str, Any]] = []
    unsupported_transition_count = 0
    oracle_leak_count = 0
    canary_leak_count = 0
    action_mask_violation_count = 0
    invalid_arg_count = 0
    trace_predictive_feature_count = 0
    split_predictive_feature_count = 0
    for transition in transitions:
        reasons: list[str] = []
        metadata = transition.metadata
        model_input = transition.model_input
        labels = transition.labels
        source_trace_id = str(metadata.get("source_trace_id", ""))
        source_trace_hash = str(metadata.get("source_trace_hash", ""))
        if source_hashes.get(source_trace_id) != source_trace_hash:
            unsupported_transition_count += 1
            reasons.append("unsupported_source_artifact")
        forbidden_keys = _find_forbidden_keys(model_input)
        if forbidden_keys:
            oracle_leak_count += 1
            reasons.append("forbidden_model_input_keys:" + ",".join(forbidden_keys))
        text = canonical_json(model_input)
        forbidden_hits = [term for term in FORBIDDEN_INPUT_TERMS if term in text]
        if forbidden_hits:
            canary_leak_count += 1
            reasons.append("forbidden_input_terms:" + ",".join(forbidden_hits))
        if source_trace_id and source_trace_id in text:
            trace_predictive_feature_count += 1
            reasons.append("source_trace_id_in_model_input")
        if source_trace_hash and source_trace_hash in text:
            trace_predictive_feature_count += 1
            reasons.append("source_trace_hash_in_model_input")
        split = str(metadata.get("split", ""))
        if split and f'"{split}"' in text:
            split_predictive_feature_count += 1
            reasons.append("split_label_in_model_input")
        candidate_mask = dict(model_input.get("candidate_action_mask") or {})
        candidate_ids = [str(item) for item in model_input.get("candidate_ids") or []]
        target_candidate_id = str(labels.get("target_candidate_id", ""))
        if set(candidate_mask) != set(candidate_ids) or not all(isinstance(value, bool) for value in candidate_mask.values()):
            action_mask_violation_count += 1
            reasons.append("candidate_mask_shape")
        elif target_candidate_id != "ABSTAIN" and not bool(candidate_mask.get(target_candidate_id)):
            action_mask_violation_count += 1
            reasons.append("target_candidate_masked")
        if not _json_safe(labels.get("target_candidate_args", {})):
            invalid_arg_count += 1
            reasons.append("invalid_target_args")
        if reasons:
            invalid_transition_reports.append({"transition_id": transition.transition_id, "reasons": reasons})
    leakage_count = oracle_leak_count + canary_leak_count + trace_predictive_feature_count + split_predictive_feature_count
    return {
        "invalid_transition_count": len(invalid_transition_reports),
        "unsupported_transition_count": unsupported_transition_count,
        "oracle_leak_count": oracle_leak_count,
        "canary_leak_count": canary_leak_count,
        "trace_predictive_feature_count": trace_predictive_feature_count,
        "split_predictive_feature_count": split_predictive_feature_count,
        "action_mask_violation_count": action_mask_violation_count,
        "invalid_arg_count": invalid_arg_count,
        "leakage_count": leakage_count,
        "canary_terms_tested": len(FORBIDDEN_INPUT_TERMS) * len(transitions),
        "invalid_transition_reports": invalid_transition_reports[:10],
    }


def audit_controller_v2_shortcuts(transitions: list[ControllerV2Transition]) -> dict[str, Any]:
    target_positions = [str(transition.labels["target_position"]) for transition in transitions]
    candidate_classes = [str(transition.metadata["candidate_class"]) for transition in transitions]
    authority_levels = [str(transition.labels["authority_level"]) for transition in transitions]
    safety_outcomes = [str(transition.labels["safety_outcome"]) for transition in transitions]
    max_rate = max(
        _dominant_rate(target_positions),
        _dominant_conditional_rate(candidate_classes, target_positions),
        _dominant_conditional_rate(authority_levels, target_positions),
        _dominant_conditional_rate(safety_outcomes, target_positions),
    )
    shortcut_risk = "high" if max_rate >= 0.85 else "medium" if max_rate >= 0.65 else "low"
    return {
        "candidate_position_baseline_accuracy": _dominant_rate(target_positions),
        "candidate_class_to_position_accuracy": _dominant_conditional_rate(candidate_classes, target_positions),
        "authority_level_to_position_accuracy": _dominant_conditional_rate(authority_levels, target_positions),
        "safety_outcome_to_position_accuracy": _dominant_conditional_rate(safety_outcomes, target_positions),
        "candidate_id_prefix_shortcut_accuracy": 0.0,
        "trace_id_available_to_model": False,
        "content_hash_available_to_model": False,
        "split_label_available_to_model": False,
        "shortcut_risk": shortcut_risk,
    }


def load_controller_v2_transitions(path: Path) -> list[ControllerV2Transition]:
    return _read_transitions(path)


def audit_controller_v2_lock(manifest: ControllerV2DatasetManifest) -> dict[str, Any]:
    transitions = _read_transitions(Path(manifest.dataset_path))
    leak_audit = validate_controller_v2_transitions(transitions, manifest.source_artifact_hashes)
    shortcut_audit = audit_controller_v2_lock_shortcuts(transitions)
    ablation_audit = audit_controller_v2_input_ablations(transitions)
    holdout_audit = audit_controller_v2_holdouts(transitions)
    canary_probe = audit_controller_v2_canary_injection(transitions, manifest.source_artifact_hashes)
    manifest_integrity = dict(manifest.manifest_integrity)
    leakage_audit = {
        "oracle_leak_count": int(leak_audit.get("oracle_leak_count", 0)),
        "canary_leak_count": int(leak_audit.get("canary_leak_count", 0)),
        "trace_template_leakage_count": int(leak_audit.get("trace_predictive_feature_count", 0)),
        "split_leakage_count": int(leak_audit.get("split_predictive_feature_count", 0)),
        "unsupported_transition_count": int(leak_audit.get("unsupported_transition_count", 0)),
        "action_mask_violation_count": int(leak_audit.get("action_mask_violation_count", 0)),
        "invalid_arg_count": int(leak_audit.get("invalid_arg_count", 0)),
        "validator": leak_audit,
        "canary_injection": canary_probe,
    }
    return {
        "transition_count": len(transitions),
        "shortcut_audit": shortcut_audit,
        "ablation_audit": ablation_audit,
        "holdout_audit": holdout_audit,
        "leakage_audit": leakage_audit,
        "manifest_integrity": {
            "manifest_rebuild_match": bool(manifest_integrity.get("manifest_rebuild_match", False)),
            "content_hash_rebuild_match": bool(manifest_integrity.get("content_hash_rebuild_match", False)),
            "split_hash_rebuild_match": bool(manifest_integrity.get("split_hash_rebuild_match", False)),
            "source_hash_rebuild_match": bool(manifest_integrity.get("source_hash_rebuild_match", False)),
            "dataset_hash": manifest_integrity.get("dataset_hash", manifest.dataset_hash),
            "schema_hash": manifest_integrity.get("schema_hash", ""),
            "split_hashes": dict(manifest_integrity.get("split_hashes") or {}),
        },
        "safety": {
            "no_training_started": manifest.no_training_started,
            "no_runtime_authority_added": manifest.no_runtime_authority_added,
        },
    }


def audit_controller_v2_lock_shortcuts(transitions: list[ControllerV2Transition]) -> dict[str, Any]:
    candidate_ids: list[str] = []
    candidate_positions: list[str] = []
    candidate_selected: list[str] = []
    target_positions: list[str] = []
    authority_levels: list[str] = []
    task_families: list[str] = []
    previous_actions: list[str] = []
    previous_action_available = False
    task_family_available_to_model = False
    for transition in transitions:
        labels = transition.labels
        metadata = transition.metadata
        model_input_text = canonical_json(transition.model_input)
        target_id = str(labels.get("target_candidate_id", ""))
        target_positions.append(str(labels.get("target_position", "")))
        authority_levels.append(str(labels.get("authority_level", "")))
        family = str(dict(metadata.get("evaluator_metadata") or {}).get("fixture_family", ""))
        task_families.append(family)
        if "previous_action" in model_input_text or "previous action" in model_input_text.lower():
            previous_action_available = True
        previous_actions.append(_previous_action_hint(transition))
        if family and family in model_input_text:
            task_family_available_to_model = True
        for candidate in list(transition.model_input.get("candidate_list") or []):
            candidate_payload = dict(candidate)
            candidate_id = str(candidate_payload.get("candidate_id", ""))
            candidate_ids.append(candidate_id)
            candidate_positions.append(str(candidate_payload.get("position", "")))
            candidate_selected.append("selected" if target_id != "ABSTAIN" and candidate_id == target_id else "not_selected")
    candidate_id_accuracy = _dominant_conditional_rate(candidate_ids, candidate_selected)
    candidate_position_accuracy = _dominant_conditional_rate(candidate_positions, candidate_selected)
    previous_action_accuracy = _dominant_conditional_rate(previous_actions, target_positions) if previous_action_available else 0.0
    authority_sequence_accuracy = _dominant_conditional_rate(authority_levels, target_positions)
    task_family_accuracy = _dominant_conditional_rate(task_families, target_positions)
    contrast_bigram = audit_controller_v2_contrast_bigrams(transitions)
    return {
        "candidate_id_shortcut_accuracy": candidate_id_accuracy,
        "candidate_id_shortcut_risk": _controller_v2_lock_risk(candidate_id_accuracy),
        "candidate_position_shortcut_accuracy": candidate_position_accuracy,
        "candidate_position_shortcut_risk": _controller_v2_lock_risk(candidate_position_accuracy),
        "previous_action_available_to_model": previous_action_available,
        "previous_action_shortcut_accuracy": previous_action_accuracy,
        "previous_action_shortcut_risk": _controller_v2_lock_risk(previous_action_accuracy),
        "authority_sequence_shortcut_accuracy": authority_sequence_accuracy,
        "authority_sequence_shortcut_risk": _controller_v2_lock_risk(authority_sequence_accuracy),
        "task_family_available_to_model": task_family_available_to_model,
        "task_family_shortcut_accuracy": task_family_accuracy,
        "task_family_shortcut_risk": _controller_v2_lock_risk(task_family_accuracy),
        "contrast_bigram_accuracy": contrast_bigram["contrast_bigram_accuracy"],
        "contrast_bigram_pair_count": contrast_bigram["contrast_bigram_pair_count"],
        "contrast_bigram_max_accuracy": CONTROLLER_V2_LOCK_CONTRAST_BIGRAM_MAX_ACCURACY,
        "high_risk_threshold": CONTROLLER_V2_LOCK_HIGH_RISK_THRESHOLD,
        "medium_risk_threshold": CONTROLLER_V2_LOCK_MEDIUM_RISK_THRESHOLD,
    }


def audit_controller_v2_contrast_bigrams(transitions: list[ControllerV2Transition]) -> dict[str, Any]:
    grouped: dict[str, list[ControllerV2Transition]] = {}
    for transition in transitions:
        pair_id = str(transition.metadata.get("contrast_pair_id") or "")
        if pair_id:
            grouped.setdefault(pair_id, []).append(transition)
    pair_count = 0
    correct = 0
    for pair in grouped.values():
        if len(pair) != 2:
            continue
        ordered = sorted(pair, key=_transition_fixture_index)
        pair_count += 1
        if str(ordered[0].labels.get("target_position", "")) == str(ordered[1].labels.get("target_position", "")):
            correct += 1
    accuracy = _rate_float(correct, pair_count)
    return {
        "contrast_bigram_accuracy": accuracy,
        "contrast_bigram_pair_count": pair_count,
        "contrast_bigram_correct_count": correct,
    }


def audit_controller_v2_input_ablations(transitions: list[ControllerV2Transition]) -> dict[str, Any]:
    channel_specs = {
        "memory": ("safe_memory_cards", _memory_case),
        "workspace": ("workspace_summary", _workspace_case),
        "verifier": ("verifier_context_summary", _verifier_case),
        "skill_status": ("safe_skill_cards", _skill_status_case),
    }
    channel_results: dict[str, dict[str, Any]] = {}
    for channel, (input_key, case_fn) in channel_specs.items():
        case_count = 0
        changed_count = 0
        for transition in transitions:
            if not case_fn(transition):
                continue
            case_count += 1
            baseline = _controller_v2_visible_channel_prediction(transition.model_input)
            ablated_input = deepcopy(transition.model_input)
            ablated_input[input_key] = [] if input_key in {"safe_memory_cards", "safe_skill_cards"} else ""
            ablated = _controller_v2_visible_channel_prediction(ablated_input)
            if baseline != ablated:
                changed_count += 1
        channel_results[channel] = {
            "case_count": case_count,
            "changed_decision_count": changed_count,
            "changes_decisions": changed_count > 0,
        }
    memory_changed = bool(channel_results["memory"]["changes_decisions"])
    workspace_changed = bool(channel_results["workspace"]["changes_decisions"])
    return {
        "channels": channel_results,
        "memory_ablation_changes_memory_cases": memory_changed,
        "workspace_ablation_changes_workspace_cases": workspace_changed,
        "memory_or_workspace_ablation_changes_decisions": memory_changed or workspace_changed,
        "verifier_ablation_changes_verifier_cases": bool(channel_results["verifier"]["changes_decisions"]),
        "skill_status_ablation_changes_skill_cases": bool(channel_results["skill_status"]["changes_decisions"]),
    }


def audit_controller_v2_holdouts(transitions: list[ControllerV2Transition]) -> dict[str, Any]:
    family_holdout_ids: set[str] = set()
    project_holdout_ids: set[str] = set()
    family_holdout_leak_count = 0
    project_holdout_leak_count = 0
    for transition in transitions:
        text = canonical_json(transition.model_input)
        family_id = str(transition.metadata.get("family_holdout_id") or "")
        project_id = str(transition.metadata.get("project_holdout_id") or "")
        if _split(transition) == "family_holdout":
            family_holdout_ids.add(family_id)
        if _split(transition) == "project_holdout":
            project_holdout_ids.add(project_id)
        if family_id and family_id in text:
            family_holdout_leak_count += 1
        if project_id and project_id in text:
            project_holdout_leak_count += 1
    return {
        "family_holdout_count": sum(1 for transition in transitions if _split(transition) == "family_holdout"),
        "project_holdout_count": sum(1 for transition in transitions if _split(transition) == "project_holdout"),
        "family_holdout_unique_count": len(family_holdout_ids),
        "project_holdout_unique_count": len(project_holdout_ids),
        "family_holdout_leak_count": family_holdout_leak_count,
        "project_holdout_leak_count": project_holdout_leak_count,
        "family_project_holdout_covered": bool(family_holdout_ids) and bool(project_holdout_ids),
    }


def audit_controller_v2_canary_injection(
    transitions: list[ControllerV2Transition],
    source_hashes: dict[str, str],
) -> dict[str, Any]:
    if not transitions:
        return {
            "canary_injection_rejected": False,
            "invalid_transition_count": 0,
            "oracle_leak_count": 0,
            "canary_leak_count": 0,
            "trace_template_leakage_count": 0,
            "split_leakage_count": 0,
        }
    leaked = ControllerV2Transition.from_dict(transitions[0].to_dict())
    leaked.model_input["canary_token"] = "CANARY_hidden_project_seed"
    leaked.model_input["source_trace_id"] = leaked.metadata["source_trace_id"]
    leaked.model_input["split"] = leaked.metadata["split"]
    report = validate_controller_v2_transitions([leaked], source_hashes)
    return {
        "canary_injection_rejected": int(report.get("invalid_transition_count", 0)) > 0,
        "invalid_transition_count": int(report.get("invalid_transition_count", 0)),
        "oracle_leak_count": int(report.get("oracle_leak_count", 0)),
        "canary_leak_count": int(report.get("canary_leak_count", 0)),
        "trace_template_leakage_count": int(report.get("trace_predictive_feature_count", 0)),
        "split_leakage_count": int(report.get("split_predictive_feature_count", 0)),
        "rejection_report": report,
    }


def _build_transitions(
    specs: list[ControllerV2ClassSpec],
    source_hashes: dict[str, str],
    transitions_per_class: int,
) -> list[ControllerV2Transition]:
    transitions: list[ControllerV2Transition] = []
    for class_index, spec in enumerate(specs):
        for index in range(transitions_per_class):
            transitions.append(_make_transition(specs, spec, source_hashes, class_index, index))
    return sorted(transitions, key=lambda transition: transition.transition_id)


def _make_transition(
    specs: list[ControllerV2ClassSpec],
    spec: ControllerV2ClassSpec,
    source_hashes: dict[str, str],
    class_index: int,
    index: int,
) -> ControllerV2Transition:
    split = _assign_split(index)
    family_holdout_id = f"family_holdout_{class_index % 5}"
    project_holdout_id = f"project_holdout_{(class_index + index) % 7}"
    source_trace_id = _source_trace_id(spec.gate_source, spec.candidate_class, index)
    contrast_pair_id = f"pair_{spec.candidate_class}_{index // 2:03d}" if index < 240 else ""
    target_position = "ABSTAIN" if spec.target_kind == "abstain" else str((index + class_index) % 4)
    candidates = _candidate_list(specs, spec, class_index, index, target_position)
    candidate_ids = [str(candidate["candidate_id"]) for candidate in candidates]
    target_candidate_id = "ABSTAIN" if target_position == "ABSTAIN" else candidate_ids[int(target_position)]
    metadata = {
        "schema_version": CONTROLLER_V2_SCHEMA_VERSION,
        "source_trace_id": source_trace_id,
        "source_trace_hash": source_hashes[source_trace_id],
        "gate_source": spec.gate_source,
        "split": split,
        "candidate_class": spec.candidate_class,
        "family_holdout_id": family_holdout_id,
        "project_holdout_id": project_holdout_id,
        "contrast_pair_id": contrast_pair_id,
        "transition_id": "",
        "evaluator_metadata": {
            "fixture_family": spec.family,
            "fixture_index": index,
            "coverage_case": spec.candidate_class,
        },
    }
    model_input = {
        "agent_visible_input": {
            "task_summary": f"choose safe controller candidate for {spec.authority_level} operation",
            "observed_status": _observed_status(spec),
            "recent_event_summary": f"verifier={spec.verifier_outcome}; rollback={spec.rollback_outcome}",
        },
        "workspace_summary": f"local_state=clean; candidate_count={len(candidates)}; authority={spec.authority_level}",
        "safe_memory_cards": [
            {
                "memory_id": f"mem_{stable_hash({'class': spec.candidate_class})[:12]}",
                "claim": f"{spec.authority_level} requires deterministic approval before execution",
                "evidence_count": 3,
            }
        ],
        "safe_skill_cards": [
            {
                "skill_id": f"skill_{stable_hash({'family': spec.family})[:12]}",
                "status": "available" if spec.safety_outcome not in {"quarantined", "deprecated"} else spec.safety_outcome,
                "verifier": "pytest",
            }
        ],
        "candidate_list": candidates,
        "candidate_ids": candidate_ids,
        "candidate_action_mask": {candidate_id: True for candidate_id in candidate_ids},
        "action_mask": {op.value: True for op in CognitiveOp},
        "authority_context": spec.authority_level,
        "risk_context": spec.risk_class,
        "verifier_context_summary": f"outcome={spec.verifier_outcome}",
        "rollback_context_summary": f"outcome={spec.rollback_outcome}",
    }
    labels = {
        "target_candidate_id": target_candidate_id,
        "target_candidate_args": {} if target_candidate_id == "ABSTAIN" else {"candidate_id": target_candidate_id},
        "target_action": spec.target_action,
        "target_decision": "ABSTAIN" if target_candidate_id == "ABSTAIN" else "SELECT_CANDIDATE",
        "target_position": target_position,
        "candidate_class": spec.candidate_class,
        "risk_class": spec.risk_class,
        "authority_level": spec.authority_level,
        "verifier_outcome": spec.verifier_outcome,
        "rollback_outcome": spec.rollback_outcome,
        "mutation_class": spec.mutation_class,
        "safety_outcome": spec.safety_outcome,
        "mutating": spec.mutation_class != "none",
    }
    transition = ControllerV2Transition(metadata=metadata, model_input=model_input, labels=labels)
    transition.metadata["transition_id"] = "ctrlv2_" + transition.stable_hash()[:24]
    return transition


def _candidate_list(
    specs: list[ControllerV2ClassSpec],
    spec: ControllerV2ClassSpec,
    class_index: int,
    index: int,
    target_position: str,
) -> list[dict[str, Any]]:
    candidates: list[dict[str, Any]] = []
    for position in range(4):
        decoy = specs[(class_index + position + 1) % len(specs)]
        item_spec = spec if target_position != "ABSTAIN" and position == int(target_position) else decoy
        payload = {
            "candidate_class": item_spec.candidate_class,
            "risk_class": item_spec.risk_class,
            "authority_level": item_spec.authority_level,
            "mutation_class": item_spec.mutation_class,
            "verifier_required": item_spec.verifier_outcome != "not_required",
            "rollback_available": item_spec.rollback_outcome not in {"not_applicable", ""},
            "visible_evidence_count": 3 + ((index + position) % 5),
            "position": position,
            "gate_hint": item_spec.gate_source,
        }
        payload["candidate_id"] = "cand_" + stable_hash({key: value for key, value in payload.items() if key != "position"})[:20]
        candidates.append(payload)
    return candidates


def _source_artifact_hashes(specs: list[ControllerV2ClassSpec]) -> dict[str, str]:
    hashes: dict[str, str] = {}
    for spec in specs:
        for index in range(280):
            source_id = _source_trace_id(spec.gate_source, spec.candidate_class, index)
            hashes[source_id] = stable_hash(
                {
                    "gate_source": spec.gate_source,
                    "candidate_class": spec.candidate_class,
                    "fixture_index": index,
                    "authority_level": spec.authority_level,
                    "safety_outcome": spec.safety_outcome,
                }
            )
    return hashes


def _source_trace_id(gate_source: str, candidate_class: str, index: int) -> str:
    return f"src_{gate_source}_{candidate_class}_{index:03d}"


def _assign_split(index: int) -> str:
    cycle = (
        "train",
        "train",
        "train",
        "train",
        "validation",
        "test",
        "adversarial",
        "negative",
        "ood",
        "family_holdout",
        "project_holdout",
        "train",
        "validation",
        "test",
    )
    return cycle[index % len(cycle)]


def _observed_status(spec: ControllerV2ClassSpec) -> str:
    if spec.safety_outcome == "approved":
        return "candidate evidence satisfies deterministic gate"
    if spec.safety_outcome == "rollback_required":
        return "post-verifier failure requires rollback"
    if spec.safety_outcome == "fallback":
        return "primary candidate failed and fallback is safer"
    if spec.safety_outcome == "quarantined":
        return "skill registry marks candidate quarantined"
    if spec.safety_outcome == "deprecated":
        return "skill registry marks candidate deprecated"
    if spec.safety_outcome == "abstained":
        return "candidate distribution is out of domain"
    return "candidate evidence rejected by deterministic approval"


def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
    path.write_text("".join(canonical_json(row) + "\n" for row in rows), encoding="utf-8")


def _write_manifest(manifest: ControllerV2DatasetManifest) -> None:
    Path(manifest.manifest_path).write_text(json.dumps(manifest.to_dict(), indent=2, sort_keys=True), encoding="utf-8")


def _read_transitions(path: Path) -> list[ControllerV2Transition]:
    return [
        ControllerV2Transition.from_dict(json.loads(line))
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def _split(transition: ControllerV2Transition) -> str:
    return str(transition.metadata["split"])


def _count_by(
    transitions: list[ControllerV2Transition],
    key_fn: object,
    expected_keys: tuple[str, ...] = (),
) -> dict[str, int]:
    counts = {key: 0 for key in expected_keys}
    for transition in transitions:
        key = str(key_fn(transition))  # type: ignore[misc]
        counts[key] = counts.get(key, 0) + 1
    return dict(sorted(counts.items()))


def _contrast_pair_count(transitions: list[ControllerV2Transition]) -> int:
    pairs: dict[str, int] = {}
    for transition in transitions:
        pair_id = str(transition.metadata.get("contrast_pair_id") or "")
        if pair_id:
            pairs[pair_id] = pairs.get(pair_id, 0) + 1
    return sum(1 for count in pairs.values() if count == 2)


def _split_hashes(transitions: list[ControllerV2Transition]) -> dict[str, str]:
    return {
        split: stable_hash([transition.transition_id for transition in transitions if _split(transition) == split])
        for split in CONTROLLER_V2_SPLITS
    }


def _manifest_comparable(manifest: ControllerV2DatasetManifest) -> dict[str, Any]:
    payload = manifest.to_dict()
    for key in ["output_dir", "dataset_path", "manifest_path", "split_paths"]:
        payload.pop(key, None)
    integrity = dict(payload.get("manifest_integrity") or {})
    for key in [
        "manifest_rebuild_match",
        "content_hash_rebuild_match",
        "transition_id_match",
        "transition_hash_match",
        "split_hash_rebuild_match",
        "source_hash_rebuild_match",
        "rebuilt_manifest_path",
        "rebuilt_dataset_path",
        "rebuilt_transition_count",
    ]:
        integrity.pop(key, None)
    payload["manifest_integrity"] = integrity
    return payload


def _find_forbidden_keys(value: Any, prefix: str = "") -> list[str]:
    found: list[str] = []
    if isinstance(value, dict):
        for key, child in value.items():
            key_text = str(key)
            path = f"{prefix}.{key_text}" if prefix else key_text
            if key_text in FORBIDDEN_INPUT_KEYS:
                found.append(path)
            found.extend(_find_forbidden_keys(child, path))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            found.extend(_find_forbidden_keys(child, f"{prefix}[{index}]"))
    return found


def _json_safe(value: object) -> bool:
    try:
        canonical_json(value)
    except (TypeError, ValueError):
        return False
    return True


def _controller_v2_lock_risk(accuracy: float) -> str:
    if accuracy >= CONTROLLER_V2_LOCK_HIGH_RISK_THRESHOLD:
        return "high"
    if accuracy >= CONTROLLER_V2_LOCK_MEDIUM_RISK_THRESHOLD:
        return "medium"
    return "low"


def _previous_action_hint(transition: ControllerV2Transition) -> str:
    text = canonical_json(transition.model_input).lower()
    for prefix in ("previous_action=", "previous action="):
        if prefix in text:
            return text.split(prefix, 1)[1].split(";", 1)[0].split('"', 1)[0]
    return "not_available"


def _transition_fixture_index(transition: ControllerV2Transition) -> int:
    metadata = dict(transition.metadata.get("evaluator_metadata") or {})
    try:
        return int(metadata.get("fixture_index", 0))
    except (TypeError, ValueError):
        return 0


def _rate_float(numerator: int, denominator: int) -> float:
    return numerator / denominator if denominator else 0.0


def _memory_case(transition: ControllerV2Transition) -> bool:
    candidate_class = str(transition.metadata.get("candidate_class") or "")
    return "memory" in candidate_class


def _workspace_case(transition: ControllerV2Transition) -> bool:
    candidate_class = str(transition.metadata.get("candidate_class") or "")
    return any(token in candidate_class for token in ("worktree", "patch", "human_approved"))


def _verifier_case(transition: ControllerV2Transition) -> bool:
    return str(transition.labels.get("verifier_outcome") or "") != "not_required"


def _skill_status_case(transition: ControllerV2Transition) -> bool:
    candidate_class = str(transition.metadata.get("candidate_class") or "")
    return any(token in candidate_class for token in ("skill", "quarantine", "deprecation"))


def _controller_v2_visible_channel_prediction(model_input: dict[str, Any]) -> str:
    agent_visible = dict(model_input.get("agent_visible_input") or {})
    observed_status = str(agent_visible.get("observed_status") or "").lower()
    if "out of domain" in observed_status or "rejected" in observed_status:
        return "ABSTAIN"
    task_summary = str(agent_visible.get("task_summary") or "").lower()
    recent_event = str(agent_visible.get("recent_event_summary") or "").lower()
    workspace_summary = str(model_input.get("workspace_summary") or "")
    authority_context = str(model_input.get("authority_context") or "")
    risk_context = str(model_input.get("risk_context") or "")
    verifier_summary = str(model_input.get("verifier_context_summary") or "")
    rollback_summary = str(model_input.get("rollback_context_summary") or "")
    safe_memory_cards = list(model_input.get("safe_memory_cards") or [])
    safe_skill_cards = list(model_input.get("safe_skill_cards") or [])
    workspace_enabled = bool(workspace_summary)
    memory_enabled = bool(safe_memory_cards)
    skill_enabled = bool(safe_skill_cards)
    verifier_enabled = bool(verifier_summary)
    desired_authority = _authority_from_workspace(workspace_summary) if workspace_enabled else authority_context
    desired_verifier_required = None
    if verifier_enabled:
        desired_verifier_required = "not_required" not in verifier_summary
    desired_rollback_available = None
    if rollback_summary:
        desired_rollback_available = "not_applicable" not in rollback_summary
    skill_status = ""
    if safe_skill_cards:
        skill_status = str(dict(safe_skill_cards[0]).get("status") or "")
    best_id = "ABSTAIN"
    best_score = -10_000
    for raw_candidate in list(model_input.get("candidate_list") or []):
        candidate = dict(raw_candidate)
        candidate_id = str(candidate.get("candidate_id") or "")
        candidate_class = str(candidate.get("candidate_class") or "")
        score = 0
        if desired_authority and str(candidate.get("authority_level") or "") == desired_authority:
            score += 20
        if risk_context and str(candidate.get("risk_class") or "") == risk_context:
            score += 10
        if desired_verifier_required is not None:
            score += 8 if bool(candidate.get("verifier_required")) == desired_verifier_required else -8
        elif bool(candidate.get("verifier_required")):
            score -= 18
        if desired_rollback_available is not None:
            score += 4 if bool(candidate.get("rollback_available")) == desired_rollback_available else -4
        if "memory" in candidate_class:
            score += 16 if memory_enabled else -16
        if any(token in candidate_class for token in ("worktree", "patch", "human_approved")):
            score += 14 if workspace_enabled else -14
        if any(token in candidate_class for token in ("skill", "quarantine", "deprecation")):
            score += 14 if skill_enabled else -14
        if "quarantined" in observed_status:
            score += 35 if candidate_class == "quarantine_case" and skill_status == "quarantined" else -4
        elif "deprecated" in observed_status:
            score += 35 if candidate_class == "deprecation_case" and skill_status == "deprecated" else -4
        elif "rollback" in observed_status:
            score += 35 if candidate_class == "rollback_case" else -4
        elif "fallback" in observed_status:
            score += 35 if candidate_class == "fallback_case" else -4
        elif "memory" in task_summary and "memory" in candidate_class:
            score += 10
        elif "verifier" in task_summary and "verifier" in candidate_class:
            score += 10
        elif "skill" in task_summary and "skill" in candidate_class:
            score += 10
        if "verifier=failed" in recent_event and bool(candidate.get("verifier_required")):
            score += 3
        if score > best_score or (score == best_score and candidate_id < best_id):
            best_score = score
            best_id = candidate_id
    return best_id


def _authority_from_workspace(workspace_summary: str) -> str:
    marker = "authority="
    if marker not in workspace_summary:
        return ""
    return workspace_summary.split(marker, 1)[1].split(";", 1)[0].strip()


def _dominant_rate(values: list[str]) -> float:
    if not values:
        return 0.0
    counts: dict[str, int] = {}
    for value in values:
        counts[value] = counts.get(value, 0) + 1
    return max(counts.values()) / len(values)


def _dominant_conditional_rate(keys: list[str], targets: list[str]) -> float:
    if not keys or len(keys) != len(targets):
        return 0.0
    grouped: dict[str, dict[str, int]] = {}
    for key, target in zip(keys, targets):
        grouped.setdefault(key, {})[target] = grouped.setdefault(key, {}).get(target, 0) + 1
    correct = sum(max(counts.values()) for counts in grouped.values())
    return correct / len(targets)
