# -*- coding: utf-8 -*-
"""GovernancePolicy — builtin_rules + user_rules two-layer policy engine.

- builtin_rules + user_rules two layers:
    builtin_rules — system built-in protection (resource ask /
        command deny), not modifiable by agent
    user_rules    — generated by user/approve, can be added or removed
- Evaluation flow: ToolRegistry type check → builtin_rules →
    user_rules → global fallback
- Rule generalization: smart generalization on approve (first token +
    *), high-risk commands stay exact
"""

from __future__ import annotations
import copy
import logging
import re
import threading
from collections.abc import Sequence
from dataclasses import dataclass, field
from enum import Enum
from fnmatch import fnmatch
from pathlib import Path
from typing import Any, List, Optional

import yaml

from .tool_registry import ToolRegistry, DEFAULT_REGISTRY
from ..sandbox import SandboxConfig
from ..utils.io_utils import write_yaml_atomic

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Core types
# ---------------------------------------------------------------------------


class GovernanceAction(str, Enum):
    ALLOW = "allow"
    DENY = "deny"
    ASK = "ask"
    # bash tool with no hit → sandbox fallback
    SANDBOX_FALLBACK = "sandbox_fallback"


@dataclass
class GovernanceDecision:
    action: GovernanceAction
    reason: str
    sandbox_config: SandboxConfig | None = None
    findings: list[Any] | None = None  # GuardFinding list for approval card
    # Origin of the decision, surfaced as the "Source" line on the approval
    # card. Identifies which policy mechanism drove the decision:
    #   "builtin_rules"         — Phase 2 DEFAULT_BUILTIN_RULES hit
    #   "user_rules"            — Phase 2 user_rules hit
    #   "sensitive_paths"       — Phase 1 finding (sensitive_path_detector)
    #   "detection_rules"       — Phase 1 finding (pattern_detector)
    #   "shell_evasion_checks"  — Phase 1 finding (shell_evasion_detector)
    #   "shell_danger_keywords" — Phase 1.5 hardcoded shell-danger regex
    #   "sandbox"               — Phase 3 shell sandbox fallback
    #   "No rule hit"           — Phase 3 fallback (no rule/finding matched)
    source: str = "No rule hit"


class ToolCallSpec:
    """Specification of a tool call (used by the governor for evaluation).

    Attributes:
        tool_name: Tool name, e.g. "Read", "Bash", "Write"
        target: Tool's target argument, e.g. "src/main.py", "git push"
        agent_id: Agent ID that initiated the call
        session_id: Current session ID
        raw_params: Raw tool call arguments dict
    """

    def __init__(
        self,
        tool_name: str,
        target: str,
        agent_id: str,
        session_id: str,
        raw_params: dict[str, Any] | None = None,
    ) -> None:
        self.tool_name = tool_name
        self.target = target
        self.agent_id = agent_id
        self.session_id = session_id
        self.raw_params = raw_params or {}


@dataclass
class GovernanceRule:
    """Unified policy rule.

    match format: "ToolName(pattern)"
        - ToolName: Read, Write, Bash, Browser, etc.; "*" matches all tools
        - pattern: glob pattern matched against the tool's target argument

    Examples:
        GovernanceRule(match="Bash(git *)", action=GovernanceAction.ALLOW)
        GovernanceRule(match="Write(.env*)", action=GovernanceAction.DENY)
        GovernanceRule(
            match="*(.ssh/**)", action=GovernanceAction.ASK
        )  # all tools
        GovernanceRule(match="Read(src/**)", action=GovernanceAction.ALLOW,
                   grantee="agent-abc", duration="session")

    Action semantics vary by tool type:
        Explicit-resource tools:
            allow → execute directly (no sandbox)
            deny  → reject
            ask   → ask user, execute directly after approval
        Bash-type tools:
            allow → execute in sandbox (pre-authorized, no violation expected)
            deny  → reject (no sandbox)
            ask   → ask user, execute in sandbox after approval
    """

    match: str  # "ToolName(pattern)"
    action: GovernanceAction = GovernanceAction.DENY
    reason: str = ""  # Rule description
    grantee: str = "*"  # Authorized subject
    duration: str = "permanent"  # "session" | "permanent"
    session_id: Optional[str] = None  # Chat session ID

    def _globmatch(self, pattern: str, target: str) -> bool:
        """wcmatch globmatch with directory self-match support."""
        from wcmatch import glob

        flags = (
            glob.GLOBSTAR
            | glob.BRACE
            | glob.NEGATE
            | glob.SPLIT
            | glob.DOTGLOB
        )
        if glob.globmatch(target, pattern, flags=flags):
            return True
        if pattern.endswith("/**"):
            dir_pattern = pattern[:-3]
            if glob.globmatch(target, dir_pattern, flags=flags):
                return True
        return False

    def matches_tool_call(
        self,
        tc_spec: ToolCallSpec,
        tool_type: str = "",
    ) -> bool:
        """Determine whether this rule matches the given tool call.

        Matching logic:
            1. grantee match ("*" matches all)
            2. session-level rule: session_id must match
            3. Parse self.match into (rule_tool, rule_pattern)
            4. If rule_tool is "*" it matches all tools
            5. Glob match target against rule_pattern
               - For "*" rules + shell tool, additional substring matching
                 (command may contain sensitive file paths, e.g. "cat .env")

        Note: caller is responsible for resolving target to absolute path
              (WORKSPACE_DIR replacement is done at load time).
        """
        # grantee check
        if self.grantee not in ("*", tc_spec.agent_id):
            return False
        # session-level rule: bound to a specific chat session.
        # Fail-closed: if the rule is session-scoped, the request MUST carry
        # the same session_id. A missing tc_spec.session_id means we cannot
        # prove the request belongs to the bound session, so we refuse to
        # match (do NOT silently fall through, which would let any unscoped
        # request match a session rule).
        if self.duration == "session" and self.session_id:
            if not tc_spec.session_id or self.session_id != tc_spec.session_id:
                return False
        # parse "ToolName(pattern)"
        rule_tool, rule_pattern = _parse_match(self.match)
        # "*" matches all tools (used for builtin resource protection)
        if rule_tool not in ("*", tc_spec.tool_name):
            return False
        is_wildcard = rule_tool == "*"

        use_globmatch = is_wildcard or tool_type == "file"

        if use_globmatch:
            return self._globmatch(rule_pattern, tc_spec.target)
        else:
            return fnmatch(tc_spec.target, rule_pattern)


# ---------------------------------------------------------------------------
# builtin_rules
# ---------------------------------------------------------------------------

