"""Gate 7D deterministic approval for skill graph updates.

Gate 7D is the first gate that may persist learned-recommended skill graph
updates, but the learned output still supplies only a candidate ID. The
deterministic approval layer revalidates the Gate 7C candidate, verifies trace
evidence, requires held-out and adversarial validation, writes only under the
requested state directory, records rollback artifacts, and can restore the
previous registry snapshot.
"""

from __future__ import annotations

from dataclasses import dataclass
import json
from pathlib import Path
from typing import Any

from .bytecode import CognitiveOp
from .ledger import TraceLedgerRecord, canonical_json, stable_hash
from .skill_update_recommendation import (
    DETERMINISTIC_SKILL_UPDATE_GENERATOR,
    LearnedSkillUpdateRecommendation,
    SkillUpdateCandidate,
    validate_skill_update_candidate,
)
from .skills import Skill, SkillGraph


GATE7D_SCHEMA_VERSION = 1
DETERMINISTIC_SKILL_UPDATE_APPROVER = "deterministic_skill_update_approver_v1"
ALLOWED_SKILL_UPDATE_VERIFIERS = {"pytest", "internal_verify", "manual_or_regression_suite"}
FORBIDDEN_SKILL_UPDATE_TEXT = (
    "always pass",
    "bypass verifier",
    "disable verifier",
    "hidden oracle",
    "ignore failures",
    "skip validation",
    "turn off safety",
)
FORBIDDEN_SKILL_UPDATE_METRIC_KEYS = {
    "adversarial_success_rate",
    "average_saved_steps",
    "failure_count",
    "heldout_success_rate",
    "precision",
    "promotion_evidence_count",
    "regression_status",
    "success_count",
    "trigger_attempt_count",
    "true_positive_trigger_count",
    "verifier_fail_count",
    "verifier_pass_count",
}
FALSE_TRIGGER_METRIC_KEYS = {"false_trigger_count", "false_positive_trigger_count"}
UNOBSERVED_PATH_KEYS = {"arg_path", "cwd", "directory", "file", "folder", "path", "workdir"}
UNOBSERVED_PATH_MARKERS = ("../", "/etc/", "/home/", "/tmp/", "c:\\", "\\\\", "file://", "~/.ssh")


