# consent-protocol/hushh_mcp/consent/scope_generator.py
"""
Dynamic Scope Generator - Generates and validates consent scopes dynamically.

Scopes support nested paths:
- attr.{domain}.{attribute_key}
- attr.{domain}.*
- attr.{domain}.{subintent}.*
"""

import json
import logging
from typing import Optional

from db.db_client import get_db

logger = logging.getLogger(__name__)


class DynamicScopeGenerator:
    """
    Generates and validates consent scopes dynamically based on stored attributes.

    Scope Format:
    - Specific: attr.{domain}.{attribute_key}
    - Wildcard: attr.{domain}.*
    - Domain-level: attr.{domain}

    Examples:
    - attr.financial.holdings
    - attr.subscriptions.netflix_plan
    - attr.health.*
    """

    SCOPE_PREFIX = "attr."
    WILDCARD_SUFFIX = ".*"
    _STRUCTURAL_TOP_LEVEL_SCOPE_PATHS = {
        "domain_intent",
        "schema_version",
        "updated_at",
    }
    _INTERNAL_ONLY_DOMAINS = {
        "kyc_connector",
        "kyc_workflow",
    }
    _VISIBILITY_POSTURES = {"private", "consent_required", "default_available"}

    def __init__(self):
        self._supabase = None
        self._scope_cache: dict[str, set[str]] = {}  # user_id -> set of scopes
        self._cache_ttl = 300  # 5 minutes

    @property
    def supabase(self):
        if self._supabase is None:
            self._supabase = get_db()
        return self._supabase

    def generate_scope(self, domain: str, attribute_key: str) -> str:
        """
        Generate a scope string for a specific attribute.

        Args:
            domain: The domain key (e.g., 'financial')
            attribute_key: The attribute key (e.g., 'holdings')

        Returns:
            Scope string (e.g., 'attr.financial.holdings')
        """
        domain = domain.lower().strip()
        attribute_key = attribute_key.lower().strip()
        return f"{self.SCOPE_PREFIX}{domain}.{attribute_key}"

    def generate_domain_wildcard(self, domain: str) -> str:
        """
        Generate a wildcard scope for an entire domain.

        Args:
            domain: The domain key (e.g., 'financial')

        Returns:
            Wildcard scope string (e.g., 'attr.financial.*')
        """
        domain = domain.lower().strip()
        return f"{self.SCOPE_PREFIX}{domain}{self.WILDCARD_SUFFIX}"

    def parse_scope(self, scope: str) -> tuple[Optional[str], Optional[str], bool]:
        """
        Parse a scope string into its components.

        Args:
            scope: The scope string to parse

        Returns:
            Tuple of (domain, attribute_key, is_wildcard)
            Returns (None, None, False) if invalid format
        """
        if not scope.startswith(self.SCOPE_PREFIX):
            return (None, None, False)

        remainder = scope[len(self.SCOPE_PREFIX) :].strip()
        if not remainder:
            return (None, None, False)

        parts = [part for part in remainder.split(".") if part]
        if not parts:
            return (None, None, False)

        domain = self._normalize_domain_key(parts[0])
        if not domain:
            return (None, None, False)

        if len(parts) == 1:
            # Domain-level scope (e.g., attr.financial)
            return (domain, None, False)

        if parts[-1] == "*":
            if len(parts) == 2:
                # Domain wildcard (e.g., attr.financial.*)
                return (domain, None, True)
            # Subintent/attribute wildcard (e.g., attr.financial.profile.*)
            path = self._normalize_scope_path(".".join(parts[1:-1]))
            return (domain, path or None, True)

        # Specific path (e.g., attr.financial.profile.risk_score)
        path = self._normalize_scope_path(".".join(parts[1:]))
        return (domain, path or None, False)

    def is_dynamic_scope(self, scope: str) -> bool:
        """Check if a scope is a dynamic attr.* scope."""
        return scope.startswith(self.SCOPE_PREFIX)

    @staticmethod
    def _normalize_domain_key(domain: str | None) -> str:
        return str(domain or "").strip().lower()

    @staticmethod
    def _normalize_scope_path(path: str | None) -> str:
        if not isinstance(path, str):
            return ""
        raw = path.strip().lower()
        if not raw:
            return ""
        segments: list[str] = []
        for part in raw.split("."):
            normalized_part = "".join(
                ch if (ch.isalnum() or ch == "_") else "_" for ch in part.strip()
            ).strip("_")
            if normalized_part:
                segments.append(normalized_part)
        return ".".join(segments)

    @classmethod
    def _is_internal_only_domain(cls, domain: str | None) -> bool:
        return cls._normalize_domain_key(domain) in cls._INTERNAL_ONLY_DOMAINS

    @staticmethod
    def _coerce_json_dict(value: object) -> dict:
        if isinstance(value, dict):
            return value
        if isinstance(value, str):
            text = value.strip()
            if not text:
                return {}
            try:
                parsed = json.loads(text)
            except Exception:
                return {}
            return parsed if isinstance(parsed, dict) else {}
        return {}

    @classmethod
    def _normalize_domains(cls, domains: list[str] | None) -> list[str]:
        if not domains:
            return []
        return sorted(
            {
                cls._normalize_domain_key(domain)
                for domain in domains
                if cls._normalize_domain_key(domain)
            }
        )

    @classmethod
    def _scope_visibility_metadata(
        cls,
        *,
        top_level_path: str | None,
        summary_projection: dict | None = None,
        visibility_posture: str | None = None,
        exposure_enabled: bool = True,
        default_projection_ready: bool = False,
        default_projection_updated_at: str | None = None,
    ) -> dict[str, object]:
        projection = dict(summary_projection or {})
        normalized_path = cls._normalize_scope_path(
            projection.get("top_level_scope_path") or top_level_path
        )
        consumer_visible = (
            bool(normalized_path) and normalized_path not in cls._STRUCTURAL_TOP_LEVEL_SCOPE_PATHS
        )
        requested_posture = str(visibility_posture or "").strip().lower()
        if requested_posture not in cls._VISIBILITY_POSTURES:
            requested_posture = "consent_required"
        if not exposure_enabled or not consumer_visible:
            requested_posture = "private"
        return {
            "top_level_scope_path": normalized_path,
            "consumer_visible": bool(
                projection.get("consumer_visible")
                if isinstance(projection.get("consumer_visible"), bool)
                else consumer_visible
            ),
            "internal_only": bool(
                projection.get("internal_only")
                if isinstance(projection.get("internal_only"), bool)
                else not consumer_visible
            ),
            "visibility_reason": str(
                projection.get("visibility_reason")
                or ("consumer_shareable" if consumer_visible else "structural_top_level_path")
            ).strip(),
            "visibility_posture": requested_posture,
            "default_projection_ready": bool(
                requested_posture == "default_available" and default_projection_ready
            ),
            "default_projection_updated_at": default_projection_updated_at
            if requested_posture == "default_available" and default_projection_ready
            else None,
        }

    async def _get_legacy_scope_catalog(self, user_id: str) -> dict[str, dict[str, set[str]]]:
        result = (
            self.supabase.table("pkm_index")
            .select("available_domains", "domain_summaries")
            .eq("user_id", user_id)
            .limit(1)
            .execute()
        )
        if not result.data:
            return {}

        row = result.data[0]
        available_domains = self._normalize_domains(row.get("available_domains") or [])
        domain_summaries = row.get("domain_summaries")
        if not isinstance(domain_summaries, dict):
            domain_summaries = {}

        catalog: dict[str, dict[str, set[str]]] = {
            domain: {"paths": set(), "wildcards": set()}
            for domain in available_domains
            if not self._is_internal_only_domain(domain)
        }

        for domain in available_domains:
            if self._is_internal_only_domain(domain):
                continue
            summary = domain_summaries.get(domain)
            if not isinstance(summary, dict):
                continue
            for key in (
                "intent_map",
                "sub_intents",
                "subintents",
                "available_subintents",
                "available_sub_intents",
            ):
                raw_value = summary.get(key)
                if isinstance(raw_value, list):
                    for item in raw_value:
                        normalized = self._normalize_scope_path(str(item))
                        if normalized:
                            catalog[domain]["paths"].add(normalized)
                            catalog[domain]["wildcards"].add(normalized)

        return catalog

    async def _get_user_scope_catalog(self, user_id: str) -> dict[str, dict[str, set[str]]]:
        manifest_rows = (
            self.supabase.table("pkm_manifests")
            .select("domain,top_level_scope_paths,externalizable_paths")
            .eq("user_id", user_id)
            .execute()
        )
        path_rows = (
            self.supabase.table("pkm_manifest_paths")
            .select("domain,json_path,path_type,exposure_eligibility")
            .eq("user_id", user_id)
            .execute()
        )

        catalog: dict[str, dict[str, set[str]]] = {}
        for row in manifest_rows.data or []:
            if not isinstance(row, dict):
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            if not domain or self._is_internal_only_domain(domain):
                continue
            entry = catalog.setdefault(domain, {"paths": set(), "wildcards": set()})
            for top_level_path in row.get("top_level_scope_paths") or []:
                normalized = self._normalize_scope_path(str(top_level_path))
                if normalized:
                    entry["paths"].add(normalized)
                    entry["wildcards"].add(normalized)
            for externalizable_path in row.get("externalizable_paths") or []:
                normalized = self._normalize_scope_path(str(externalizable_path))
                if normalized:
                    entry["paths"].add(normalized)

        for row in path_rows.data or []:
            if not isinstance(row, dict):
                continue
            if row.get("exposure_eligibility") is False:
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            json_path = self._normalize_scope_path(row.get("json_path"))
            if not domain or self._is_internal_only_domain(domain) or not json_path:
                continue
            entry = catalog.setdefault(domain, {"paths": set(), "wildcards": set()})
            entry["paths"].add(json_path)
            if row.get("path_type") in {"object", "array"}:
                entry["wildcards"].add(json_path)

        if catalog:
            return catalog

        return await self._get_legacy_scope_catalog(user_id)

    async def get_available_scope_entries(self, user_id: str) -> list[dict]:
        """
        Get manifest-backed scope discovery entries with provenance metadata.

        Each entry describes one requestable scope string and why it exists.
        """
        try:
            index_result = (
                self.supabase.table("pkm_index")
                .select("available_domains")
                .eq("user_id", user_id)
                .limit(1)
                .execute()
            )
            manifest_result = (
                self.supabase.table("pkm_manifests")
                .select("domain,top_level_scope_paths,externalizable_paths,manifest_version")
                .eq("user_id", user_id)
                .execute()
            )
            path_result = (
                self.supabase.table("pkm_manifest_paths")
                .select(
                    "domain,json_path,path_type,exposure_eligibility,consent_label,scope_handle"
                )
                .eq("user_id", user_id)
                .execute()
            )
            registry_result = (
                self.supabase.table("pkm_scope_registry")
                .select(
                    "domain,scope_handle,scope_label,exposure_enabled,visibility_posture,default_projection_ready,default_projection_updated_at,summary_projection,manifest_version"
                )
                .eq("user_id", user_id)
                .execute()
            )
        except Exception:
            logger.error("scope_generator.get_scope_entries_failed")
            return []

        def _source_rank(kind: str) -> int:
            return {
                "pkm_index": 1,
                "pkm_manifests.top_level_scope_paths": 2,
                "pkm_scope_registry": 3,
                "pkm_manifests.externalizable_paths": 4,
                "pkm_manifest_paths": 5,
                "legacy_metadata_fallback": 0,
            }.get(kind, 0)

        entries: dict[str, dict[str, object]] = {}

        def _upsert_scope_entry(entry: dict[str, object]) -> None:
            scope = str(entry.get("scope") or "").strip()
            if not scope:
                return
            current = entries.get(scope)
            if current is None:
                entries[scope] = entry
                return
            if _source_rank(str(entry.get("source_kind") or "")) >= _source_rank(
                str(current.get("source_kind") or "")
            ):
                merged = {**current, **entry}
            else:
                merged = {**entry, **current}
            for key in (
                "registry_handle",
                "label",
                "meta_reference",
                "path",
                "domain",
                "manifest_revision",
            ):
                if not merged.get(key):
                    merged[key] = current.get(key) or entry.get(key)
            merged["wildcard"] = bool(current.get("wildcard") or entry.get("wildcard"))
            merged["exposure_eligibility"] = bool(
                current.get("exposure_eligibility") or entry.get("exposure_eligibility")
            )
            entries[scope] = merged

        index_domains = self._normalize_domains(
            (index_result.data or [{}])[0].get("available_domains") if index_result.data else []
        )
        manifest_rows = manifest_result.data or []
        path_rows = path_result.data or []
        registry_rows = registry_result.data or []

        registry_by_top_level: dict[tuple[str, str], dict[str, object]] = {}
        enabled_consumer_top_levels_by_domain: dict[str, set[str]] = {}
        all_consumer_top_levels_by_domain: dict[str, set[str]] = {}
        known_domains = set(index_domains)
        for row in manifest_rows:
            if not isinstance(row, dict):
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            if domain:
                known_domains.add(domain)
        for row in path_rows:
            if not isinstance(row, dict):
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            if domain:
                known_domains.add(domain)
        for row in registry_rows:
            if not isinstance(row, dict):
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            if domain:
                known_domains.add(domain)
            summary_projection = self._coerce_json_dict(row.get("summary_projection"))
            visibility = self._scope_visibility_metadata(
                top_level_path=summary_projection.get("top_level_scope_path"),
                summary_projection=summary_projection,
                visibility_posture=row.get("visibility_posture"),
                exposure_enabled=row.get("exposure_enabled") is not False,
                default_projection_ready=row.get("default_projection_ready") is True,
                default_projection_updated_at=str(row.get("default_projection_updated_at") or "")
                or None,
            )
            top_level_path = self._normalize_scope_path(visibility.get("top_level_scope_path"))
            if domain and top_level_path:
                if (
                    visibility.get("consumer_visible") is not False
                    and visibility.get("internal_only") is not True
                    and visibility.get("visibility_posture") != "private"
                ):
                    all_consumer_top_levels_by_domain.setdefault(domain, set()).add(top_level_path)
                    if visibility.get("visibility_posture") != "private":
                        enabled_consumer_top_levels_by_domain.setdefault(domain, set()).add(
                            top_level_path
                        )
                registry_by_top_level[(domain, top_level_path)] = {
                    "registry_handle": str(row.get("scope_handle") or "").strip() or None,
                    "label": str(row.get("scope_label") or "").strip() or None,
                    "manifest_revision": row.get("manifest_version"),
                    "source_kind": "pkm_scope_registry",
                    "exposure_enabled": row.get("exposure_enabled") is not False,
                    "visibility_posture": visibility.get("visibility_posture"),
                    "default_projection_ready": visibility.get("default_projection_ready") is True,
                    "default_projection_updated_at": visibility.get(
                        "default_projection_updated_at"
                    ),
                    "consumer_visible": visibility.get("consumer_visible") is not False,
                    "internal_only": visibility.get("internal_only") is True,
                    "visibility_reason": visibility.get("visibility_reason"),
                    "top_level_scope_path": top_level_path,
                }

        for domain in sorted(known_domains):
            domain_internal = self._is_internal_only_domain(domain)
            domain_top_levels = all_consumer_top_levels_by_domain.get(domain, set())
            enabled_top_levels = enabled_consumer_top_levels_by_domain.get(domain, set())
            if (
                not domain_internal
                and domain_top_levels
                and enabled_top_levels != domain_top_levels
            ):
                continue
            _upsert_scope_entry(
                {
                    "scope": self.generate_domain_wildcard(domain),
                    "domain": domain,
                    "path": None,
                    "wildcard": True,
                    "source_kind": "pkm_index",
                    "registry_handle": None,
                    "label": f"{domain.replace('_', ' ').title()} Domain",
                    "exposure_eligibility": True,
                    "manifest_revision": None,
                    "meta_reference": "domain wildcard derived from discovered PKM domains",
                    "consumer_visible": not domain_internal,
                    "internal_only": domain_internal,
                    "visibility_reason": "internal_runtime_domain"
                    if domain_internal
                    else "consumer_shareable",
                    "visibility_posture": "private" if domain_internal else "consent_required",
                    "default_projection_ready": False,
                    "default_projection_updated_at": None,
                }
            )

        manifest_externalizable_paths: set[tuple[str, str]] = set()
        for row in manifest_rows:
            if not isinstance(row, dict):
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            if not domain:
                continue
            domain_internal = self._is_internal_only_domain(domain)
            manifest_version = row.get("manifest_version")
            top_level_paths = [
                self._normalize_scope_path(path)
                for path in (row.get("top_level_scope_paths") or [])
            ]
            for path in [path for path in top_level_paths if path]:
                registry_meta = registry_by_top_level.get((domain, path), {})
                if registry_meta.get("visibility_posture") == "private":
                    continue
                _upsert_scope_entry(
                    {
                        "scope": f"{self.SCOPE_PREFIX}{domain}.{path}{self.WILDCARD_SUFFIX}",
                        "domain": domain,
                        "path": path,
                        "wildcard": True,
                        "source_kind": "pkm_manifests.top_level_scope_paths",
                        "registry_handle": registry_meta.get("registry_handle"),
                        "label": registry_meta.get("label") or path.replace("_", " ").title(),
                        "exposure_eligibility": True,
                        "manifest_revision": registry_meta.get("manifest_revision")
                        or manifest_version,
                        "meta_reference": "manifest top-level scope path",
                        "consumer_visible": False
                        if domain_internal
                        else registry_meta.get("consumer_visible") is not False,
                        "internal_only": True
                        if domain_internal
                        else registry_meta.get("internal_only") is True,
                        "visibility_reason": "internal_runtime_domain"
                        if domain_internal
                        else registry_meta.get("visibility_reason"),
                        "visibility_posture": "private"
                        if domain_internal
                        else registry_meta.get("visibility_posture") or "consent_required",
                        "default_projection_ready": registry_meta.get("default_projection_ready")
                        is True,
                        "default_projection_updated_at": registry_meta.get(
                            "default_projection_updated_at"
                        ),
                    }
                )
            for raw_path in row.get("externalizable_paths") or []:
                path = self._normalize_scope_path(raw_path)
                if not path:
                    continue
                manifest_externalizable_paths.add((domain, path))
                top_level = path.split(".", 1)[0]
                registry_meta = registry_by_top_level.get((domain, top_level), {})
                if registry_meta.get("visibility_posture") == "private":
                    continue
                _upsert_scope_entry(
                    {
                        "scope": self.generate_scope(domain, path),
                        "domain": domain,
                        "path": path,
                        "wildcard": False,
                        "source_kind": "pkm_manifests.externalizable_paths",
                        "registry_handle": registry_meta.get("registry_handle"),
                        "label": path.replace("_", " ").replace(".", " ").title(),
                        "exposure_eligibility": True,
                        "manifest_revision": registry_meta.get("manifest_revision")
                        or manifest_version,
                        "meta_reference": "externalizable manifest path",
                        "consumer_visible": False
                        if domain_internal
                        else registry_meta.get("consumer_visible") is not False,
                        "internal_only": True
                        if domain_internal
                        else registry_meta.get("internal_only") is True,
                        "visibility_reason": "internal_runtime_domain"
                        if domain_internal
                        else registry_meta.get("visibility_reason"),
                        "visibility_posture": "private"
                        if domain_internal
                        else registry_meta.get("visibility_posture") or "consent_required",
                        "default_projection_ready": registry_meta.get("default_projection_ready")
                        is True,
                        "default_projection_updated_at": registry_meta.get(
                            "default_projection_updated_at"
                        ),
                    }
                )

        for row in path_rows:
            if not isinstance(row, dict):
                continue
            if row.get("exposure_eligibility") is False:
                continue
            domain = self._normalize_domain_key(row.get("domain"))
            path = self._normalize_scope_path(row.get("json_path"))
            if not domain or not path:
                continue
            domain_internal = self._is_internal_only_domain(domain)
            top_level = path.split(".", 1)[0]
            registry_meta = registry_by_top_level.get((domain, top_level), {})
            if registry_meta.get("visibility_posture") == "private":
                continue
            _upsert_scope_entry(
                {
                    "scope": self.generate_scope(domain, path),
                    "domain": domain,
                    "path": path,
                    "wildcard": False,
                    "source_kind": "pkm_manifest_paths",
                    "registry_handle": str(row.get("scope_handle") or "").strip()
                    or registry_meta.get("registry_handle"),
                    "label": str(row.get("consent_label") or "").strip()
                    or registry_meta.get("label")
                    or path.replace("_", " ").replace(".", " ").title(),
                    "exposure_eligibility": True,
                    "manifest_revision": registry_meta.get("manifest_revision"),
                    "meta_reference": "manifest path row marked exposure eligible",
                    "consumer_visible": False
                    if domain_internal
                    else registry_meta.get("consumer_visible") is not False,
                    "internal_only": True
                    if domain_internal
                    else registry_meta.get("internal_only") is True,
                    "visibility_reason": "internal_runtime_domain"
                    if domain_internal
                    else registry_meta.get("visibility_reason"),
                    "visibility_posture": "private"
                    if domain_internal
                    else registry_meta.get("visibility_posture") or "consent_required",
                    "default_projection_ready": registry_meta.get("default_projection_ready")
                    is True,
                    "default_projection_updated_at": registry_meta.get(
                        "default_projection_updated_at"
                    ),
                }
            )

        if entries:
            return [entries[scope] for scope in sorted(entries)]

        legacy_catalog = await self._get_legacy_scope_catalog(user_id)
        for domain, entry in legacy_catalog.items():
            _upsert_scope_entry(
                {
                    "scope": self.generate_domain_wildcard(domain),
                    "domain": domain,
                    "path": None,
                    "wildcard": True,
                    "source_kind": "legacy_metadata_fallback",
                    "registry_handle": None,
                    "label": f"{domain.replace('_', ' ').title()} Domain",
                    "exposure_eligibility": True,
                    "manifest_revision": None,
                    "meta_reference": "legacy metadata fallback domain wildcard",
                    "consumer_visible": True,
                    "internal_only": False,
                    "visibility_reason": "consumer_shareable",
                    "visibility_posture": "consent_required",
                    "default_projection_ready": False,
                    "default_projection_updated_at": None,
                }
            )
            for path in sorted(entry.get("wildcards", set())):
                _upsert_scope_entry(
                    {
                        "scope": f"{self.SCOPE_PREFIX}{domain}.{path}{self.WILDCARD_SUFFIX}",
                        "domain": domain,
                        "path": path,
                        "wildcard": True,
                        "source_kind": "legacy_metadata_fallback",
                        "registry_handle": None,
                        "label": path.replace("_", " ").replace(".", " ").title(),
                        "exposure_eligibility": True,
                        "manifest_revision": None,
                        "meta_reference": "legacy metadata fallback wildcard path",
                        "consumer_visible": True,
                        "internal_only": False,
                        "visibility_reason": "consumer_shareable",
                        "visibility_posture": "consent_required",
                        "default_projection_ready": False,
                        "default_projection_updated_at": None,
                    }
                )
            for path in sorted(entry.get("paths", set())):
                _upsert_scope_entry(
                    {
                        "scope": self.generate_scope(domain, path),
                        "domain": domain,
                        "path": path,
                        "wildcard": False,
                        "source_kind": "legacy_metadata_fallback",
                        "registry_handle": None,
                        "label": path.replace("_", " ").replace(".", " ").title(),
                        "exposure_eligibility": True,
                        "manifest_revision": None,
                        "meta_reference": "legacy metadata fallback exact path",
                        "consumer_visible": True,
                        "internal_only": False,
                        "visibility_reason": "consumer_shareable",
                        "visibility_posture": "consent_required",
                        "default_projection_ready": False,
                        "default_projection_updated_at": None,
                    }
                )
        return [entries[scope] for scope in sorted(entries)]

    def matches_wildcard(self, scope: str, wildcard: str) -> bool:
        """
        Check if a specific scope matches a wildcard pattern.

        Args:
            scope: The specific scope (e.g., 'attr.financial.holdings')
            wildcard: The wildcard pattern (e.g., 'attr.financial.*')

        Returns:
            True if the scope matches the wildcard
        """
        granted_domain, granted_path, granted_wildcard = self.parse_scope(wildcard)
        requested_domain, requested_path, _requested_wildcard = self.parse_scope(scope)

        if granted_domain is None or requested_domain is None:
            return scope == wildcard
        if granted_domain != requested_domain:
            return False

        if not granted_wildcard:
            return scope == wildcard

        # attr.{domain}.* grants everything under that domain.
        if granted_path is None:
            return True

        # attr.{domain}.{subintent}.* grants everything under that subintent path.
        if requested_path is None:
            return False
        return requested_path == granted_path or requested_path.startswith(f"{granted_path}.")

    async def validate_scope(self, scope: str, user_id: Optional[str] = None) -> bool:
        """
        Validate that a scope is valid.

        Validates against PKM manifests, manifest paths, and index-backed discovery.

        Args:
            scope: The scope to validate
            user_id: Optional user ID to check against stored data

        Returns:
            True if the scope is valid
        """
        domain, _attribute_key, _is_wildcard = self.parse_scope(scope)

        if domain is None:
            return False
        domain = self._normalize_domain_key(domain)
        if not domain:
            return False

        # If no user_id, just validate format
        if user_id is None:
            return True

        try:
            scope_catalog = await self._get_user_scope_catalog(user_id)
            if not scope_catalog:
                logger.debug("scope_generator.no_pkm_index (user=[redacted])")
                return False

            domain_catalog = scope_catalog.get(domain)
            if domain_catalog is None:
                return False
            domain_paths = domain_catalog.get("paths", set())
            domain_wildcards = domain_catalog.get("wildcards", set())

            # Domain-level scope is valid when domain exists.
            if _attribute_key is None:
                return True

            candidate_path = self._normalize_scope_path(_attribute_key)
            if not candidate_path:
                return False

            if _is_wildcard:
                if candidate_path in domain_wildcards:
                    return True
                return any(
                    path == candidate_path or path.startswith(f"{candidate_path}.")
                    for path in domain_paths
                )

            return candidate_path in domain_paths or candidate_path in domain_wildcards
        except Exception as e:
            logger.error(f"Error validating scope {scope}: {e}")
            return False

    async def get_available_scopes(
        self,
        user_id: str,
        *,
        include_internal: bool = False,
        include_exact_paths: bool = False,
    ) -> list[str]:
        """
        Get all valid consent scopes for a user from manifest-backed discovery.

        Args:
            user_id: The user ID

        Returns:
            List of exact and wildcard scope strings
        """
        try:
            scopes: set[str] = {"pkm.read"}
            for entry in await self.get_available_scope_entries(user_id):
                if not include_internal and (
                    entry.get("internal_only") is True or entry.get("consumer_visible") is False
                ):
                    continue
                if not include_internal and entry.get("visibility_posture") == "private":
                    continue
                if not include_exact_paths and entry.get("wildcard") is not True:
                    continue
                scope = str(entry.get("scope") or "").strip()
                if scope:
                    scopes.add(scope)
            return sorted(scopes)
        except Exception as e:
            logger.error("scope_generator.get_scopes_failed (user=[redacted]): %s", e)
            return []

    async def get_available_wildcards(
        self, user_id: str, *, include_internal: bool = False
    ) -> list[str]:
        """
        Get all valid wildcard scopes for a user from PKM index metadata.

        Args:
            user_id: The user ID

        Returns:
            List of wildcard scope strings
        """
        scopes = await self.get_available_scopes(
            user_id,
            include_internal=include_internal,
            include_exact_paths=False,
        )
        return sorted(
            scope for scope in scopes if scope == "pkm.read" or scope.endswith(self.WILDCARD_SUFFIX)
        )

    async def check_scope_access(
        self,
        requested_scope: str,
        granted_scopes: list[str],
        user_id: Optional[str] = None,
    ) -> bool:
        """
        Check if a requested scope is covered by granted scopes.

        Args:
            requested_scope: The scope being requested
            granted_scopes: List of scopes that have been granted
            user_id: Optional user ID for validation

        Returns:
            True if access should be granted
        """
        # Direct match
        if requested_scope in granted_scopes:
            return True

        # Check wildcard matches
        for granted in granted_scopes:
            if self.matches_wildcard(requested_scope, granted):
                return True

        # Check if vault.owner is granted (full access)
        if "vault.owner" in granted_scopes:
            return True

        return False

    async def expand_wildcard(self, wildcard: str, user_id: str) -> list[str]:
        """
        Expand a wildcard scope into specific scopes for a user.

        Args:
            wildcard: The wildcard scope (e.g., 'attr.financial.*')
            user_id: The user ID (unused; kept for API stability)

        Returns:
            List of matching exact scopes
        """
        domain, _, is_wildcard = self.parse_scope(wildcard)
        if not is_wildcard or domain is None:
            return [wildcard]
        scope_catalog = await self._get_user_scope_catalog(user_id)
        domain_catalog = scope_catalog.get(domain)
        if not domain_catalog:
            return [wildcard]
        _, wildcard_path, _ = self.parse_scope(wildcard)
        matched = []
        for path in sorted(domain_catalog.get("paths", set())):
            if (
                wildcard_path is None
                or path == wildcard_path
                or path.startswith(f"{wildcard_path}.")
            ):
                matched.append(self.generate_scope(domain, path))
        return matched or [wildcard]

    def get_scope_display_info(self, scope: str) -> dict:
        """
        Get rich display information for a scope, including icon/color from domain contracts.

        Args:
            scope: The scope string

        Returns:
            Dict with display_name, domain, attribute, is_wildcard, icon_name, color_hex, description
        """
        from hushh_mcp.services.domain_contracts import get_canonical_domain_metadata

        domain, attribute_key, is_wildcard = self.parse_scope(scope)

        if domain is None:
            return {
                "display_name": scope,
                "domain": None,
                "attribute": None,
                "is_wildcard": False,
                "icon_name": None,
                "color_hex": None,
                "description": None,
            }

        # Pull rich metadata from domain contracts
        metadata = get_canonical_domain_metadata(domain)
        icon_name = metadata.icon_name if metadata else None
        color_hex = metadata.color_hex if metadata else None
        domain_display = metadata.display_name if metadata else domain.title()
        domain_description = metadata.description if metadata else None

        if is_wildcard:
            display_name = f"All {domain_display} Data"
            description = domain_description
        elif attribute_key:
            attr_display = attribute_key.replace("_", " ").title()
            display_name = f"{domain_display} — {attr_display}"
            description = f"{attr_display} within {domain_display}"
        else:
            display_name = domain_display
            description = domain_description

        return {
            "display_name": display_name,
            "domain": domain,
            "attribute": attribute_key,
            "is_wildcard": is_wildcard,
            "icon_name": icon_name,
            "color_hex": color_hex,
            "description": description,
        }

    async def prewarm_validator(self) -> None:
        """
        Exercise format-only validation paths so the first request does not pay
        singleton setup, parser, wildcard, and domain metadata import costs.

        Per-user scope catalogs remain lazy because they are backed by PKM rows
        and must be evaluated against the requesting user's latest manifests.
        """
        warmup_scopes = (
            "attr.financial.*",
            "attr.financial.profile.*",
            "attr.financial.profile.risk_score",
        )
        for scope in warmup_scopes:
            await self.validate_scope(scope)
            self.get_scope_display_info(scope)

        self.matches_wildcard(
            "attr.financial.profile.risk_score",
            "attr.financial.profile.*",
        )


# Singleton instance
_scope_generator: Optional[DynamicScopeGenerator] = None


def get_scope_generator() -> DynamicScopeGenerator:
    """Get singleton DynamicScopeGenerator instance."""
    global _scope_generator
    if _scope_generator is None:
        _scope_generator = DynamicScopeGenerator()
    return _scope_generator