# (match, reason) pairs for builtin ASK rules, grouped by concern. Expanded
# into GovernanceRule objects below. All share action=ASK / grantee="*" /
# duration="permanent" defaults, so only match+reason need to vary.
_BUILTIN_ASK_SPECS: List[tuple[str, str]] = [
    # ── Resource protection (any tool access requires confirmation) ──
    ("*(**/.env*)", "Env file may contain secrets/credentials"),
    ("*(**/.ssh/**)", "SSH credentials directory"),
    ("*(**/*.pem)", "Private key file"),
    ("*(**/*.key)", "Private key file"),
    ("*(**/*.p12)", "PKCS#12 certificate bundle"),
    ("*(**/*.pfx)", "PKCS#12 certificate bundle"),
    ("*(**/.aws/**)", "AWS credentials directory"),
    ("*(**/.gnupg/**)", "GPG keys directory"),
    ("*(**/.kube/**)", "Kubernetes config directory"),
    ("*(**/.netrc)", "Netrc login credentials file"),
    ("*(**/.npmrc)", "npm auth token file"),
    ("*(**/.pypirc)", "PyPI API token file"),
    # ── LLM provider / AI coding-assistant configs (may hold API keys) ──
    # Generic-name fallback (api_key*/apikey*) catches future providers
    # whose key file is named accordingly; vendor dirs cover the known
    # ones. Shell rc files and PowerShell profiles are included because
    # users frequently `export`/`setx` API keys into them.
    ("*(**/api_key*)", "API key file by generic name"),
    ("*(**/apikey*)", "API key file by generic name"),
    ("*(**/.anthropic/**)", "Anthropic CLI/SDK config may contain API key"),
    ("*(**/.config/anthropic/**)", "Anthropic config may contain API key"),
    ("*(**/.openai/**)", "OpenAI config may contain API key"),
    ("*(**/.config/openai/**)", "OpenAI config (XDG) may contain API key"),
    ("*(**/.codex/**)", "OpenAI Codex CLI config may contain API key"),
    ("*(**/.gemini/**)", "Gemini CLI config may contain API key"),
    ("*(**/.config/gemini/**)", "Gemini config (XDG) may contain API key"),
    ("*(**/.claude/**)", "Claude config/memory may contain API key"),
    ("*(**/.cursor/**)", "Cursor config may contain token"),
    ("*(**/.copilot/**)", "GitHub Copilot config may contain token"),
    ("*(**/.codeium/**)", "Codeium config may contain token"),
    ("*(**/.opencode/**)", "OpenCode config may contain API key"),
    # ── Shell rc / history (users often `export` API keys here) ──
    ("*(**/.bashrc)", "Shell rc may export API keys"),
    ("*(**/.zshrc)", "Shell rc may export API keys"),
    ("*(**/.profile)", "Shell rc may export API keys"),
    ("*(**/.bash_profile)", "Shell rc may export API keys"),
    ("*(**/.bash_history)", "Shell history may contain typed secrets"),
    ("*(**/.zsh_history)", "Shell history may contain typed secrets"),
    # ── Windows (PowerShell profile + history may hold keys) ──
    ("*(**/Microsoft.PowerShell_profile.ps1)", "PS profile may setx API keys"),
    ("*(**/ConsoleHost_history.txt)", "PowerShell may contain typed secrets"),
    ("Bash(sudo *)", "Privilege escalation, ASK"),
    ("Bash(gh repo delete *)", "Repository deletion, ASK"),
    ("Bash(gh api -X DELETE *)", "Destructive GitHub API calls, ASK"),
    ("Bash(gh api --method DELETE *)", "Destructive GitHub API calls, ASK"),
    ("Bash(gh api --method=DELETE *)", "Destructive GitHub API calls, ASK"),
]

# (match, reason) pairs for builtin DENY rules — hard walls, never allowed.
# NOTE: these glob patterns are best-effort fast-path checks. The
# authoritative shell danger detection is in
# ``_check_shell_danger_keywords()`` below, which uses regex to catch
# command variants that fnmatch cannot match.
_BUILTIN_DENY_SPECS: List[tuple[str, str]] = [
    ("Bash(rm * -rf *//*)", "Root filesystem deletion"),
]

DEFAULT_BUILTIN_RULES: List[GovernanceRule] = [
    GovernanceRule(match=m, action=GovernanceAction.ASK, reason=r)
    for m, r in _BUILTIN_ASK_SPECS
] + [
    GovernanceRule(match=m, action=GovernanceAction.DENY, reason=r)
    for m, r in _BUILTIN_DENY_SPECS
]


# ---------------------------------------------------------------------------
# Shell danger keyword patterns (supplement to glob-based deny rules)
# ---------------------------------------------------------------------------

_SHELL_DANGER_PATTERNS: list[tuple[re.Pattern[str], str]] = [
    # rm with recursive flag targeting root-ish paths (/, /*, /etc, ...)
    # Catches: rm -rf /, rm -r -f /*, rm    -rf  /, rm / -rf, etc.
    (
        re.compile(
            r"\brm\b(?=[^;|&]*\s+-[a-zA-Z]*[rR])[^;|&]*\s+/(?:\s|$|\*)",
        ),
        "Recursive deletion targeting root filesystem",
    ),
    # Fork bomb patterns
    (
        re.compile(
            r":\(\)\s*\{\s*:\|:\s*&\s*\}",
        ),
        "Fork bomb",
    ),
    # Direct disk write
    (
        re.compile(
            r">\s*/dev/[sh]d[a-z]|\bdd\b[^;|&]*\bof\s*=\s*/dev/",
        ),
        "Direct disk device write",
    ),
    # mkfs (filesystem formatting)
    (
        re.compile(
            r"\bmkfs\b",
        ),
        "Filesystem formatting",
    ),
]


def _check_shell_danger_keywords(command: str) -> Optional[str]:
    """Check a shell command for dangerous patterns via regex.

    Returns the deny reason if a danger pattern matches, else ``None``.
    This supplements the glob-based builtin deny rules to catch command
    variants that ``fnmatch`` cannot match (e.g. flag reordering,
    extra whitespace, absolute path to sudo, piped sudo, etc.).
    """
    # Normalize whitespace for more reliable matching
    normalized = " ".join(command.split())
    for pattern, reason in _SHELL_DANGER_PATTERNS:
        if pattern.search(normalized):
            return reason
    return None


# ---------------------------------------------------------------------------
# Default sandbox deny_paths — paths forbidden for sandbox processes
# ---------------------------------------------------------------------------

DEFAULT_ENV_BLACKLIST: List[str] = [
    "OPENAI_API_KEY",
    "ANTHROPIC_API_KEY",
    "AWS_SECRET_ACCESS_KEY",
    "AWS_ACCESS_KEY_ID",
]


DEFAULT_SANDBOX_DENY_PATHS: List[str] = [
    # SSH keys and config
    "~/.ssh",
    # AWS credentials
    "~/.aws",
    # GPG keys
    "~/.gnupg",
    # Kubernetes config
    "~/.kube",
    # Google Cloud credentials
    "~/.config/gcloud",
    # Azure CLI credentials
    "~/.azure",
    # Docker auth config
    "~/.docker/config.json",
    # Generic env files (may contain secrets)
    "~/.env",
    # Claude Code config & memory
    "~/.claude",
    # macOS Keychain database
    "~/Library/Keychains",
    # Browser credentials
    "~/Library/Application Support/Google/Chrome/Default/Login Data",
    "~/Library/Application Support/Firefox/Profiles",
    # Git credentials
    "~/.git-credentials",
    "~/.gitconfig",  # may contain tokens in URL
    # Terraform state (may contain sensitive outputs)
    "~/.terraformrc",
    # HashiCorp Vault token
    "~/.vault-token",
    # npm / yarn auth tokens
    "~/.npmrc",
    "~/.yarnrc",
    # pip / PyPI API tokens
    "~/.pypirc",
    # Other common sensitive configs
    # NOTE: ~/.config/gh intentionally excluded — gh CLI needs to read its
    # auth config (hosts.yml) for operations. The gh ALLOW rule + DENY for
    # destructive subcommands provides the safety boundary instead.
    "~/.config/nix",  # Nix config
    "~/.netrc",  # generic login credentials
]