@dataclass(slots=True)
class SkillUpdateApprovalDecision:
    approval_id: str
    candidate_id: str
    approved: bool
    reason_code: str
    evidence_verified: bool
    validation_passed: bool
    regression_suite_passed: bool
    reversible: bool = True
    deterministic_approver: str = DETERMINISTIC_SKILL_UPDATE_APPROVER

    def to_dict(self) -> dict[str, Any]:
        return {
            "approval_id": self.approval_id,
            "candidate_id": self.candidate_id,
            "approved": self.approved,
            "reason_code": self.reason_code,
            "evidence_verified": self.evidence_verified,
            "validation_passed": self.validation_passed,
            "regression_suite_passed": self.regression_suite_passed,
            "reversible": self.reversible,
            "deterministic_approver": self.deterministic_approver,
            "schema_version": GATE7D_SCHEMA_VERSION,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "SkillUpdateApprovalDecision":
        return cls(
            approval_id=str(data["approval_id"]),
            candidate_id=str(data.get("candidate_id", "")),
            approved=bool(data["approved"]),
            reason_code=str(data.get("reason_code", "")),
            evidence_verified=bool(data.get("evidence_verified", False)),
            validation_passed=bool(data.get("validation_passed", False)),
            regression_suite_passed=bool(data.get("regression_suite_passed", False)),
            reversible=bool(data.get("reversible", True)),
            deterministic_approver=str(data.get("deterministic_approver", DETERMINISTIC_SKILL_UPDATE_APPROVER)),
        )


@dataclass(slots=True)
class SkillUpdateExecutionResult:
    update_id: str
    candidate_id: str
    target_skill_id: str
    target_skill_name: str
    update_kind: str
    approved: bool
    written: bool
    rolled_back: bool
    rollback_successful: bool
    reason_code: str
    evidence_verified: bool
    validation_passed: bool
    regression_suite_passed: bool
    post_update_regression_rate: float
    post_update_regression_suite_run: bool
    post_update_regression_suite_passed: bool
    skill_reuse_improvement_detected: bool
    registry_path: str
    history_path: str
    rollback_artifact_path: str
    before_registry_hash: str
    after_registry_hash: str
    final_registry_hash: str
    approval: SkillUpdateApprovalDecision

    def to_dict(self) -> dict[str, Any]:
        return {
            "update_id": self.update_id,
            "candidate_id": self.candidate_id,
            "target_skill_id": self.target_skill_id,
            "target_skill_name": self.target_skill_name,
            "update_kind": self.update_kind,
            "approved": self.approved,
            "written": self.written,
            "rolled_back": self.rolled_back,
            "rollback_successful": self.rollback_successful,
            "reason_code": self.reason_code,
            "evidence_verified": self.evidence_verified,
            "validation_passed": self.validation_passed,
            "regression_suite_passed": self.regression_suite_passed,
            "post_update_regression_rate": self.post_update_regression_rate,
            "post_update_regression_suite_run": self.post_update_regression_suite_run,
            "post_update_regression_suite_passed": self.post_update_regression_suite_passed,
            "skill_reuse_improvement_detected": self.skill_reuse_improvement_detected,
            "registry_path": self.registry_path,
            "history_path": self.history_path,
            "rollback_artifact_path": self.rollback_artifact_path,
            "before_registry_hash": self.before_registry_hash,
            "after_registry_hash": self.after_registry_hash,
            "final_registry_hash": self.final_registry_hash,
            "approval": self.approval.to_dict(),
            "schema_version": GATE7D_SCHEMA_VERSION,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "SkillUpdateExecutionResult":
        return cls(
            update_id=str(data["update_id"]),
            candidate_id=str(data.get("candidate_id", "")),
            target_skill_id=str(data.get("target_skill_id", "")),
            target_skill_name=str(data.get("target_skill_name", "")),
            update_kind=str(data.get("update_kind", "")),
            approved=bool(data["approved"]),
            written=bool(data["written"]),
            rolled_back=bool(data.get("rolled_back", False)),
            rollback_successful=bool(data.get("rollback_successful", False)),
            reason_code=str(data.get("reason_code", "")),
            evidence_verified=bool(data.get("evidence_verified", False)),
            validation_passed=bool(data.get("validation_passed", False)),
            regression_suite_passed=bool(data.get("regression_suite_passed", False)),
            post_update_regression_rate=float(data.get("post_update_regression_rate", 1.0)),
            post_update_regression_suite_run=bool(data.get("post_update_regression_suite_run", False)),
            post_update_regression_suite_passed=bool(data.get("post_update_regression_suite_passed", False)),
            skill_reuse_improvement_detected=bool(data.get("skill_reuse_improvement_detected", False)),
            registry_path=str(data.get("registry_path", "")),
            history_path=str(data.get("history_path", "")),
            rollback_artifact_path=str(data.get("rollback_artifact_path", "")),
            before_registry_hash=str(data.get("before_registry_hash", "")),
            after_registry_hash=str(data.get("after_registry_hash", "")),
            final_registry_hash=str(data.get("final_registry_hash", "")),
            approval=SkillUpdateApprovalDecision.from_dict(dict(data["approval"])),
        )


@dataclass(slots=True)
class SkillUpdateImpactAudit:
    approved_skill_update_count: int
    rejected_bad_update_count: int
    bad_update_reject_rate: float
    rollback_count: int
    rollback_success_rate: float
    post_update_regression_rate: float
    post_update_regression_suite_run_count: int
    post_update_regression_failure_count: int
    skill_reuse_improvement_detected: bool
    deterministic_skill_graph_write_count: int
    skill_store_update_write_count: int
    rollback_restore_write_count: int
    skill_store_write_count: int
    history_log_append_count: int
    rollback_artifact_count: int
    learned_skill_graph_writes_executed: int
    write_path_containment_passed: bool
    writes_outside_state_dir_count: int
    evidence_hash_mismatch_rejected_count: int
    action_mask_rejected_count: int
    heldout_validation_pass_count: int
    adversarial_validation_pass_count: int

    def to_dict(self) -> dict[str, Any]:
        return {
            "approved_skill_update_count": self.approved_skill_update_count,
            "rejected_bad_update_count": self.rejected_bad_update_count,
            "bad_update_reject_rate": self.bad_update_reject_rate,
            "rollback_count": self.rollback_count,
            "rollback_success_rate": self.rollback_success_rate,
            "post_update_regression_rate": self.post_update_regression_rate,
            "post_update_regression_suite_run_count": self.post_update_regression_suite_run_count,
            "post_update_regression_failure_count": self.post_update_regression_failure_count,
            "skill_reuse_improvement_detected": self.skill_reuse_improvement_detected,
            "deterministic_skill_graph_write_count": self.deterministic_skill_graph_write_count,
            "skill_store_update_write_count": self.skill_store_update_write_count,
            "rollback_restore_write_count": self.rollback_restore_write_count,
            "skill_store_write_count": self.skill_store_write_count,
            "history_log_append_count": self.history_log_append_count,
            "rollback_artifact_count": self.rollback_artifact_count,
            "learned_skill_graph_writes_executed": self.learned_skill_graph_writes_executed,
            "write_path_containment_passed": self.write_path_containment_passed,
            "writes_outside_state_dir_count": self.writes_outside_state_dir_count,
            "evidence_hash_mismatch_rejected_count": self.evidence_hash_mismatch_rejected_count,
            "action_mask_rejected_count": self.action_mask_rejected_count,
            "heldout_validation_pass_count": self.heldout_validation_pass_count,
            "adversarial_validation_pass_count": self.adversarial_validation_pass_count,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "SkillUpdateImpactAudit":
        return cls(
            approved_skill_update_count=int(data["approved_skill_update_count"]),
            rejected_bad_update_count=int(data["rejected_bad_update_count"]),
            bad_update_reject_rate=float(data["bad_update_reject_rate"]),
            rollback_count=int(data["rollback_count"]),
            rollback_success_rate=float(data["rollback_success_rate"]),
            post_update_regression_rate=float(data["post_update_regression_rate"]),
            post_update_regression_suite_run_count=int(data["post_update_regression_suite_run_count"]),
            post_update_regression_failure_count=int(data["post_update_regression_failure_count"]),
            skill_reuse_improvement_detected=bool(data["skill_reuse_improvement_detected"]),
            deterministic_skill_graph_write_count=int(data["deterministic_skill_graph_write_count"]),
            skill_store_update_write_count=int(data["skill_store_update_write_count"]),
            rollback_restore_write_count=int(data["rollback_restore_write_count"]),
            skill_store_write_count=int(data["skill_store_write_count"]),
            history_log_append_count=int(data["history_log_append_count"]),
            rollback_artifact_count=int(data["rollback_artifact_count"]),
            learned_skill_graph_writes_executed=int(data["learned_skill_graph_writes_executed"]),
            write_path_containment_passed=bool(data["write_path_containment_passed"]),
            writes_outside_state_dir_count=int(data["writes_outside_state_dir_count"]),
            evidence_hash_mismatch_rejected_count=int(data["evidence_hash_mismatch_rejected_count"]),
            action_mask_rejected_count=int(data["action_mask_rejected_count"]),
            heldout_validation_pass_count=int(data["heldout_validation_pass_count"]),
            adversarial_validation_pass_count=int(data["adversarial_validation_pass_count"]),
        )


class DeterministicSkillUpdateApprover:
    def approve(
        self,
        candidate: SkillUpdateCandidate,
        trace_index: dict[str, TraceLedgerRecord],
        registry: "VersionedSkillRegistry",
        action_mask: dict[str, bool] | None = None,
        *,
        post_update_regression_rate: float = 0.0,
    ) -> SkillUpdateApprovalDecision:
        valid, reason = validate_skill_update_candidate(candidate)
        evidence_verified, evidence_reason = verify_skill_update_candidate_evidence(candidate, trace_index)
        validation_passed, validation_reason = verify_skill_update_generalization(candidate)
        payload_safe, payload_safety_reason = verify_skill_update_candidate_payload_safety(candidate)
        target_verified, target_reason = registry.verify_candidate_target(candidate)
        mask_allowed = bool((action_mask or {CognitiveOp.UPDATE_SKILL.value: True}).get(CognitiveOp.UPDATE_SKILL.value, False))
        regression_suite_passed = post_update_regression_rate == 0.0
        approved = all([valid, evidence_verified, validation_passed, payload_safe, target_verified, mask_allowed, regression_suite_passed])
        if not valid:
            reason_code = reason
        elif not evidence_verified:
            reason_code = evidence_reason
        elif not validation_passed:
            reason_code = validation_reason
        elif not payload_safe:
            reason_code = payload_safety_reason
        elif not target_verified:
            reason_code = target_reason
        elif not mask_allowed:
            reason_code = "action_mask_violation"
        elif not regression_suite_passed:
            reason_code = "post_update_regression_detected"
        else:
            reason_code = "approved_for_deterministic_skill_graph_update"
        payload = {
            "candidate_id": candidate.candidate_id,
            "candidate_hash": stable_hash(candidate.to_dict()),
            "evidence_verified": evidence_verified,
            "validation_passed": validation_passed,
            "payload_safe": payload_safe,
            "target_verified": target_verified,
            "mask_allowed": mask_allowed,
            "post_update_regression_rate": post_update_regression_rate,
            "reason_code": reason_code,
            "approved": approved,
            "approver": DETERMINISTIC_SKILL_UPDATE_APPROVER,
        }
        return SkillUpdateApprovalDecision(
            approval_id="skillupd_approval_" + stable_hash(payload)[:20],
            candidate_id=candidate.candidate_id,
            approved=approved,
            reason_code=reason_code,
            evidence_verified=evidence_verified,
            validation_passed=validation_passed,
            regression_suite_passed=regression_suite_passed,
        )


class VersionedSkillRegistry:
    def __init__(self, registry_path: str | Path, *, approver: DeterministicSkillUpdateApprover | None = None) -> None:
        self.registry_path = Path(registry_path)
        self.registry_path.parent.mkdir(parents=True, exist_ok=True)
        self.history_path = self.registry_path.with_name(self.registry_path.stem + "_history.jsonl")
        self.rollback_dir = self.registry_path.with_name(self.registry_path.stem + "_rollback")
        self.rollback_dir.mkdir(parents=True, exist_ok=True)
        self.graph = SkillGraph(self.registry_path)
        self.approver = approver or DeterministicSkillUpdateApprover()
        self.deterministic_skill_graph_write_count = 0
        self.skill_store_update_write_count = 0
        self.rollback_restore_write_count = 0
        self.skill_store_write_count = 0
        self.history_log_append_count = 0
        self.rollback_artifact_count = 0
        self.learned_skill_graph_writes_executed = 0

    def seed_skills(self, skills: list[Skill]) -> None:
        self.graph.skills = {skill.name: Skill.from_dict(skill.to_dict()) for skill in skills}
        self.graph.save()

    def verify_candidate_target(self, candidate: SkillUpdateCandidate) -> tuple[bool, str]:
        existing = self.graph.get(candidate.target_skill_name)
        if candidate.update_kind == "create_skill":
            if existing is not None:
                return False, "target_skill_already_exists"
            target_id = str(candidate.target_skill_id)
            if any(str(skill.skill_id or skill.name) == target_id for skill in self.graph.skills.values()):
                return False, "duplicate_skill_id"
            return True, "ok"
        if existing is None:
            return False, "target_skill_not_found"
        if str(existing.skill_id or existing.name) != candidate.target_skill_id:
            return False, "target_skill_id_mismatch"
        if int(existing.version) != int(candidate.base_version):
            return False, "version_conflict"
        if existing.status in {"deprecated", "quarantined"} and candidate.update_kind not in {"deprecate_skill", "quarantine_skill"}:
            if existing.status == "deprecated":
                return False, "deprecated_skill_update_rejected"
            return False, "quarantined_skill_resurrection_rejected"
        return True, "ok"

    def execute_recommendation(
        self,
        *,
        update_id: str,
        recommendation: LearnedSkillUpdateRecommendation,
        candidates: list[SkillUpdateCandidate],
        trace_index: dict[str, TraceLedgerRecord],
        action_mask: dict[str, bool] | None = None,
        rollback_after_write: bool = False,
        rollback_reason: str = "",
        post_update_regression_rate: float = 0.0,
    ) -> SkillUpdateExecutionResult:
        if recommendation.decision != "accept":
            approval = _rejected_approval(recommendation.candidate_id, recommendation.reason_code)
            return self._result_without_write(update_id, None, approval, post_update_regression_rate)
        candidate = next((item for item in candidates if item.candidate_id == recommendation.candidate_id), None)
        if candidate is None:
            approval = _rejected_approval(recommendation.candidate_id, "candidate_id_not_found")
            return self._result_without_write(update_id, None, approval, post_update_regression_rate)
        return self.write_candidate(
            update_id=update_id,
            candidate=candidate,
            trace_index=trace_index,
            action_mask=action_mask,
            rollback_after_write=rollback_after_write,
            rollback_reason=rollback_reason,
            post_update_regression_rate=post_update_regression_rate,
        )

    def write_candidate(
        self,
        *,
        update_id: str,
        candidate: SkillUpdateCandidate,
        trace_index: dict[str, TraceLedgerRecord],
        action_mask: dict[str, bool] | None = None,
        rollback_after_write: bool = False,
        rollback_reason: str = "",
        post_update_regression_rate: float = 0.0,
    ) -> SkillUpdateExecutionResult:
        before_payload = self.registry_payload()
        before_hash = stable_hash(before_payload)
        approval = self.approver.approve(candidate, trace_index, self, action_mask, post_update_regression_rate=post_update_regression_rate)
        if not approval.approved:
            return self._result_without_write(update_id, candidate, approval, post_update_regression_rate, before_hash=before_hash)
        rollback_path = self._write_rollback_artifact(update_id, before_payload)
        updated_skill = _skill_from_candidate(candidate, self.graph.get(candidate.target_skill_name))
        self.graph.skills[updated_skill.name] = updated_skill
        self.graph.save()
        self.deterministic_skill_graph_write_count += 1
        self.skill_store_update_write_count += 1
        self.skill_store_write_count += 1
        after_hash = self.registry_hash()
        post_update_regression_passed = _post_update_regression_suite_passed(candidate, self.graph)
        actual_post_update_regression_rate = 0.0 if post_update_regression_passed else 1.0
        reuse_improved = _skill_reuse_improved(candidate, self.graph)
        rolled_back = False
        rollback_successful = False
        final_hash = after_hash
        if rollback_after_write:
            rolled_back = True
            self._restore_payload(before_payload)
            self.rollback_restore_write_count += 1
            self.skill_store_write_count += 1
            final_hash = self.registry_hash()
            rollback_successful = final_hash == before_hash
        self._append_history(
            {
                "update_id": update_id,
                "candidate_id": candidate.candidate_id,
                "target_skill_id": candidate.target_skill_id,
                "target_skill_name": candidate.target_skill_name,
                "update_kind": candidate.update_kind,
                "approval": approval.to_dict(),
                "rollback_artifact_path": str(rollback_path),
                "rolled_back": rolled_back,
                "rollback_reason": rollback_reason,
                "before_registry_hash": before_hash,
                "after_registry_hash": after_hash,
                "final_registry_hash": final_hash,
            }
        )
        return SkillUpdateExecutionResult(
            update_id=update_id,
            candidate_id=candidate.candidate_id,
            target_skill_id=candidate.target_skill_id,
            target_skill_name=candidate.target_skill_name,
            update_kind=candidate.update_kind,
            approved=True,
            written=True,
            rolled_back=rolled_back,
            rollback_successful=rollback_successful,
            reason_code=approval.reason_code,
            evidence_verified=approval.evidence_verified,
            validation_passed=approval.validation_passed,
            regression_suite_passed=approval.regression_suite_passed,
            post_update_regression_rate=actual_post_update_regression_rate,
            post_update_regression_suite_run=True,
            post_update_regression_suite_passed=post_update_regression_passed,
            skill_reuse_improvement_detected=reuse_improved,
            registry_path=str(self.registry_path),
            history_path=str(self.history_path),
            rollback_artifact_path=str(rollback_path),
            before_registry_hash=before_hash,
            after_registry_hash=after_hash,
            final_registry_hash=final_hash,
            approval=approval,
        )

    def registry_payload(self) -> dict[str, Any]:
        return {"skills": [skill.to_dict() for skill in sorted(self.graph.skills.values(), key=lambda item: item.name)]}

    def registry_hash(self) -> str:
        return stable_hash(self.registry_payload())

    def write_paths(self) -> list[Path]:
        paths = [self.registry_path, self.history_path]
        if self.rollback_dir.exists():
            paths.extend(sorted(self.rollback_dir.glob("*.json")))
        return paths

    def writes_outside_state_dir_count(self) -> int:
        state_root = self.registry_path.parent.resolve()
        return sum(1 for path in self.write_paths() if not _path_within(path.resolve(), state_root))

    def _write_rollback_artifact(self, update_id: str, payload: dict[str, Any]) -> Path:
        rollback_path = self.rollback_dir / f"{update_id}_rollback.json"
        rollback_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
        self.rollback_artifact_count += 1
        return rollback_path

    def _restore_payload(self, payload: dict[str, Any]) -> None:
        self.graph.skills = {str(item["name"]): Skill.from_dict(dict(item)) for item in payload.get("skills", [])}
        self.graph.save()

    def _append_history(self, record: dict[str, Any]) -> None:
        self.history_path.parent.mkdir(parents=True, exist_ok=True)
        with self.history_path.open("a", encoding="utf-8") as handle:
            handle.write(canonical_json(record) + "\n")
        self.history_log_append_count += 1

    def recover_from_rollback_artifact(self, *, recovery_id: str, rollback_artifact_path: str | Path) -> tuple[bool, str]:
        path = Path(rollback_artifact_path).resolve()
        rollback_root = self.rollback_dir.resolve()
        if not _path_within(path, rollback_root):
            return False, "rollback_artifact_outside_registry"
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return False, "rollback_artifact_unreadable"
        if not isinstance(payload, dict) or not isinstance(payload.get("skills"), list):
            return False, "invalid_rollback_payload"
        before_hash = self.registry_hash()
        self._restore_payload(payload)
        self.skill_store_write_count += 1
        after_hash = self.registry_hash()
        self._append_history(
            {
                "recovery_id": recovery_id,
                "record_type": "registry_corruption_recovery",
                "rollback_artifact_path": str(path),
                "before_registry_hash": before_hash,
                "after_registry_hash": after_hash,
                "deterministic_recovery": True,
            }
        )
        return True, "registry_recovered_from_rollback"

    def _result_without_write(
        self,
        update_id: str,
        candidate: SkillUpdateCandidate | None,
        approval: SkillUpdateApprovalDecision,
        post_update_regression_rate: float,
        *,
        before_hash: str = "",
    ) -> SkillUpdateExecutionResult:
        return SkillUpdateExecutionResult(
            update_id=update_id,
            candidate_id=approval.candidate_id,
            target_skill_id=candidate.target_skill_id if candidate is not None else "",
            target_skill_name=candidate.target_skill_name if candidate is not None else "",
            update_kind=candidate.update_kind if candidate is not None else "",
            approved=False,
            written=False,
            rolled_back=False,
            rollback_successful=False,
            reason_code=approval.reason_code,
            evidence_verified=approval.evidence_verified,
            validation_passed=approval.validation_passed,
            regression_suite_passed=approval.regression_suite_passed,
            post_update_regression_rate=post_update_regression_rate,
            post_update_regression_suite_run=False,
            post_update_regression_suite_passed=False,
            skill_reuse_improvement_detected=False,
            registry_path=str(self.registry_path),
            history_path=str(self.history_path),
            rollback_artifact_path="",
            before_registry_hash=before_hash,
            after_registry_hash=before_hash,
            final_registry_hash=before_hash,
            approval=approval,
        )


def verify_skill_update_candidate_evidence(candidate: SkillUpdateCandidate, trace_index: dict[str, TraceLedgerRecord]) -> tuple[bool, str]:
    evidence = candidate.evidence_bundle
    source_ids = [str(item) for item in evidence.get("source_trace_ids", [])]
    source_hashes = [str(item) for item in evidence.get("source_trace_hashes", [])]
    if not source_ids or len(source_ids) != len(source_hashes):
        return False, "source_trace_mismatch"
    source_id_set = set(source_ids)
    for trace_id, expected_hash in zip(source_ids, source_hashes):
        trace = trace_index.get(trace_id)
        if trace is None:
            return False, "source_trace_not_found"
        if not trace.verify_hash():
            return False, "source_trace_hash_invalid"
        if trace.content_hash != expected_hash:
            return False, "source_trace_hash_mismatch"
    verifier_trace_ids = {str(item.get("trace_id", "")) for item in evidence.get("verifier_evidence", [])}
    if not verifier_trace_ids or not verifier_trace_ids <= source_id_set:
        return False, "verifier_evidence_not_linked"
    if not all(bool(item.get("success", False)) for item in evidence.get("verifier_evidence", [])):
        return False, "verifier_evidence_failed"
    return True, "ok"


def verify_skill_update_generalization(candidate: SkillUpdateCandidate) -> tuple[bool, str]:
    tests = [dict(item) for item in candidate.generalization_tests]
    if len(tests) < 2:
        return False, "generalization_tests_missing"
    splits = {str(test.get("split", "")) for test in tests}
    if "heldout" not in splits:
        return False, "heldout_validation_missing"
    if "adversarial" not in splits:
        return False, "adversarial_validation_missing"
    if not all(bool(test.get("success", False)) and str(test.get("test_id", "")).strip() for test in tests):
        return False, "generalization_validation_failed"
    return True, "ok"


def verify_skill_update_candidate_payload_safety(candidate: SkillUpdateCandidate) -> tuple[bool, str]:
    if candidate.deterministic_generator != DETERMINISTIC_SKILL_UPDATE_GENERATOR:
        return False, "candidate_generator_mismatch"
    delta = dict(candidate.proposed_delta)
    description = str(delta.get("description", ""))
    if _contains_forbidden_skill_update_text(description):
        return False, "poisoned_skill_description"
    if _contains_false_trigger_metric(delta):
        return False, "false_trigger_update"
    if _contains_forbidden_success_metric(delta):
        return False, "poisoned_success_metric"
    verifier_values = _verifier_values(delta)
    if any(not _verifier_value_allowed(value) for value in verifier_values):
        return False, "poisoned_verifier_name"
    preconditions = delta.get("preconditions")
    if isinstance(preconditions, dict):
        if _contains_unobserved_path_argument(preconditions):
            return False, "unobserved_path_arg"
        if _contains_forbidden_skill_update_text(canonical_json(preconditions)):
            return False, "poisoned_preconditions"
    elif preconditions is not None and _contains_forbidden_skill_update_text(str(preconditions)):
        return False, "poisoned_preconditions"
    return True, "ok"


def _contains_forbidden_skill_update_text(value: object) -> bool:
    text = str(value).lower()
    return any(marker in text for marker in FORBIDDEN_SKILL_UPDATE_TEXT)


def _contains_false_trigger_metric(value: object) -> bool:
    if isinstance(value, dict):
        return any(str(key) in FALSE_TRIGGER_METRIC_KEYS or _contains_false_trigger_metric(item) for key, item in value.items())
    if isinstance(value, list):
        return any(_contains_false_trigger_metric(item) for item in value)
    return False


def _contains_forbidden_success_metric(value: object) -> bool:
    if isinstance(value, dict):
        return any(str(key) in FORBIDDEN_SKILL_UPDATE_METRIC_KEYS or _contains_forbidden_success_metric(item) for key, item in value.items())
    if isinstance(value, list):
        return any(_contains_forbidden_success_metric(item) for item in value)
    return False


def _verifier_values(delta: dict[str, Any]) -> list[str]:
    values: list[str] = []
    for key in ("verifier", "required_verifier"):
        raw = delta.get(key)
        if raw is None:
            continue
        if isinstance(raw, list):
            values.extend(str(item) for item in raw)
        else:
            values.append(str(raw))
    return values


def _verifier_value_allowed(value: str) -> bool:
    normalized = value.strip()
    if not normalized:
        return False
    if normalized not in ALLOWED_SKILL_UPDATE_VERIFIERS:
        return False
    lowered = normalized.lower()
    return not any(marker in lowered for marker in (";", "&&", "|", "$(", "`", "bash", "python -c"))


def _contains_unobserved_path_argument(value: object) -> bool:
    if isinstance(value, dict):
        for key, item in value.items():
            key_text = str(key).lower()
            if key_text in UNOBSERVED_PATH_KEYS or key_text.endswith("_path") or key_text.endswith("_file"):
                if _path_argument_unobserved(item):
                    return True
            if _contains_unobserved_path_argument(item):
                return True
    elif isinstance(value, list):
        return any(_contains_unobserved_path_argument(item) for item in value)
    return False


def _path_argument_unobserved(value: object) -> bool:
    if isinstance(value, list):
        return any(_path_argument_unobserved(item) for item in value)
    text = str(value).strip().lower()
    if not text:
        return False
    if text.startswith("src/") and ".." not in text:
        return False
    return any(marker in text for marker in UNOBSERVED_PATH_MARKERS) or text.startswith("/")


def build_skill_update_impact_audit(
    *,
    results: list[SkillUpdateExecutionResult],
    bad_update_attempt_count: int,
    registry: VersionedSkillRegistry,
) -> SkillUpdateImpactAudit:
    approved = [result for result in results if result.approved and result.written]
    rejected_bad = [result for result in results if not result.approved]
    rollbacks = [result for result in results if result.rolled_back]
    regression_rates = [result.post_update_regression_rate for result in approved]
    post_update_suite_runs = [result for result in approved if result.post_update_regression_suite_run]
    post_update_suite_failures = [result for result in post_update_suite_runs if not result.post_update_regression_suite_passed]
    writes_outside_state_dir_count = registry.writes_outside_state_dir_count()
    heldout_passes = 0
    adversarial_passes = 0
    evidence_mismatch_rejections = 0
    action_mask_rejections = 0
    for result in results:
        if result.reason_code == "source_trace_hash_mismatch":
            evidence_mismatch_rejections += 1
        if result.reason_code == "action_mask_violation":
            action_mask_rejections += 1
        if result.approved and result.validation_passed:
            heldout_passes += 1
            adversarial_passes += 1
    return SkillUpdateImpactAudit(
        approved_skill_update_count=len(approved),
        rejected_bad_update_count=len(rejected_bad),
        bad_update_reject_rate=_rate(len(rejected_bad), bad_update_attempt_count),
        rollback_count=len(rollbacks),
        rollback_success_rate=_rate(sum(1 for result in rollbacks if result.rollback_successful), len(rollbacks)),
        post_update_regression_rate=max(regression_rates) if regression_rates else 0.0,
        post_update_regression_suite_run_count=len(post_update_suite_runs),
        post_update_regression_failure_count=len(post_update_suite_failures),
        skill_reuse_improvement_detected=any(result.skill_reuse_improvement_detected for result in approved),
        deterministic_skill_graph_write_count=registry.deterministic_skill_graph_write_count,
        skill_store_update_write_count=registry.skill_store_update_write_count,
        rollback_restore_write_count=registry.rollback_restore_write_count,
        skill_store_write_count=registry.skill_store_write_count,
        history_log_append_count=registry.history_log_append_count,
        rollback_artifact_count=registry.rollback_artifact_count,
        learned_skill_graph_writes_executed=registry.learned_skill_graph_writes_executed,
        write_path_containment_passed=writes_outside_state_dir_count == 0,
        writes_outside_state_dir_count=writes_outside_state_dir_count,
        evidence_hash_mismatch_rejected_count=evidence_mismatch_rejections,
        action_mask_rejected_count=action_mask_rejections,
        heldout_validation_pass_count=heldout_passes,
        adversarial_validation_pass_count=adversarial_passes,
    )


def skill_update_approval_artifact_hash(payload: dict[str, Any]) -> str:
    return stable_hash(payload)


def _skill_from_candidate(candidate: SkillUpdateCandidate, existing: Skill | None) -> Skill:
    delta = dict(candidate.proposed_delta)
    if candidate.update_kind == "create_skill":
        return Skill(
            name=candidate.target_skill_name,
            description=str(delta.get("description", f"Evidence-backed skill {candidate.target_skill_name}.")),
            triggers=[str(item) for item in delta.get("triggers", [candidate.target_skill_name])],
            steps=[str(item) for item in delta.get("steps", ["Apply deterministic verified skill update"])],
            verifier=str(delta.get("verifier", "pytest")),
            promoted=candidate.proposed_status == "promoted",
            success_count=0,
            failure_count=0,
            version=candidate.proposed_version,
            metadata={"source": "gate7d_skill_update_approval", "candidate_id": candidate.candidate_id},
            skill_id=candidate.target_skill_id,
            status=candidate.proposed_status,
            required_verifier=[str(item) for item in delta.get("required_verifier", ["pytest"])],
            promotion_evidence_count=len(candidate.evidence_bundle.get("source_trace_ids", [])),
            heldout_success_rate=1.0,
            average_saved_steps=1.0,
            regression_status="passed",
            verifier_pass_count=len(candidate.evidence_bundle.get("verifier_evidence", [])),
            precision=1.0,
            adversarial_success_rate=1.0,
        )
    if existing is None:
        raise ValueError("existing skill required for non-create update")
    data = existing.to_dict()
    for key in ["description", "triggers", "steps", "required_verifier", "preconditions", "parameter_slots", "bytecode_plan"]:
        if key in delta:
            data[key] = delta[key]
    data["version"] = candidate.proposed_version
    data["status"] = candidate.proposed_status
    data["promoted"] = candidate.proposed_status == "promoted"
    data["metadata"] = {
        **dict(data.get("metadata") or {}),
        "gate7d_update": {
            "candidate_id": candidate.candidate_id,
            "update_kind": candidate.update_kind,
            "source_trace_ids": list(candidate.evidence_bundle.get("source_trace_ids", [])),
        },
    }
    if candidate.update_kind == "deprecate_skill":
        data["metadata"] = {**dict(data["metadata"]), "deprecation_reason": str(delta.get("reason", "gate7d_deprecation"))}
    if candidate.update_kind == "quarantine_skill":
        data["quarantine_reason"] = str(delta.get("quarantine_reason", "gate7d_quarantine"))
    return Skill.from_dict(data)


def _skill_reuse_improved(candidate: SkillUpdateCandidate, graph: SkillGraph) -> bool:
    matches = graph.match(" ".join([candidate.target_skill_name, *[str(item) for item in candidate.proposed_delta.get("triggers", [])]]), promoted_only=True)
    if candidate.proposed_status in {"deprecated", "quarantined"}:
        return all(match.skill.name != candidate.target_skill_name for match in matches)
    return any(match.skill.name == candidate.target_skill_name and match.skill.version == candidate.proposed_version for match in matches)


def _post_update_regression_suite_passed(candidate: SkillUpdateCandidate, graph: SkillGraph) -> bool:
    skill = graph.get(candidate.target_skill_name)
    if skill is None:
        return False
    if skill.version != candidate.proposed_version:
        return False
    if skill.status != candidate.proposed_status:
        return False
    if candidate.update_kind in {"deprecate_skill", "quarantine_skill"}:
        default_matches = graph.match(candidate.target_skill_name, promoted_only=True)
        return all(match.skill.name != candidate.target_skill_name for match in default_matches)
    if candidate.update_kind == "create_skill":
        return bool(skill.triggers and skill.steps and skill.required_verifier)
    return _skill_reuse_improved(candidate, graph)


def _path_within(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
    except ValueError:
        return False
    return True


def _rejected_approval(candidate_id: str, reason_code: str) -> SkillUpdateApprovalDecision:
    payload = {"candidate_id": candidate_id, "reason_code": reason_code, "approved": False, "approver": DETERMINISTIC_SKILL_UPDATE_APPROVER}
    return SkillUpdateApprovalDecision(
        approval_id="skillupd_approval_" + stable_hash(payload)[:20],
        candidate_id=candidate_id,
        approved=False,
        reason_code=reason_code,
        evidence_verified=False,
        validation_passed=False,
        regression_suite_passed=False,
    )


def _rate(numerator: int | float, denominator: int | float) -> float:
    if denominator == 0:
        return 0.0
    return float(numerator) / float(denominator)