FILE_READ_TOOLS: frozenset[str] = frozenset(
    {"Read", "ViewImage", "ViewVideo", "SendFileToUser"},
)
FILE_WRITE_TOOLS: frozenset[str] = frozenset({"Write", "Edit", "Append"})


# ---------------------------------------------------------------------------
# Default user_rules (for cold start)
# ---------------------------------------------------------------------------

# Path-level / environment rules kept here manually (cannot be expressed
# by ``@tool_descriptor``). Tool-level ``ToolName(**)`` ALLOW/ASK/DENY
# rules are generated from descriptor ``default_policy`` fields via
# :func:`get_default_user_rules`.
_PATH_LEVEL_USER_RULES: List[GovernanceRule] = [
    # NOTE: Internal tools need no rule — Phase 0 short-circuits them.
    # ── File tools (operations within WORKSPACE_DIR, always allowed) ──
    GovernanceRule(
        match="Read(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File read within workspace",
    ),
    GovernanceRule(
        match="Write(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File write within workspace",
    ),
    GovernanceRule(
        match="Edit(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File edit within workspace",
    ),
    GovernanceRule(
        match="Append(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File append within workspace",
    ),
    GovernanceRule(
        match="Grep(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="Content search within workspace",
    ),
    GovernanceRule(
        match="Glob(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File listing within workspace",
    ),
    GovernanceRule(
        match="ViewImage(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="Image view within workspace",
    ),
    GovernanceRule(
        match="ViewVideo(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="Video view within workspace",
    ),
    GovernanceRule(
        match="SendFileToUser(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File send within workspace",
    ),
    GovernanceRule(
        match="DesktopScreenshot(WORKSPACE_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="File send within workspace",
    ),
    # ── /tmp ──
    GovernanceRule(
        match="*(/tmp/**)",
        action=GovernanceAction.ALLOW,
        reason="Scratch directory access for tmp",
    ),
    # Coding Mode, project dir
    GovernanceRule(
        match="*(CODING_PROJECT_DIR/**)",
        action=GovernanceAction.ALLOW,
        reason="Coding project dir",
    ),
    # ALLOW all other gh operations
    GovernanceRule(
        match="Bash(gh)",
        action=GovernanceAction.ALLOW,
        reason="GitHub CLI operations",
    ),
    GovernanceRule(
        match="Bash(gh *)",
        action=GovernanceAction.ALLOW,
        reason="GitHub CLI operations",
    ),
]


_AUTO_DEFAULT_USER_RULES_CACHE: List[GovernanceRule] | None = None
_AUTO_DEFAULT_USER_RULES_LOCK = threading.Lock()


def _auto_default_user_rules() -> List[GovernanceRule]:
    """Build tool-level rules from ``@tool_descriptor(default_policy=...)``.

    Import of ``agents.tools`` is deferred so this stays safe when
    ``policy`` is imported before tool modules.

    Rule order follows ``get_builtin_tool_funcs()`` collection order
    (``agents/tools/__init__.py`` import order). Current auto rules are
    non-overlapping ``ToolName(**)`` ALLOW entries, so order does not
    change Phase 2 first-match outcomes; keep that invariant when adding
    ASK/DENY defaults.

    Successful builds are cached so request paths do not rescan descriptors.
    """
    global _AUTO_DEFAULT_USER_RULES_CACHE
    with _AUTO_DEFAULT_USER_RULES_LOCK:
        if _AUTO_DEFAULT_USER_RULES_CACHE is not None:
            return list(_AUTO_DEFAULT_USER_RULES_CACHE)

        try:
            from ..agents import tools as _agents_tools  # noqa: F401
            from ..runtime.tool_registry import get_builtin_tool_funcs
            from .tool_registry import snake_to_pascal

            _ = _agents_tools
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "Failed to build auto default user rules from descriptors: %s",
                exc,
            )
            return []

        action_map = {
            "allow": GovernanceAction.ALLOW,
            "ask": GovernanceAction.ASK,
            "deny": GovernanceAction.DENY,
        }
        rules: List[GovernanceRule] = []
        for fn in get_builtin_tool_funcs():
            desc = getattr(fn, "_tool_descriptor", None)
            if desc is None:
                continue
            gov = getattr(desc, "governance", None)
            if gov is None or not gov.default_policy:
                continue
            policy_key = gov.default_policy.strip().lower()
            action = action_map.get(policy_key)
            if action is None:
                raise ValueError(
                    f"Invalid default_policy {gov.default_policy!r} on "
                    f"tool {desc.name!r}; expected allow|ask|deny",
                )
            pname = gov.policy_name or snake_to_pascal(desc.name)
            rules.append(
                GovernanceRule(
                    match=f"{pname}(**)",
                    action=action,
                    reason=gov.policy_reason or f"Auto-registered: {pname}",
                ),
            )
        _AUTO_DEFAULT_USER_RULES_CACHE = rules
        return list(rules)


def get_default_user_rules() -> List[GovernanceRule]:
    """Path-level defaults + auto-generated tool-level descriptor rules.

    Concatenation order (matters for Phase 2 first-match on ``user_rules``):

    1. ``_PATH_LEVEL_USER_RULES`` — workspace / tmp / coding-project / gh
    2. ``_auto_default_user_rules()`` — ``ToolName(**)`` from descriptors

    Today tool-level ALLOW rules are supersets of the path-level ALLOW rules
    for the same tools, so reordering relative to the historical interleaved
    list is semantically equivalent. Do **not** add tool-level DENY/ASK
    defaults that would need to lose to a more specific path ALLOW without
    revisiting this order.

    If tests (or callers) replace module-level ``DEFAULT_USER_RULES`` with a
    plain ``list``, that override is returned as-is so migration tests can
    inject synthetic default rules.
    """
    # Late lookup avoids binding the proxy before it exists; a patched plain
    # list (``type is list``) wins over the auto-built defaults.
    override = globals().get("DEFAULT_USER_RULES")
    if type(override) is list:
        return list(override)
    return list(_PATH_LEVEL_USER_RULES) + _auto_default_user_rules()


# Backward-compatible name used by tests and callers. Resolved lazily so
# descriptor imports are available when the list is first read.
class _DefaultUserRulesProxy(Sequence):
    """Lazy :class:`~collections.abc.Sequence` over default user rules.

    Does **not** subclass ``list``. Prefer
    :func:`get_default_user_rules` / ``list(DEFAULT_USER_RULES)`` for
    mutation or persistence. ``+`` materializes to a real ``list``.
    """

    def _rules(self) -> List[GovernanceRule]:
        return get_default_user_rules()

    def __iter__(self):
        return iter(self._rules())

    def __len__(self) -> int:
        return len(self._rules())

    def __getitem__(self, index):
        return self._rules()[index]

    def __repr__(self) -> str:
        return repr(self._rules())

    def __eq__(self, other: object) -> bool:
        if other is self:
            return True
        # str/bytes are Sequences but must not compare element-wise.
        if isinstance(other, (str, bytes)):
            return False
        if isinstance(other, (list, tuple, Sequence)):
            return list(self._rules()) == list(other)
        return NotImplemented

    def __add__(self, other: object) -> List[GovernanceRule]:
        if isinstance(other, (str, bytes)):
            return NotImplemented
        if isinstance(other, (list, tuple, Sequence)):
            return list(self._rules()) + list(other)
        return NotImplemented

    def __radd__(self, other: object) -> List[GovernanceRule]:
        if isinstance(other, (str, bytes)):
            return NotImplemented
        if isinstance(other, (list, tuple, Sequence)):
            return list(other) + list(self._rules())
        return NotImplemented


DEFAULT_USER_RULES: List[
    GovernanceRule
] = _DefaultUserRulesProxy()  # type: ignore[assignment]


# ---------------------------------------------------------------------------
# GovernancePolicy
# ---------------------------------------------------------------------------


@dataclass
class DetectionRuleConfig:
    """A single pattern-based detection rule (loaded from policy.yaml)."""

    id: str
    tools: List[str] = field(default_factory=list)
    params: List[str] = field(default_factory=list)
    category: str = "command_injection"
    severity: str = "HIGH"
    patterns: List[str] = field(default_factory=list)
    exclude_patterns: List[str] = field(default_factory=list)
    description: str = ""
    remediation: str = ""


@dataclass
class GovernancePolicy:
    """Policy: builtin_rules + user_rules two layers, first-match-wins.

    Loaded from policy_dir/policy.yaml, held by ResourceGovernor.

    Evaluation flow (v2.0):
        Phase 1: Deep security scan (findings accumulation)
        Phase 2: builtin_rules + user_rules first-match-wins
        Phase 3: Fallback + execution_level threshold

    Lifecycle:
        load → evaluate (hot path) → add_rule (user approve) → save
    """

    version: str = "2.0"
    builtin_rules: List[GovernanceRule] = field(default_factory=list)
    user_rules: List[GovernanceRule] = field(default_factory=list)
    env_blacklist: List[str] = field(default_factory=list)
    audit_level: str = "all"  # "all" | "write_only" | "none"

    # v2.0 fields
    execution_level: str = "smart"  # "off" | "auto" | "smart" | "strict"
    sensitive_paths: List[str] = field(default_factory=list)
    shell_evasion_checks: dict[str, bool] = field(default_factory=dict)
    detection_rules: List[DetectionRuleConfig] = field(default_factory=list)

    # Default-rule migrations already applied to this policy (rule match
    # strings). Once a migration is recorded here, the corresponding default
    # user rule is never re-added on load — so deleting it stays deleted.
    applied_migrations: List[str] = field(default_factory=list)

    # Internal reference to registry (defaults to module-level
    # DEFAULT_REGISTRY)
    _registry: ToolRegistry = field(
        default=None,
        repr=False,
        compare=False,
    )

    def __post_init__(self) -> None:
        if self._registry is None:
            self._registry = DEFAULT_REGISTRY

    # ------------------------------------------------------------------
    # Read-only view
    # ------------------------------------------------------------------

    @property
    def rules(self) -> List[GovernanceRule]:
        """Merge builtin_rules + user_rules (read-only snapshot).

        Note: returns a snapshot; modifications will not affect the
        original list.
        Use add_rule() to add new rules.
        """
        return list(self.builtin_rules) + list(self.user_rules)

    def evaluate(  # noqa: E501  pylint: disable=too-many-return-statements,too-many-branches
        self,
        tc_spec: ToolCallSpec,
    ) -> GovernanceDecision:
        """Evaluate policy decision for a tool call.

        Three-phase evaluation (v2.0):
            Phase 0: Type check — unknown → DENY, internal → ALLOW
            Phase 1: Deep security scan (accumulates findings)
                     CRITICAL findings → immediate DENY
            Phase 2: Policy rules first-match-wins
                     (builtin_rules + user_rules)
            Phase 3: Fallback + execution_level threshold
                     shell: findings honored (MEDIUM+ -> ASK in SMART),
                     else SANDBOX_FALLBACK; other tools -> fallback

        Returns: GovernanceDecision (with optional findings attached)
        """
        # ── Phase 0: ToolRegistry type check ──
        tool_type = self._registry.get_type(tc_spec.tool_name)
        if tool_type == "unknown":
            return GovernanceDecision(
                action=GovernanceAction.DENY,
                reason=f"Unregistered tool: {tc_spec.tool_name}",
                source="unknown tool",
            )
        if tool_type == "internal":
            return GovernanceDecision(
                action=GovernanceAction.ALLOW,
                reason="internal",
            )

        # execution_level == OFF → skip Phase 1, go to Phase 2
        skip_deep_scan = self.execution_level.lower() == "off"

        # ── Phase 1: Deep security scan ──
        findings: list[Any] = []
        if not skip_deep_scan:
            findings = self._deep_security_scan(tc_spec, tool_type)
            # CRITICAL findings → immediate DENY
            if any(getattr(f, "severity", "") == "CRITICAL" for f in findings):
                top = _top_finding(findings)
                return GovernanceDecision(
                    action=GovernanceAction.DENY,
                    reason=getattr(top, "description", "") or top.title,
                    findings=findings,
                    source=_findings_source(findings),
                )

        # ── Phase 1.5: Shell danger keyword detection ──
        # Regex-based check that catches command variants missed by
        # the fnmatch-based builtin deny rules.
        if tool_type == "shell" and tc_spec.target:
            danger_reason = _check_shell_danger_keywords(tc_spec.target)
            if danger_reason:
                return GovernanceDecision(
                    action=GovernanceAction.DENY,
                    reason=danger_reason,
                    findings=findings,
                    source="shell_danger_keywords",
                )

        # ── Phase 2: builtin_rules + user_rules (first-match-wins) ──
        is_strict = self.execution_level == "strict"

        for rule in self.builtin_rules:
            if rule.matches_tool_call(
                tc_spec,
                tool_type=tool_type,
            ):
                action = GovernanceAction(rule.action.value)
                # STRICT mode: override ALLOW → ASK so every tool
                # call requires approval (DENY still honoured).
                if action == GovernanceAction.ALLOW and is_strict:
                    return GovernanceDecision(
                        action=GovernanceAction.ASK,
                        reason="STRICT mode: all tool calls require approval",
                        findings=findings or None,
                        source="STRICT mode",
                    )
                return GovernanceDecision(
                    action=action,
                    reason=rule.reason,
                    findings=findings or None,
                    source="builtin_rules",
                )

        for rule in self.user_rules:
            if rule.matches_tool_call(
                tc_spec,
                tool_type=tool_type,
            ):
                action = GovernanceAction(rule.action.value)
                if action == GovernanceAction.ALLOW and is_strict:
                    return GovernanceDecision(
                        action=GovernanceAction.ASK,
                        reason="STRICT mode: all tool calls require approval",
                        findings=findings or None,
                        source="STRICT mode",
                    )
                return GovernanceDecision(
                    action=action,
                    reason=rule.reason,
                    findings=findings or None,
                    source="user_rules",
                )

        # ── Phase 3: Fallback + execution_level threshold ──
        if tool_type == "shell":
            # STRICT mode: shell commands also require approval
            if is_strict:
                return GovernanceDecision(
                    action=GovernanceAction.ASK,
                    reason="STRICT mode: all tool calls require approval",
                    findings=findings or None,
                    source="STRICT mode",
                )
            # Finding-driven approval for shell: when the deep scan surfaced
            # findings (e.g. a frontend custom rule matched the command),
            # honor the execution-level severity threshold. MEDIUM+ findings
            # in SMART, or any finding in AUTO/OFF, escalate to ASK so the
            # user's own detection rules take effect for shell tools too.
            # INFO/LOW findings (fallback returns ALLOW) fall through to the
            # sandbox path below, where sandbox isolation is the safety net.
            if findings:
                fb = self._apply_execution_level_fallback(tc_spec, findings)
                if fb.action is not GovernanceAction.ALLOW:
                    return fb
            # No findings (or only INFO/LOW): route to sandbox. When the
            # sandbox is unusable, ResourceGovernor downgrades this to ALLOW.
            return GovernanceDecision(
                action=GovernanceAction.SANDBOX_FALLBACK,
                reason="sandbox fallback",
                findings=findings or None,
                source="sandbox",
            )
        return self._apply_execution_level_fallback(tc_spec, findings)

    def _deep_security_scan(
        self,
        tc_spec: ToolCallSpec,
        tool_type: str,
    ) -> list[Any]:
        """Run deep security detectors (Phase 1).

        Merges policy.yaml detection_rules/shell_evasion_checks with the
        frontend Security page configuration (config.json →
        security.tool_guard). Uses the mtime-cached load_config() so there
        is negligible overhead on the hot path.  On config read failure the
        scan falls back to policy.yaml-only rules (graceful degradation).

        Delegates to governance.detectors module.
        Returns a list of GuardFinding objects.
        """
        try:
            from .detectors import run_deep_scan

            # Merge config.json custom_rules + shell_evasion_checks into
            # the policy.yaml-sourced rules so frontend settings take
            # effect in governance evaluation.
            detection_rules, shell_evasion_checks = self._merge_config_rules()

            return run_deep_scan(
                tool_name=tc_spec.tool_name,
                target=tc_spec.target,
                tool_type=tool_type,
                sensitive_paths=self.sensitive_paths,
                detection_rules=detection_rules,
                shell_evasion_checks=shell_evasion_checks,
                raw_params=tc_spec.raw_params,
            )
        except Exception as exc:
            logger.warning(
                "deep_security_scan failed: %s; continuing without",
                exc,
            )
            return []

    def _merge_config_rules(
        self,
    ) -> tuple[list[Any], dict[str, bool]]:
        """Merge policy.yaml rules with config.json security.tool_guard.

        Returns:
            (merged_detection_rules, merged_shell_evasion_checks)

        - custom_rules from config are appended to policy detection_rules.
        - disabled_rules from config filter out rules by ID from both
          policy.yaml and config custom_rules.
        - shell_evasion_checks are OR-merged: enabled in either source
          means enabled.
        - On any config read error, returns the unmodified policy.yaml
          values (graceful degradation).
        """
        try:
            from ..config import load_config

            cfg = load_config().security.tool_guard
        except Exception as exc:
            logger.warning(
                "_merge_config_rules: config load failed: %s; "
                "using policy.yaml rules only",
                exc,
            )
            return list(self.detection_rules), dict(self.shell_evasion_checks)

        disabled_ids = set(cfg.disabled_rules or [])

        # Start with policy.yaml detection_rules, filtering disabled
        merged_rules: list[Any] = [
            r for r in self.detection_rules if r.id not in disabled_ids
        ]

        # Append config.json custom_rules (excluding disabled)
        for cr in cfg.custom_rules or []:
            if cr.id not in disabled_ids:
                merged_rules.append(cr)

        # Merge shell evasion checks: config values override policy.yaml
        # defaults on overlapping keys (so the UI can turn checks OFF).
        merged_evasion = dict(self.shell_evasion_checks)
        for check_name, enabled in (cfg.shell_evasion_checks or {}).items():
            merged_evasion[check_name] = enabled

        return merged_rules, merged_evasion

    def _apply_execution_level_fallback(
        self,
        tc_spec: ToolCallSpec,  # pylint: disable=unused-argument
        findings: list[Any],
    ) -> GovernanceDecision:
        """Apply execution_level threshold to determine final decision.

        Reached only when no builtin/user rule matched ("No rule hit"):
        neither an explicit ALLOW/ASK (user_rules) nor an implicit one
        (builtin_rules) fired. The decision then depends on the deep-scan
        findings and the execution_level threshold.

        No findings (nothing matched, nothing flagged):
            - STRICT: ASK (all tools require approval)
            - SMART / AUTO: ALLOW (deep scan cleared it and no protective
              rule objected — asking here only trains users to rubber-stamp
              benign calls; finding-driven approval is restored)
            - OFF: ALLOW (guard effectively disabled)
        Has findings:
            - STRICT: ASK (any finding triggers approval)
            - SMART: INFO/LOW → ALLOW; MEDIUM+ → ASK
            - AUTO / OFF: ASK (a finding surfaced despite the relaxed scan)
        """
        level = self.execution_level

        if not findings:
            # No findings and no rule hit.
            if level == "strict":
                return GovernanceDecision(
                    action=GovernanceAction.ASK,
                    reason="STRICT mode: all tool calls require approval",
                    source="fallback",
                )
            # SMART / AUTO / OFF: the deep scan surfaced nothing and no
            # builtin/user rule objected. Asking the user here produces a
            # flood of low-value prompts, so allow the call (finding-driven
            # approval). Sensitive-path / dangerous-command protection still
            # lives in Phase 1/1.5 and the builtin ASK/DENY rules.
            return GovernanceDecision(
                action=GovernanceAction.ALLOW,
                reason=f"{level.upper()} mode: no findings, no rule hit.",
                source="fallback",
            )

        # Has findings — decide based on execution_level.
        # source is the policy key of the top finding; reason is that
        # finding's own description (no mode shell).
        top = _top_finding(findings)
        finding_reason = getattr(top, "description", "") or top.title
        finding_source = _findings_source(findings)
        max_sev = _max_severity(findings)

        if level == "strict":
            return GovernanceDecision(
                action=GovernanceAction.ASK,
                reason=finding_reason,
                findings=findings,
                source=finding_source,
            )

        if level == "smart":
            if max_sev in ("INFO", "LOW"):
                return GovernanceDecision(
                    action=GovernanceAction.ALLOW,
                    reason=finding_reason,
                    findings=findings,
                    source=finding_source,
                )
            return GovernanceDecision(
                action=GovernanceAction.ASK,
                reason=finding_reason,
                findings=findings,
                source=finding_source,
            )

        # AUTO / OFF: a finding surfaced, so ask.
        return GovernanceDecision(
            action=GovernanceAction.ASK,
            reason=finding_reason,
            findings=findings,
            source=finding_source,
        )

    def evaluate_source(self, tc_spec: ToolCallSpec) -> str:
        """Determine which rule source a tool call matches.

        Returns:
            "builtin_rules"  — matched builtin_rules
            "user_rules"     — matched user_rules
            "fallback" — no match, global fallback
        """
        tool_type = self._registry.get_type(tc_spec.tool_name)
        for rule in self.builtin_rules:
            if rule.matches_tool_call(
                tc_spec,
                tool_type=tool_type,
            ):
                return "builtin_rules"
        for rule in self.user_rules:
            if rule.matches_tool_call(
                tc_spec,
                tool_type=tool_type,
            ):
                return "user_rules"
        return "fallback"

    # ------------------------------------------------------------------
    # Dynamic mutation (only operates on user_rules)
    # ------------------------------------------------------------------

    def add_rule(self, rule: GovernanceRule) -> None:
        """Prepend a rule to user_rules (called after user approval).

        New rules are inserted at the beginning so they take priority
        (first-match-wins evaluation). This ensures a newly added DENY
        can override an earlier ALLOW (e.g. Browser(**) → ALLOW).

        Deduplication:
            If an existing user_rule has the same ``match`` / ``action`` /
            ``grantee`` / ``duration`` / ``session_id`` tuple, it is
            removed first so the new rule still ends up at the head
            (refreshing its priority) without growing the list. This
            avoids unbounded growth from repeated identical approvals.

        Note: builtin_rules are read-only and cannot be modified via
            this method.

        TODO: enforce a hard cap (e.g. 1024 user rules) and an LRU /
            TTL eviction policy so long-running workspaces do not pay
            an ever-growing linear scan in ``evaluate``. The current
            dedup is a best-effort stop-gap.
        """
        # Drop any prior rule with the same identity so we don't keep
        # accumulating duplicates on repeated approvals.
        self.user_rules = [
            r
            for r in self.user_rules
            if not (
                r.match == rule.match
                and r.action == rule.action
                and r.grantee == rule.grantee
                and r.duration == rule.duration
                and r.session_id == rule.session_id
            )
        ]
        self.user_rules.insert(0, rule)

    def remove_rule(self, index: int) -> None:
        """Remove a rule from user_rules (Console UI / admin operation).

        Note: builtin_rules cannot be deleted via this method.
        """
        if 0 <= index < len(self.user_rules):
            self.user_rules.pop(index)


# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------


def _parse_match(match_str: str) -> tuple[str, str]:
    """Parse "ToolName(pattern)" → (tool_name, pattern).

    Uses index to find the first "(", since the tool name never contains "(".
    Uses rindex to find the last ")" to handle cases where pattern
    contains ")".
    Example: "Bash(git *)" → ("Bash", "git *")
             "*(.env*)"   → ("*", ".env*")      # wildcard for all tools
             "Read(src/**)" → ("Read", "src/**")
             "Bash(echo $(date))" → ("Bash", "echo $(date)")
    """
    paren = match_str.index("(")
    close = match_str.rindex(")")
    tool_name = match_str[:paren]
    pattern = match_str[paren + 1 : close]
    return tool_name, pattern


_SEVERITY_ORDER = ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]

# Map a GuardFinding's ``detector`` field to the policy.yaml key that
# drives it. Used so ``GovernanceDecision.source`` reports which policy
# section produced a finding-driven decision, rather than the threat
# category.
_DETECTOR_TO_SOURCE: dict[str, str] = {
    "sensitive_path_detector": "sensitive_paths",
    "pattern_detector": "detection_rules",
    "shell_evasion_detector": "shell_evasion_checks",
}


def _max_severity(findings: list[Any]) -> str:
    """Return the highest severity string from a list of findings."""
    if not findings:
        return "SAFE"
    for sev in _SEVERITY_ORDER:
        if any(getattr(f, "severity", "") == sev for f in findings):
            return sev
    return "SAFE"


def _top_finding(findings: list[Any]) -> Any:
    """Return the highest-severity finding (ties → first encountered)."""
    for sev in _SEVERITY_ORDER:
        for f in findings:
            if getattr(f, "severity", "") == sev:
                return f
    return findings[0]


def _findings_source(findings: list[Any]) -> str:
    """Pick the approval-card ``source`` for a finding-driven decision.

    Returns the policy key of the highest-severity finding, derived from
    its ``detector`` field (e.g. ``"detection_rules"``). Falls back to
    ``"detection_rules"`` when a finding carries no recognisable
    detector, and to ``"builtin_rules"`` when there are no findings.
    """
    if not findings:
        return "builtin_rules"
    top = _top_finding(findings)
    detector = getattr(top, "detector", None)
    if detector:
        return _DETECTOR_TO_SOURCE.get(detector, "detection_rules")
    return "detection_rules"


# ---------------------------------------------------------------------------
# Load / persist
# ---------------------------------------------------------------------------


def load_governance_policy(
    policy_dir: str,
    workspace_dir: str,
    coding_project_dir: str = "",
) -> GovernancePolicy:
    """Load from policy_dir/policy.yaml; return default policy if missing.

    Args:
        policy_dir: directory containing policy.yaml
        workspace_dir: used to replace WORKSPACE_DIR placeholders in rules
        coding_project_dir: used to replace CODING_PROJECT_DIR placeholders
            in rules; defaults to ``workspace_dir`` when empty

    Supports both v1.0 and v2.0 YAML formats.
    """
    path = Path(policy_dir) / "policy.yaml"
    if not path.exists():
        return _create_default_policy(workspace_dir, coding_project_dir)

    try:
        with open(path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f)
    except Exception:
        return _create_default_policy(workspace_dir, coding_project_dir)

    if not isinstance(data, dict):
        return _create_default_policy(workspace_dir, coding_project_dir)

    version = data.get("version", "1.0")
    audit_level = data.get("audit_level", "all")

    # ── builtin_rules: ALWAYS sourced from code (not from YAML) ──
    # Rationale: builtin_rules encode system-level protections (e.g. ASK on
    # ``.env`` / ``.ssh``). Honoring a YAML override here would let a
    # tampered or accidentally truncated policy file silently weaken those
    # protections. Any ``builtin_rules`` key in YAML is intentionally
    # ignored.
    if "builtin_rules" in data:
        logger.warning(
            "load_governance_policy: ignoring 'builtin_rules' in %s; "
            "builtin rules are managed in code and cannot be overridden "
            "via YAML.",
            path,
        )
    builtin_rules = copy.deepcopy(DEFAULT_BUILTIN_RULES)

    # ── user_rules ──
    user_rules = _parse_rules(data.get("user_rules", []))

    # ── env_blacklist ──
    env_blacklist = data.get("env_blacklist", [])
    if not isinstance(env_blacklist, list):
        env_blacklist = []

    # ── Applied default-rule migrations (absent in pre-migration files) ──
    applied_migrations = _parse_applied_migrations(data)

    # ── Cold start / migration: fill in missing default rules ──
    if not user_rules:
        user_rules = copy.deepcopy(get_default_user_rules())
    else:
        user_rules = _merge_missing_default_user_rules(
            user_rules,
            workspace_dir,
            coding_project_dir,
            applied_migrations=applied_migrations,
        )
    # Record all DEFAULT_USER_RULES as applied so the next save
    # makes any user deletion of a default rule stick.
    applied_migrations = sorted(
        set(applied_migrations) | {r.match for r in get_default_user_rules()},
    )
    if not env_blacklist:
        env_blacklist = list(DEFAULT_ENV_BLACKLIST)

    # ── v2.0 fields ──
    execution_level = data.get("execution_level", "smart")
    if execution_level not in ("off", "auto", "smart", "strict"):
        logger.warning(
            "load_governance_policy: invalid execution_level '%s'; "
            "defaulting to 'smart'.",
            execution_level,
        )
        execution_level = "smart"

    sensitive_paths = data.get("sensitive_paths", [])
    if not isinstance(sensitive_paths, list):
        sensitive_paths = []

    shell_evasion_checks = data.get("shell_evasion_checks", {})
    if not isinstance(shell_evasion_checks, dict):
        shell_evasion_checks = {}

    detection_rules = _parse_detection_rules(
        data.get("detection_rules", []),
    )
    # Merge bundled defaults with any user-defined detection rules from
    # policy.yaml.  User rules override defaults when they share an ID.
    detection_rules = _merge_default_detection_rules(detection_rules)

    # ── Replace WORKSPACE_DIR / CODING_PROJECT_DIR with actual paths ──
    cpd = coding_project_dir or workspace_dir
    if workspace_dir:
        _resolve_placeholders(builtin_rules, workspace_dir, cpd)
        _resolve_placeholders(user_rules, workspace_dir, cpd)

    return GovernancePolicy(
        version=version,
        builtin_rules=builtin_rules,
        user_rules=user_rules,
        env_blacklist=env_blacklist,
        audit_level=audit_level,
        execution_level=execution_level,
        sensitive_paths=sensitive_paths,
        shell_evasion_checks=shell_evasion_checks,
        detection_rules=detection_rules,
        applied_migrations=applied_migrations,
    )


def save_governance_policy(
    policy: GovernancePolicy,
    policy_dir: str,
    workspace_dir: str = "",
    coding_project_dir: str = "",
) -> None:
    """Write policy.yaml (v2.0 format).

    Args:
        policy: policy to persist
        policy_dir: directory to write policy.yaml
        workspace_dir: workspace path, used to restore actual paths back to
                       WORKSPACE_DIR placeholders (keeps yaml portable)
        coding_project_dir: coding project path, used to restore actual
                       paths back to CODING_PROJECT_DIR placeholders
    """
    builtin_rules = copy.deepcopy(policy.builtin_rules)
    user_rules = copy.deepcopy(policy.user_rules)

    # ── Restore actual paths to WORKSPACE_DIR / CODING_PROJECT_DIR ──
    if workspace_dir:
        cpd = coding_project_dir or workspace_dir
        _unresolve_placeholders(builtin_rules, workspace_dir, cpd)
        _unresolve_placeholders(user_rules, workspace_dir, cpd)

    path = Path(policy_dir) / "policy.yaml"
    # NOTE: builtin_rules are NOT written to YAML. They are always loaded
    # from DEFAULT_BUILTIN_RULES in code. This prevents a tampered policy
    # file from weakening system-level protections.
    data: dict[str, Any] = {
        "version": policy.version,
        "execution_level": policy.execution_level,
        "audit_level": policy.audit_level,
        "env_blacklist": list(policy.env_blacklist),
        "user_rules": [_rule_to_dict(r) for r in user_rules],
    }

    # v2.0 fields (only write when non-empty to keep YAML clean)
    if policy.applied_migrations:
        data["applied_migrations"] = list(policy.applied_migrations)
    if policy.sensitive_paths:
        data["sensitive_paths"] = list(policy.sensitive_paths)
    if policy.shell_evasion_checks:
        data["shell_evasion_checks"] = dict(policy.shell_evasion_checks)
    # Only persist user-added/overridden detection rules. The bundled
    # defaults from dangerous_shell_commands.yaml are always merged at
    # load time, so writing them here would cause stale rules to survive
    # after an upgrade removes or renames a bundled rule.
    bundled_ids = _get_default_detection_rule_ids()
    user_detection_rules = [
        r for r in policy.detection_rules if r.id not in bundled_ids
    ]
    if user_detection_rules:
        data["detection_rules"] = [
            _detection_rule_to_dict(r) for r in user_detection_rules
        ]

    write_yaml_atomic(
        path,
        data,
        default_flow_style=False,
        allow_unicode=True,
        sort_keys=False,
    )


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


# Default shell evasion checks (migrated from config.json)
_DEFAULT_SHELL_EVASION_CHECKS: dict[str, bool] = {
    "command_substitution": True,
    "obfuscated_flags": True,
    "backslash_escaped_whitespace": True,
    "backslash_escaped_operators": True,
    "newlines": True,
    "comment_quote_desync": True,
    "quoted_newline": True,
}

# Default sensitive paths (migrated from config.json)
_DEFAULT_SENSITIVE_PATHS: List[str] = [
    "~/.ssh/",
    "~/.aws/",
    "~/.gnupg/",
    "~/.kube/",
    "~/.config/gcloud/",
    "~/.azure/",
    "~/.docker/config.json",
    "~/.env",
    "~/.git-credentials",
    "~/.netrc",
    "~/.npmrc",
    "~/.pypirc",
]


def _create_default_policy(
    workspace_dir: str = "",
    coding_project_dir: str = "",
) -> GovernancePolicy:
    """Create a policy with full default rules (cold start, v2.0)."""
    builtin_rules = copy.deepcopy(DEFAULT_BUILTIN_RULES)
    user_rules = copy.deepcopy(get_default_user_rules())
    if workspace_dir:
        cpd = coding_project_dir or workspace_dir
        _resolve_placeholders(builtin_rules, workspace_dir, cpd)
        _resolve_placeholders(user_rules, workspace_dir, cpd)
    return GovernancePolicy(
        version="2.0",
        builtin_rules=builtin_rules,
        user_rules=user_rules,
        env_blacklist=list(DEFAULT_ENV_BLACKLIST),
        execution_level="smart",
        sensitive_paths=list(_DEFAULT_SENSITIVE_PATHS),
        shell_evasion_checks=dict(_DEFAULT_SHELL_EVASION_CHECKS),
        detection_rules=_get_default_detection_rules(),
        applied_migrations=sorted(r.match for r in get_default_user_rules()),
    )


def _parse_applied_migrations(data: dict[str, Any]) -> List[str]:
    """Read the applied default-rule migration markers from policy YAML."""
    applied = data.get("applied_migrations", [])
    if not isinstance(applied, list):
        return []
    return [m for m in applied if isinstance(m, str)]


def _merge_missing_default_user_rules(
    user_rules: List[GovernanceRule],
    workspace_dir: str = "",
    coding_project_dir: str = "",
    applied_migrations: List[str] | None = None,
) -> List[GovernanceRule]:
    """Append default user rules missing from a persisted policy.

    A rule whose match is recorded in ``applied_migrations`` is never
    re-added: the migration already ran once, so a missing rule means
    the user deleted it on purpose.
    """
    applied = set(applied_migrations or [])
    merged = list(user_rules)
    for default_rule in get_default_user_rules():
        if default_rule.match in applied:
            continue
        if _has_equivalent_rule(
            merged,
            default_rule,
            workspace_dir,
            coding_project_dir,
        ):
            continue
        merged.append(copy.deepcopy(default_rule))
    return merged


def _has_equivalent_rule(
    rules: List[GovernanceRule],
    default_rule: GovernanceRule,
    workspace_dir: str = "",
    coding_project_dir: str = "",
) -> bool:
    """Return True when a persisted policy already has this default rule."""
    candidates = {default_rule.match}
    if workspace_dir:
        resolved = copy.deepcopy(default_rule)
        cpd = coding_project_dir or workspace_dir
        _resolve_placeholders([resolved], workspace_dir, cpd)
        candidates.add(resolved.match)
    return any(rule.match in candidates for rule in rules)


def _resolve_placeholders(
    rules: List[GovernanceRule],
    workspace_dir: str,
    coding_project_dir: str = "",
):
    """Replace WORKSPACE_DIR / CODING_PROJECT_DIR placeholders in rules
    with the actual paths (in-place)."""
    for rule in rules:
        if workspace_dir and "WORKSPACE_DIR" in rule.match:
            rule.match = rule.match.replace("WORKSPACE_DIR", workspace_dir)
        if coding_project_dir and "CODING_PROJECT_DIR" in rule.match:
            rule.match = rule.match.replace(
                "CODING_PROJECT_DIR",
                coding_project_dir,
            )


def _unresolve_placeholders(
    rules: List[GovernanceRule],
    workspace_dir: str,
    coding_project_dir: str = "",
):
    """Restore actual paths in rules back to WORKSPACE_DIR /
    CODING_PROJECT_DIR placeholders (in-place).

    When ``coding_project_dir`` coincides with ``workspace_dir`` the two
    cannot be distinguished in an already-resolved pattern, so the shared
    path is restored as ``WORKSPACE_DIR`` (the coding dir is still covered
    by the workspace rules in that case).
    """
    # Build (actual_path, placeholder) pairs, longest path first.
    # avoiding CODING_PROJECT_DIR is substring of WORKSPACE_DIR
    pairs: list[tuple[str, str]] = []
    if coding_project_dir and coding_project_dir != workspace_dir:
        pairs.append((coding_project_dir, "CODING_PROJECT_DIR"))
    if workspace_dir:
        pairs.append((workspace_dir, "WORKSPACE_DIR"))
    pairs.sort(key=lambda pair: len(pair[0]), reverse=True)

    for rule in rules:
        for actual, placeholder in pairs:
            if actual and actual in rule.match:
                rule.match = rule.match.replace(actual, placeholder)


def _parse_rules(
    items: Optional[list[dict[str, Any]]],
) -> List[GovernanceRule]:
    """Parse a list of GovernanceRule from YAML list."""
    if not items:
        return []
    rules = []
    for item in items:
        if not isinstance(item, dict) or "match" not in item:
            continue
        action_str = item.get("action", "deny")
        try:
            action = GovernanceAction(action_str)
        except ValueError:
            action = GovernanceAction.DENY
        rules.append(
            GovernanceRule(
                match=item["match"],
                action=action,
                reason=item.get("reason", ""),
                grantee=item.get("grantee", "*"),
                duration=item.get("duration", "permanent"),
                session_id=item.get("session_id"),
            ),
        )
    return rules


def _rule_to_dict(rule: GovernanceRule) -> dict[str, Any]:
    """Serialize a GovernanceRule to dict (for YAML output)."""
    d = {
        "match": rule.match,
        "action": rule.action.value,
    }
    if rule.reason:
        d["reason"] = rule.reason
    if rule.grantee != "*":
        d["grantee"] = rule.grantee
    if rule.duration != "permanent":
        d["duration"] = rule.duration
    if rule.session_id:
        d["session_id"] = rule.session_id
    return d


def _detection_rule_to_dict(rule: DetectionRuleConfig) -> dict[str, Any]:
    """Serialize a DetectionRuleConfig to dict (for YAML output)."""
    d: dict[str, Any] = {"id": rule.id}
    if rule.tools:
        d["tools"] = rule.tools
    if rule.params:
        d["params"] = rule.params
    if rule.category != "command_injection":
        d["category"] = rule.category
    d["severity"] = rule.severity
    if rule.patterns:
        d["patterns"] = rule.patterns
    if rule.exclude_patterns:
        d["exclude_patterns"] = rule.exclude_patterns
    if rule.description:
        d["description"] = rule.description
    if rule.remediation:
        d["remediation"] = rule.remediation
    return d


_DEFAULT_DETECTION_RULES_CACHE: List[DetectionRuleConfig] | None = None
_DEFAULT_DETECTION_RULES_LOCK = threading.Lock()


def _guard_rule_to_detection_config(guard_rule: Any) -> DetectionRuleConfig:
    """Convert a ToolGuard ``GuardRule`` into
    a governance ``DetectionRuleConfig``."""
    return DetectionRuleConfig(
        id=guard_rule.id,
        tools=list(guard_rule.tools),
        params=list(guard_rule.params),
        category=str(guard_rule.category.value)
        if hasattr(guard_rule.category, "value")
        else str(guard_rule.category),
        severity=str(guard_rule.severity.value)
        if hasattr(guard_rule.severity, "value")
        else str(guard_rule.severity),
        patterns=list(guard_rule.patterns),
        exclude_patterns=list(guard_rule.exclude_patterns),
        description=guard_rule.description,
        remediation=guard_rule.remediation,
    )


def _get_default_detection_rules() -> List[DetectionRuleConfig]:
    """Load bundled detection rules via the ToolGuard rule loader.

    Reuses ``load_rules_from_yaml`` from the ToolGuard subsystem to avoid
    duplicating YAML I/O and validation logic.  The loaded ``GuardRule``
    objects are converted into governance ``DetectionRuleConfig`` instances.

    The result is cached (thread-safe) and returned as a deep copy so that
    callers cannot mutate shared state.  On any failure an empty list is
    returned (graceful degradation — Phase 2 builtin_rules still provide
    baseline protection).
    """
    global _DEFAULT_DETECTION_RULES_CACHE  # noqa: PLW0603
    with _DEFAULT_DETECTION_RULES_LOCK:
        if _DEFAULT_DETECTION_RULES_CACHE is not None:
            return copy.deepcopy(_DEFAULT_DETECTION_RULES_CACHE)

        try:
            from ..security.tool_guard.guardians.rule_guardian import (
                load_rules_from_yaml,
                _DEFAULT_RULES_DIR,
            )

            yaml_path = _DEFAULT_RULES_DIR / "dangerous_shell_commands.yaml"
            guard_rules = load_rules_from_yaml(yaml_path)
            rules = [_guard_rule_to_detection_config(r) for r in guard_rules]
            _DEFAULT_DETECTION_RULES_CACHE = rules
            return copy.deepcopy(rules)
        except Exception as exc:
            logger.warning(
                "_get_default_detection_rules: failed to load: %s; "
                "detection_rules will be empty",
                exc,
            )
            _DEFAULT_DETECTION_RULES_CACHE = []
            return []


def _get_default_detection_rule_ids() -> frozenset[str]:
    """Return the set of IDs that belong to bundled default rules."""
    return frozenset(r.id for r in _get_default_detection_rules())


def _merge_default_detection_rules(
    user_rules: List[DetectionRuleConfig],
) -> List[DetectionRuleConfig]:
    """Merge bundled defaults with user-defined detection rules.

    Bundled rules form the base.  If the user defines a rule with the
    same ID in policy.yaml, the user version wins (override).  User
    rules with new IDs are appended after the defaults.
    """
    base_rules = _get_default_detection_rules()
    # Build ordered map: defaults first, then user overrides
    rules_by_id: dict[str, DetectionRuleConfig] = {r.id: r for r in base_rules}
    for r in user_rules:
        rules_by_id[r.id] = r  # user version overrides default
    return list(rules_by_id.values())


def _parse_detection_rules(
    items: list[dict[str, Any]] | None,
) -> List[DetectionRuleConfig]:
    """Parse detection_rules from YAML into DetectionRuleConfig."""
    if not items:
        return []
    rules: List[DetectionRuleConfig] = []
    for item in items:
        if not isinstance(item, dict) or "id" not in item:
            continue
        try:
            rules.append(
                DetectionRuleConfig(
                    id=item["id"],
                    tools=item.get("tools", []),
                    params=item.get("params", []),
                    category=item.get("category", "command_injection"),
                    severity=item.get("severity", "HIGH"),
                    patterns=item.get("patterns", []),
                    exclude_patterns=item.get("exclude_patterns", []),
                    description=item.get("description", ""),
                    remediation=item.get("remediation", ""),
                ),
            )
        except Exception as exc:
            logger.warning(
                "Skipping invalid detection rule '%s': %s",
                item.get("id", "?"),
                exc,
            )
    return rules
