"""Egress policy module: central PDP for search engines, LLM endpoints,
embeddings, and URL fetches.

This is an in-process correctness guardrail, NOT a hard security boundary.
It defends against honest misconfiguration, prompt-injection-induced
URL fetches, accidental egress, and the LangGraph silent-expansion bug.
It does NOT defend against compromised dependencies, code-execution
in the LDR process, or a determined adversary who can modify the policy
module itself. Operators needing a hard boundary should layer OS-level
controls (network namespaces, firewall rules, restricted Docker).

Vocabulary borrowed from XACML / zero-trust:
  PDP — Policy Decision Point (the evaluate_* functions in this module)
  PEP — Policy Enforcement Point (the call sites that consult the PDP)
"""

from __future__ import annotations

import socket
import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from urllib.parse import unquote, urlsplit

from loguru import logger

from ..network_utils import is_private_ip
from ...utilities.type_utils import unwrap_setting


# Bounded DNS lookup timeout; getaddrinfo has no native timeout kwarg, so
# we run it inside a single-shot ThreadPoolExecutor and abandon the
# Future after the timeout elapses. This avoids the previous
# socket.setdefaulttimeout() approach, which mutated a process-global
# setting and could corrupt unrelated network code under concurrency.
_DNS_TIMEOUT_SEC = 2.0


# Hard limit: after this many denied fetches in a single run, fail closed
# even on otherwise-legal fetches. Prevents an exhaust attack via a malicious
# indexed document looping the agent through hundreds of denied URLs.
MAX_DENIED_FETCHES_PER_RUN = 50

# Only SECURITY-relevant denials count toward the per-run quota. Benign parse
# failures (a mailto:/ftp:/tel: link or a malformed href scraped from a page)
# are common in legitimate documents and must NOT exhaust the budget — doing
# so would make a long PUBLIC_ONLY run start refusing legitimate public URLs
# mid-run. These reasons are still audit-logged; they just don't tick the quota.
# ``dangerous_scheme`` (javascript:/data:/file:/… hrefs) belongs here too: these
# are non-fetchable, non-network schemes scraped from ordinary HTML (onclick
# handlers, inline data: URIs), so — like ``unsupported_scheme`` — they can't
# cause egress and shouldn't let a doc full of data: URIs exhaust the budget.
_NON_QUOTA_DENIAL_REASONS = frozenset(
    {"url_malformed", "unsupported_scheme", "no_hostname", "dangerous_scheme"}
)


class EgressScope(str, Enum):
    """User-declared egress boundary for a research run.

    - STRICT: only the user's primary engine; no expansion at all.
    - PUBLIC_ONLY: any public (external web/academic) engine.
    - PRIVATE_ONLY: any private (local collection / library) engine.
    - BOTH: any classified engine (preserves pre-policy behavior).
    - ADAPTIVE: scope FOLLOWS the primary engine — a concrete private
      primary behaves as PRIVATE_ONLY, a concrete public primary as
      PUBLIC_ONLY, and an unclassifiable primary as BOTH.
      The default: most users never touch scope and "it just matches my
      main engine." Resolved to a concrete scope at context construction;
      the stored EgressContext carries the RESOLVED scope, not ADAPTIVE.
    """

    STRICT = "strict"
    PUBLIC_ONLY = "public_only"
    PRIVATE_ONLY = "private_only"
    BOTH = "both"
    ADAPTIVE = "adaptive"


# Code-side single source of truth for the default egress scope, used by
# every reader that needs a fallback for a MISSING policy.egress_scope key
# (partial snapshots from the programmatic API, un-bootstrapped settings
# DBs). Import THIS instead of hardcoding a string literal: scattered
# literals are how the registry default ("adaptive") and the code
# fallbacks ("both") drifted apart in the first place. Must match the
# registered default in defaults/default_settings.json — pinned by
# tests/security/test_egress_policy.py::
# test_default_scope_constant_matches_registry.
DEFAULT_EGRESS_SCOPE: str = EgressScope.ADAPTIVE.value


@dataclass(frozen=True)
class EgressContext:
    """Frozen per-run policy snapshot.

    Constructed once via ``context_from_snapshot()`` at run-start.
    The dataclass is frozen, but mutable internals (``_dns_cache``,
    ``_fetch_denial_count``) use ``field(init=False, default_factory=...)``
    so they can accumulate state during a run. Counters live inside a
    dict because direct ``int`` field reassignment fails on ``frozen=True``.
    """

    scope: EgressScope
    primary_engine: str
    require_local_llm: bool
    require_local_embeddings: bool
    local_hostnames: tuple[str, ...] = ()
    username: Optional[str] = None
    # Mutable internal state (kept off the public surface).
    _dns_cache: dict = field(init=False, default_factory=dict, repr=False)
    _fetch_denial_count: dict = field(
        init=False, default_factory=lambda: {"count": 0}, repr=False
    )
    # Guards _dns_cache + _fetch_denial_count mutations. Scope is the
    # cache writes only — DNS I/O must NOT happen while holding the lock,
    # or concurrent subagent threads serialize on each other's lookups.
    _lock: threading.RLock = field(
        init=False, default_factory=threading.RLock, repr=False
    )


@dataclass(frozen=True)
class Decision:
    """Result of a PDP evaluation. ``reason`` is a short machine code
    (e.g. ``"unclassified"``, ``"scope_mismatch"``) — never the user's
    rejected query or URL content.
    """

    allowed: bool
    reason: str


class PolicyDeniedError(RuntimeError):
    """Raised by PEPs when a policy decision is hard-stop denial.

    Raising (rather than returning a graceful empty result) ensures
    consistent denial latency, which mitigates the LangGraph timing-leak
    pattern where an LLM could infer policy state from how fast a denied
    tool call returns.
    """

    def __init__(self, decision: Decision, target: str = ""):
        self.decision = decision
        self.target = target
        super().__init__(f"policy_denied: {decision.reason}")


def _is_nat64_wrapped_metadata(hostname: str) -> bool:
    """True iff ``hostname`` parses as an IPv6 NAT64 address wrapping a
    cloud-metadata IPv4 (AWS / Azure / GCE / etc).

    The wrapping passes ``is_private_ip`` (because IPv6 link-local
    ranges match) but the embedded IPv4 actually reaches the metadata
    endpoint. We classify these as PUBLIC so STRICT and PRIVATE_ONLY
    refuse to fetch them.
    """
    try:
        import ipaddress

        from ..ssrf_validator import is_nat64_wrapped_metadata_ip

        candidate = hostname
        if candidate.startswith("[") and candidate.endswith("]"):
            candidate = candidate[1:-1]
        ip = ipaddress.ip_address(candidate)
        return is_nat64_wrapped_metadata_ip(ip)
    except (ValueError, TypeError):
        return False
    except Exception:  # pragma: no cover - defensive
        return False


def _resolve_with_timeout(hostname: str) -> Optional[list]:
    """Resolve ``hostname`` via getaddrinfo with a bounded timeout.

    Returns the addrinfo list on success or None on timeout / lookup
    failure. getaddrinfo has no native timeout, so we drive it from a
    single-shot worker thread and abandon the Future on timeout — far
    safer than socket.setdefaulttimeout(), which mutates a
    process-global setting and races with unrelated network code.
    """

    def _do_lookup():
        return socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)

    # NB: do NOT use ``with ThreadPoolExecutor(...)`` here. The context
    # manager's __exit__ calls shutdown(wait=True), which blocks until the
    # worker thread finishes — so a hung getaddrinfo would defeat the whole
    # point of the timeout (the call would return only after the OS DNS
    # timeout, not after _DNS_TIMEOUT_SEC). Instead we abandon the worker on
    # timeout via shutdown(wait=False): the caller returns promptly and the
    # orphaned thread dies on its own when getaddrinfo eventually returns.
    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ldr-dns")
    future = executor.submit(_do_lookup)
    try:
        return future.result(timeout=_DNS_TIMEOUT_SEC)
    except (FutureTimeout, socket.gaierror, socket.timeout, OSError):
        return None
    except Exception:  # pragma: no cover - defensive
        return None
    finally:
        # wait=False so a timed-out lookup does not re-block here; a
        # completed lookup leaves an idle thread that shuts down at once.
        executor.shutdown(wait=False)


_UNCACHED = object()


def _cache_classification(
    ctx: EgressContext, hostname: str, value: bool
) -> bool:
    """First-writer-wins cache set; returns the value now in the cache.

    Concurrent subagent threads may classify the same hostname outside the
    lock (the DNS lookup is intentionally unsynchronized). If a round-robin
    DNS name resolves to a private IP for one thread and a public IP for
    another, last-writer-wins would let a later call flip an earlier call's
    result — cache *incoherence* that could relax PRIVATE_ONLY/STRICT on a
    subsequent fetch. Pinning the first writer's value makes the per-run
    classification stable and deterministic regardless of completion order.
    """
    with ctx._lock:
        existing = ctx._dns_cache.get(hostname, _UNCACHED)
        if existing is not _UNCACHED:
            return existing
        ctx._dns_cache[hostname] = value
        return value


def _classify_host(
    hostname: str, ctx: EgressContext, allow_dns: bool = True
) -> Optional[bool]:
    """Classify a hostname as local (True) / public (False) / unknown (None).

    Uses the run-scoped DNS cache to avoid repeated ``getaddrinfo`` lookups.
    Falls back to public on DNS timeout (fail-safe).

    Lock discipline: cache reads and writes are guarded by ``ctx._lock``;
    the DNS lookup itself runs OUTSIDE the lock so concurrent subagent
    threads don't serialize on each other's DNS calls. Cache writes are
    first-writer-wins (``_cache_classification``) so a hostname's
    classification is stable for the run even under concurrent disagreeing
    lookups.
    """
    if not hostname:
        return None

    # Cache hit — read under the lock so concurrent writers don't tear
    # the dict.
    with ctx._lock:
        if hostname in ctx._dns_cache:
            return ctx._dns_cache[hostname]

    # User-declared local hostnames override DNS classification.
    if hostname in ctx.local_hostnames:
        return _cache_classification(ctx, hostname, True)

    # NAT64-wrapped metadata IPs (e.g. 64:ff9b::169.254.169.254) classify
    # as link-local by is_private_ip but actually wrap AWS/GCE instance
    # metadata. Force them to PUBLIC so STRICT/PRIVATE_ONLY don't allow
    # them. Check happens BEFORE is_private_ip so the wrapping wins.
    if _is_nat64_wrapped_metadata(hostname):
        return _cache_classification(ctx, hostname, False)

    # String-literal check first (no DNS needed for IPs and known literals).
    # If the helper itself raises (malformed hostname, broken stdlib edge case),
    # fall through to DNS resolution rather than fail the whole evaluation —
    # the DNS path has its own bounded timeout and error handling.
    from ..ssrf_validator import is_ip_blocked

    try:
        # A literal cloud-metadata IP (169.254.169.254, fd00:ec2::254, …) passes
        # is_private_ip (link-local) but must NOT be treated as local — otherwise
        # STRICT/PRIVATE_ONLY would accept an IMDS SSRF target as a "local"
        # inference/search host. Mirror the DNS-branch metadata block below.
        # is_ip_blocked returns False for non-IP hostnames (ValueError → False),
        # so legitimate hostnames still fall through to is_private_ip / DNS.
        if is_ip_blocked(
            hostname, allow_localhost=True, allow_private_ips=True
        ):
            return _cache_classification(ctx, hostname, False)
        if is_private_ip(hostname):
            return _cache_classification(ctx, hostname, True)
    except Exception as exc:
        logger.debug(
            "is_private_ip raised on hostname, falling back to DNS",
            error=str(exc),
        )

    # DNS-resolved classification — runs OUTSIDE the lock.
    if not allow_dns:
        # The caller (the advisory warning-banner render path) opted OUT of
        # the synchronous getaddrinfo so a settings-page render never blocks
        # up to _DNS_TIMEOUT_SEC on a lookup. Return "unknown" so adaptive
        # scope resolution falls back to the engine's static classification.
        # Deliberately NOT cached — a later enforcement-path call (allow_dns
        # default True) must still resolve this host for real.
        return None
    addr_info = _resolve_with_timeout(hostname)
    if addr_info is None:
        # Fail-safe: unresolvable / timeout → treat as public.
        return _cache_classification(ctx, hostname, False)

    # If any resolved IP is private (and not a NAT64 metadata wrap),
    # classify as local.
    for entry in addr_info:
        ip_str = entry[4][0]
        try:
            # A hostname that resolves to a cloud-metadata IP must NOT be
            # treated as local: link-local metadata IPs (169.254.169.254, …)
            # pass is_private_ip, which would classify the host local and let
            # STRICT/PRIVATE_ONLY fetch it. Classify as public so those scopes
            # refuse it — mirrors the literal-IP metadata block in evaluate_url
            # and the NAT64 handling just below. (Under PUBLIC_ONLY/BOTH the
            # SSRF validator at the actual fetch is the metadata backstop.)
            if is_ip_blocked(
                ip_str, allow_localhost=True, allow_private_ips=True
            ):
                return _cache_classification(ctx, hostname, False)
            if _is_nat64_wrapped_metadata(ip_str):
                return _cache_classification(ctx, hostname, False)
            if is_private_ip(ip_str):
                return _cache_classification(ctx, hostname, True)
        except Exception:  # pragma: no cover - defensive
            continue
    return _cache_classification(ctx, hostname, False)


def _get_engine_class(engine_name: str):
    """Lazy load the engine class from the registry to avoid circular imports."""
    # Import is lazy to break the dependency cycle:
    # security/* → engines/* → security/*
    from ...web_search_engines.engine_registry import ENGINE_REGISTRY
    from ..module_whitelist import get_safe_module_class

    entry = ENGINE_REGISTRY.get(engine_name)
    if entry is None:
        return None
    try:
        return get_safe_module_class(entry.module_path, entry.class_name)
    except Exception:
        return None


def _engine_flags(engine_cls):
    """Read the (is_public, is_local, url_setting) triple from an engine class.

    Uses ``is True`` / ``is False`` semantics: a missing attribute is treated
    as "not declared", which is distinct from an attribute explicitly set
    to ``False``.
    """
    is_public = getattr(engine_cls, "is_public", None)
    is_local = getattr(engine_cls, "is_local", None)
    url_setting = getattr(engine_cls, "url_setting", None)
    return is_public, is_local, url_setting


def _resolve_collection_is_public(
    engine_name: str, username: Optional[str]
) -> bool:
    """Look up a collection engine's per-collection ``is_public`` flag.

    ``engine_name`` is ``"collection_<uuid>"`` or ``"library"``. The
    aggregate ``library`` is always private (it spans all local docs).
    Fails closed to private (False) on any lookup error or missing row —
    a collection we can't classify must NOT be treated as public.
    """
    if engine_name == "library" or not engine_name.startswith("collection_"):
        return False
    collection_id = engine_name[len("collection_") :]
    try:
        from ...database.models.library import Collection
        from ...database.session_context import get_user_db_session

        with get_user_db_session(username) as session:
            row = (
                session.query(Collection)
                .filter(Collection.id == collection_id)
                .first()
            )
            return bool(getattr(row, "is_public", False)) if row else False
    except Exception:  # noqa: silent-exception - fail closed to private
        logger.debug(
            "could not resolve collection is_public; treating as private",
            engine=engine_name,
        )
        return False


def _engine_bucket(
    engine_name: str,
    ctx: EgressContext,
    settings_snapshot: dict,
    metadata: Optional[dict] = None,
    allow_dns: bool = True,
):
    """Return the ``(is_public, is_local)`` classification for an engine based
    on its **Python class flags**, with a fail-up URL override.

    The class-level ``is_public`` / ``is_local`` flags express the engine's
    *nature* — whether it queries the public internet or searches local data —
    regardless of where the engine's server happens to be hosted.  This is used
    by ``evaluate_engine`` and ``_resolve_adaptive_scope`` so that e.g. a
    locally-hosted SearXNG (which proxies to Google/Bing) is still treated as
    a public engine.

    The URL override is asymmetric and can only make the classification
    MORE restrictive: a local-nature engine whose ``url_setting`` resolves
    to a public host is reclassified public (queries would leave the box),
    while a public-nature engine on a local URL stays public.

    - Static engines: read declared ``is_public``/``is_local`` flags directly.
    - ``library`` / ``collection_<uuid>``: per-collection ``is_public``
      flag (from ``metadata`` when the caller has the engine config, else a
      direct DB lookup), defaulting private.
    - Anything else (unknown): ``(None, None)``.
    """
    engine_cls = _get_engine_class(engine_name)
    if engine_cls is None:
        if engine_name == "library" or engine_name.startswith("collection_"):
            if metadata is not None and metadata.get("is_public") is not None:
                is_public = bool(metadata.get("is_public"))
            else:
                is_public = _resolve_collection_is_public(
                    engine_name, ctx.username
                )
            # A collection is ALWAYS a local knowledge base — it lives on this
            # machine — so it stays searchable under PRIVATE_ONLY regardless of
            # its public flag. ``is_public`` is ADDITIVE, not exclusive: marking
            # a collection public (non-sensitive content — papers you fetched,
            # public publications) ALSO makes it eligible under PUBLIC_ONLY and
            # OK to process with cloud inference. A private collection is
            # local-only and excluded from public-scope runs. Hence
            # ``(is_public, is_local=True)`` rather than the old mutually-
            # exclusive ``(is_public, not is_public)`` which wrongly hid a
            # public collection from private runs.
            return is_public, True
        return None, None
    is_public, is_local, url_setting = _engine_flags(engine_cls)
    # Fail-up URL override (asymmetric, only ever MORE restrictive): a
    # local-data engine (Elasticsearch, Paperless) whose configured URL
    # resolves to a PUBLIC host is reclassified as public — querying it
    # sends the user's queries off the box, so PRIVATE_ONLY must deny it
    # at selection time (matching pre-static-flags behavior). The reverse
    # direction is deliberately NOT applied: a public-nature engine
    # (SearXNG) hosted on localhost stays public, because its data source
    # is the internet regardless of where the proxy runs.
    if url_setting and is_local is True and is_public is not True:
        host_classification = _classify_engine_url(
            url_setting, settings_snapshot, ctx, allow_dns=allow_dns
        )
        if host_classification is False:
            # False from _classify_engine_url == "connect target is a
            # PUBLIC host" -> fail up to public.
            return True, False
    return is_public, is_local


# Denial reasons that justify dropping an engine from an LLM candidate
# list. Only ACTIVE policy denials qualify — see filter_engines_by_egress.
_PREFILTER_STRIP_REASONS = frozenset(
    {
        "scope_mismatch_public_only",
        "scope_mismatch_private_only",
        "strict_not_primary",
        # Registered engine without classification flags: the factory's
        # evaluate_engine call fails it closed too, so stripping matches
        # enforcement exactly.
        "unclassified",
    }
)


def filter_engines_by_egress(
    engine_names: list[str],
    ctx: EgressContext,
    settings_snapshot: dict,
) -> list[str]:
    """ADVISORY pre-filter for a candidate engine list: remove engines the
    egress scope actively denies so the LLM doesn't waste selection slots
    on them. Uses the engine's static class flags (``is_public`` /
    ``is_local``) — a locally-hosted SearXNG is still treated as public
    because it queries the internet.

    This is a UX optimization in FRONT of the factory PEP, not an
    enforcement point — so it must never be STRICTER than the factory.
    Names unknown to the static registry (``engine_unknown``) are KEPT
    (except under STRICT, where the not-the-primary gate fires before the
    registry lookup — exactly as it does in the factory):
    they may be retriever-backed or dynamically injected engines that the
    factory evaluates via its own retriever path; dropping them here would
    silently hide legitimate engines from LLM selection. The factory
    still denies anything truly disallowed at instantiation.

    Callers: the LangGraph tool builder, or any code that builds a
    selection set before presenting it to an LLM.
    """
    allowed: list[str] = []
    for name in engine_names:
        decision = evaluate_engine(
            name, ctx, settings_snapshot=settings_snapshot
        )
        if decision.allowed or decision.reason not in _PREFILTER_STRIP_REASONS:
            allowed.append(name)
    return allowed


def filter_candidates_by_egress(
    engine_names: list[str],
    settings_snapshot: Optional[dict],
) -> list[str]:
    """Best-effort scope pre-filter for an LLM candidate list.

    Wraps ``filter_engines_by_egress`` with the snapshot plumbing
    LLM-selection callers share: read ``policy.egress_scope`` /
    ``search.tool`` from the snapshot (flat or ``{"value": ...}`` shaped),
    build the context, filter. Returns the input list unchanged when the
    snapshot is missing, the scope is BOTH (nothing to strip), or anything
    fails — the factory PEP still enforces at child instantiation, so this
    helper must never break engine selection.
    """
    if not settings_snapshot:
        return engine_names
    try:
        scope_raw = str(
            _get_setting_value(
                settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
            )
            or DEFAULT_EGRESS_SCOPE
        ).lower()
        # BOTH (or an unrecognized value) strips nothing at selection time;
        # adaptive is included because it can RESOLVE to a restrictive scope.
        if scope_raw not in (
            "private_only",
            "public_only",
            "strict",
            "adaptive",
        ):
            return engine_names
        primary = resolve_run_primary_engine(settings_snapshot)
        ctx = context_from_snapshot(
            settings_snapshot,
            primary,
            username=settings_snapshot.get("_username"),
        )
        return filter_engines_by_egress(engine_names, ctx, settings_snapshot)
    except Exception:  # noqa: silent-exception - advisory pre-filter only
        logger.bind(policy_audit=True).debug(
            "egress candidate pre-filter failed; "
            "factory PEP still enforces at child instantiation"
        )
        return engine_names


def evaluate_engine(
    engine_name: str,
    ctx: EgressContext,
    *,
    settings_snapshot: dict,
    metadata: Optional[dict] = None,
) -> Decision:
    """Decide whether an engine may be instantiated under the current policy.

    Allows iff (a) the engine's classification is compatible with the scope's
    bucket AND (b) if scope == STRICT, the engine name equals the primary.
    Unclassified engines always fail closed.

    ``metadata`` (optional) carries the engine's config entry (e.g. the
    per-collection ``is_public`` from search_config) so callers that already
    have it avoid a redundant DB lookup; when absent it's resolved directly.
    """
    if settings_snapshot is None:
        return Decision(False, "no_snapshot")
    try:
        # STRICT: only the primary engine is permitted.
        if (
            ctx.scope == EgressScope.STRICT
            and engine_name != ctx.primary_engine
        ):
            return Decision(False, "strict_not_primary")

        # A truly unknown engine (not in the static registry and not a
        # collection) fails closed as engine_unknown. ``library`` /
        # ``collection_*`` are classified by _engine_bucket below.
        is_collection = engine_name == "library" or engine_name.startswith(
            "collection_"
        )
        if _get_engine_class(engine_name) is None and not is_collection:
            return Decision(False, "engine_unknown")

        is_public, is_local = _engine_bucket(
            engine_name, ctx, settings_snapshot, metadata
        )

        # Fail-closed for unclassified engines.
        if is_public is not True and is_local is not True:
            return Decision(False, "unclassified")

        # Scope/bucket compatibility.
        if ctx.scope == EgressScope.PUBLIC_ONLY and is_public is not True:
            return Decision(False, "scope_mismatch_public_only")
        if ctx.scope == EgressScope.PRIVATE_ONLY and is_local is not True:
            return Decision(False, "scope_mismatch_private_only")

        return Decision(True, "allowed")
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception(
            "evaluate_engine internal error", engine=engine_name
        )
        return Decision(False, "internal_error")


def _classify_engine_url(
    url_setting: str,
    settings_snapshot: dict,
    ctx: EgressContext,
    allow_dns: bool = True,
) -> Optional[bool]:
    """Classify an engine's configured URL via DNS resolution.

    For list-typed settings (Elasticsearch ``hosts``), returns False (public)
    if ANY entry classifies as public — safer fail-up.
    Returns True if local, False if public, None if undetermined.
    """
    value = _get_setting_value(settings_snapshot, url_setting, None)
    if not value:
        return None

    entries = value if isinstance(value, list) else [value]
    any_public = False
    any_local = False
    for entry in entries:
        if not isinstance(entry, str):
            continue
        try:
            parsed = urlsplit(entry if "://" in entry else f"http://{entry}")
            hostname = parsed.hostname
        except Exception:
            continue
        if not hostname:
            continue
        # Decode percent-encoding before classifying, identically to
        # evaluate_url: the HTTP client decodes the host before connecting,
        # so "192%2e168%2e1%2e1" must be classified as 192.168.1.1, not as
        # an unresolvable literal.
        hostname = unquote(hostname)
        result = _classify_host(hostname, ctx, allow_dns=allow_dns)
        if result is True:
            any_local = True
        elif result is False:
            any_public = True

    if any_public:
        return False  # "any public" wins under list-typed settings
    if any_local:
        return True
    return None


_CLOUD_LLM_PROVIDERS = frozenset(
    {
        "openai",
        "anthropic",
        "google",
        "openrouter",
        "deepseek",
        "xai",
        "ionos",
    }
)

# Providers that default to a localhost endpoint and ship without any
# remote routing. The LLM gate at ``config/llm_config.py`` uses this set
# as a SNAPSHOT-LESS ALLOW-LIST: when ``get_llm`` is invoked with no
# settings snapshot (background helpers, scaffolding paths, tests), only
# these providers may proceed; everything else — including ambiguous
# providers like ``openai_endpoint`` and any future cloud provider not
# yet enumerated in ``_CLOUD_LLM_PROVIDERS`` — fails closed at the gate.
# Keeping the set tight is the point: anything new defaults to refused.
_LOCAL_DEFAULT_LLM_PROVIDERS = frozenset(
    {
        "ollama",
        "lmstudio",
        "llamacpp",
    }
)

# Embeddings analogue of ``_LOCAL_DEFAULT_LLM_PROVIDERS``. Used by the
# embeddings gate as a SNAPSHOT-LESS ALLOW-LIST: when ``get_embeddings``
# runs without a settings snapshot (the ``embeddings.require_local`` toggle
# is then unreadable), only these localhost-default providers may proceed.
# ``openai`` (and any future cloud embeddings provider) fails closed —
# matching the LLM gate so a snapshot-less background caller can't ship the
# local corpus to a cloud embedder. ``sentence_transformers`` runs fully
# in-process; ``ollama`` defaults to a localhost endpoint.
_LOCAL_DEFAULT_EMBEDDING_PROVIDERS = frozenset(
    {
        "sentence_transformers",
        "ollama",
    }
)


def _is_user_registered_llm(provider: str) -> bool:
    """True when ``provider`` is a user-registered in-process LLM.

    The programmatic API (``quick_summary(llms={"mock": ...})``) and plugins
    register LLM objects in the LLM registry. Built-in providers are ALSO
    auto-registered there by ``discover_providers()``, so registry membership
    alone cannot discriminate — a name that is registered but NOT one of the
    auto-discovered built-ins is user-supplied in-process code. Such objects
    carry no configurable endpoint the PDP could classify (denying them with
    ``provider_url_unset`` just breaks the documented offline workflow), the
    operator injected them deliberately at code level, and under
    PRIVATE_ONLY/STRICT the PEP-578 audit hook still blocks any stray
    outbound socket they might open.

    A user registration that *shadows* a built-in name (e.g. "openai") is NOT
    treated as user code: the discovered-name check keeps it on the strict
    path, and the ``_CLOUD_LLM_PROVIDERS`` gate fires first anyway.

    Fails closed (False) on any error.
    """
    try:
        from ...llm.llm_registry import is_llm_registered
        from ...llm.providers import discover_providers
        from ...llm.providers.base import normalize_provider

        if not provider or not is_llm_registered(provider):
            return False
        discovered = {normalize_provider(key) for key in discover_providers()}
        return normalize_provider(provider) not in discovered
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception(
            "user-registered LLM check failed", provider=provider
        )
        return False


def evaluate_llm_endpoint(
    provider: str,
    ctx: EgressContext,
    *,
    settings_snapshot: dict,
) -> Decision:
    """Decide whether an LLM provider may be instantiated.

    Only meaningful when ``ctx.require_local_llm`` is True; otherwise allow.
    """
    if settings_snapshot is None:
        return Decision(False, "no_snapshot")
    try:
        if not ctx.require_local_llm:
            return Decision(True, "no_local_requirement")

        # Hard-cloud providers always fail under require_local_llm.
        if provider in _CLOUD_LLM_PROVIDERS:
            return Decision(False, "provider_cloud_only")

        # User-registered in-process LLMs (programmatic API / plugins) have
        # no endpoint to classify; allow them — the audit hook still
        # backstops stray sockets under PRIVATE_ONLY/STRICT.
        if _is_user_registered_llm(provider):
            return Decision(True, "user_registered_llm")

        # Providers with a configurable URL: classify the URL.
        url_key = f"llm.{provider}.url"
        url_value = _get_setting_value(settings_snapshot, url_key, None)
        if not url_value:
            # Local-default providers (ollama, lmstudio, llamacpp) without
            # an override fall back to their localhost defaults. Use the
            # shared constant so this can't drift from the snapshot-less
            # allow-list.
            if provider in _LOCAL_DEFAULT_LLM_PROVIDERS:
                return Decision(True, "provider_local_default")
            return Decision(False, "provider_url_unset")

        try:
            parsed = urlsplit(url_value)
            # Percent-decode the host before classifying — the HTTP client
            # decodes it before connect, so "http://127%2e0%2e0%2e1:11434"
            # must read as the local 127.0.0.1 (matches evaluate_url and
            # _classify_engine_url). Without this a legitimate percent-encoded
            # local endpoint is wrongly classified public and denied.
            hostname = unquote(parsed.hostname) if parsed.hostname else None
        except Exception:
            return Decision(False, "url_malformed")

        classification = _classify_host(hostname, ctx) if hostname else None
        if classification is True:
            return Decision(True, "provider_local")
        return Decision(False, "provider_remote")
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception(
            "evaluate_llm_endpoint internal error", provider=provider
        )
        return Decision(False, "internal_error")


def evaluate_embeddings(
    provider: str,
    ctx: EgressContext,
    *,
    settings_snapshot: dict,
) -> Decision:
    """Decide whether an embeddings provider may be instantiated.

    Only meaningful when ``ctx.require_local_embeddings`` is True.
    OpenAI is treated as unambiguously cloud unless ``embeddings.openai.base_url``
    is configured to a local host.
    """
    if settings_snapshot is None:
        return Decision(False, "no_snapshot")
    try:
        if not ctx.require_local_embeddings:
            return Decision(True, "no_local_requirement")

        if provider == "sentence_transformers":
            return Decision(True, "provider_local")

        if provider == "ollama":
            url = _get_setting_value(
                settings_snapshot, "embeddings.ollama.url", None
            ) or _get_setting_value(settings_snapshot, "llm.ollama.url", None)
            if not url:
                return Decision(True, "provider_local_default")
            parsed = urlsplit(url)
            host = unquote(parsed.hostname) if parsed.hostname else None
            classification = _classify_host(host, ctx) if host else None
            return (
                Decision(True, "provider_local")
                if classification is True
                else Decision(False, "provider_remote")
            )

        if provider == "openai":
            base_url = _get_setting_value(
                settings_snapshot, "embeddings.openai.base_url", None
            )
            if base_url:
                parsed = urlsplit(base_url)
                host = unquote(parsed.hostname) if parsed.hostname else None
                classification = _classify_host(host, ctx) if host else None
                if classification is True:
                    return Decision(True, "provider_local_endpoint")
            return Decision(False, "provider_cloud")

        return Decision(False, "provider_unknown")
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception(
            "evaluate_embeddings internal error", provider=provider
        )
        return Decision(False, "internal_error")


_DANGEROUS_SCHEMES = frozenset(
    {"javascript", "data", "file", "vbscript", "about"}
)

# Cloud-metadata endpoints reachable by HOSTNAME rather than literal IP. GCP's
# IMDS answers on metadata.google.internal / metadata.goog (resolving to
# 169.254.169.254); is_ip_blocked only catches the IP literal, so these names
# need an explicit block to honor the "metadata is NEVER permitted, regardless
# of scope" invariant in evaluate_url. AWS/Azure/Alibaba IMDS have no such
# hostname (IP only) and are covered by the is_ip_blocked / alt-encoding path.
_METADATA_HOSTNAMES = frozenset({"metadata.google.internal", "metadata.goog"})


def _normalize_alt_ipv4(host: str) -> Optional[str]:
    """Return the canonical dotted-quad for an IPv4 host written in an
    alternate encoding the libc resolver accepts but ``ipaddress.ip_address``
    rejects — octal (``0251.0376.0251.0376``), hex (``0xa9fea9fe``), integer
    (``2852039166``) and short forms (``169.254.43518``). Returns ``None`` for
    anything ``inet_aton`` won't parse (real hostnames, IPv6, junk).

    ``socket.getaddrinfo`` resolves these numeric forms to the same address the
    HTTP client will ``connect()`` to, so the metadata-IP block in
    ``evaluate_url`` must classify them by their *real* target — otherwise
    ``http://0251.0376.0251.0376/`` slips past ``is_ip_blocked`` (which only
    parses canonical notation) and reads as an allowed public host.
    """
    try:
        import socket

        return socket.inet_ntoa(socket.inet_aton(host))
    except (OSError, UnicodeError, TypeError):
        return None


def _quota_ctx(ctx: EgressContext) -> EgressContext:
    """Return the EgressContext whose counter backs the per-RUN denied-fetch
    quota.

    Each call site (ContentFetcher, full_search, download_service, the audit
    hook) builds its OWN EgressContext from the snapshot, so a per-context
    counter would reset the budget every time a new engine/fetcher is built —
    letting a malicious document evade ``MAX_DENIED_FETCHES_PER_RUN`` by
    spreading denied fetches across contexts. The run's ARMED active context is
    set once at run start and re-armed identically on pool workers, so it is
    shared across the whole run; anchoring the counter there makes the quota
    truly per-run. Falls back to the passed ctx when no run context is armed
    (snapshot-less / programmatic callers, settings-page calls), keeping those
    isolated calls self-contained.
    """
    try:
        from .audit_hook import get_active_context

        active = get_active_context()
    except Exception:  # pragma: no cover - defensive
        active = None
    return active if active is not None else ctx


def evaluate_url(url: str, ctx: EgressContext) -> Decision:
    """Decide whether an arbitrary URL may be fetched (e.g., by ``fetch_content``).

    Enforces the run's denied-fetch quota to prevent exhaustion attacks
    via malicious indexed documents that loop the agent through hundreds
    of denied fetches.
    """
    try:
        # Quota check: hard fail after MAX_DENIED_FETCHES_PER_RUN. Anchor the
        # counter to the run's active context (``_quota_ctx``) so the budget is
        # per-RUN, not per-EgressContext. Read under that ctx's lock so
        # concurrent subagents see a consistent count.
        qctx = _quota_ctx(ctx)
        with qctx._lock:
            if qctx._fetch_denial_count["count"] >= MAX_DENIED_FETCHES_PER_RUN:
                return Decision(False, "denial_quota_exceeded")

        if not isinstance(url, str) or not url:
            return _record_denial(ctx, "url_malformed")

        try:
            parsed = urlsplit(url)
        except Exception:
            return _record_denial(ctx, "url_malformed")

        if parsed.scheme.lower() in _DANGEROUS_SCHEMES:
            return _record_denial(ctx, "dangerous_scheme")
        if parsed.scheme.lower() not in ("http", "https"):
            return _record_denial(ctx, "unsupported_scheme")
        if not parsed.hostname:
            return _record_denial(ctx, "no_hostname")

        # HTTP client libraries (requests/urllib3) percent-DECODE the host
        # in the netloc before the socket connect, but urlsplit().hostname
        # preserves the encoding. Classifying the encoded form lets
        # "http://192%2e168%2e1%2e1/" read as an unresolvable public host
        # (DNS fails) while the client actually connects to the private
        # 192.168.1.1 — a scope bypass under PUBLIC_ONLY. Classify the
        # DECODED host so the policy sees the real connect target.
        host = unquote(parsed.hostname)
        # Trailing dots are insignificant to the resolver (getaddrinfo strips
        # them), so "169.254.169.254." and "metadata.google.internal." must be
        # classified identically to their bare forms — otherwise a trailing
        # dot dodges the metadata checks below. Mirrors the SSRF validator's
        # rstrip(".").
        host = host.rstrip(".")

        # Cloud-metadata endpoints (AWS/GCE/Azure IMDS at 169.254.169.254
        # etc.) are NEVER permitted, regardless of scope. They classify as
        # link-local/private, so STRICT and PRIVATE_ONLY would otherwise
        # ALLOW them — a credential-theft path for prompt-injected fetches
        # and, more importantly, for the audit-hook net which calls
        # evaluate_url directly on raw socket.connect targets (bypassing the
        # SSRF validator that the explicit fetch PEPs run first). Reuse
        # is_ip_blocked(allow_private_ips=True), which returns True only for
        # the always-blocked metadata set (+ NAT64 wraps) and also unwraps
        # IPv4-mapped IPv6 forms.
        try:
            from ..ssrf_validator import is_ip_blocked

            if is_ip_blocked(
                host, allow_localhost=True, allow_private_ips=True
            ):
                return _record_denial(ctx, "blocked_metadata_ip")
            # is_ip_blocked only parses canonical IP notation, so an
            # alternate-encoded metadata literal (octal/hex/integer) slips
            # through above. Normalize it to the dotted-quad the resolver
            # would connect to and re-check, so the "metadata IPs are NEVER
            # permitted, regardless of scope" invariant holds for those forms
            # too — and they get the explicit blocked_metadata_ip reason
            # rather than leaking into the scope-mismatch bucket.
            alt = _normalize_alt_ipv4(host)
            if (
                alt is not None
                and alt != host
                and is_ip_blocked(
                    alt, allow_localhost=True, allow_private_ips=True
                )
            ):
                return _record_denial(ctx, "blocked_metadata_ip")
            # Metadata endpoints reachable by hostname (GCP) — is_ip_blocked
            # can't see these because they aren't IP literals. Block by name so
            # the invariant holds under PUBLIC_ONLY/BOTH too (the SSRF validator
            # backstops the actual fetch, but evaluate_url's own guarantee must
            # not depend on it).
            if host.lower() in _METADATA_HOSTNAMES:
                return _record_denial(ctx, "blocked_metadata_ip")
        except Exception:  # noqa: silent-exception - defensive, see below
            # Non-IP host or helper error: fall through to normal
            # classification (DNS path has its own handling).
            pass

        classification = _classify_host(host, ctx)

        if ctx.scope == EgressScope.STRICT:
            # Under STRICT, URLs whose host is private are allowed; public
            # hosts (DOI redirects, citations) are not. This is a deliberate
            # trade-off: prompt-injection spoofing of provenance tags is a
            # bigger risk than false-positives on DOI redirects.
            if classification is True:
                return Decision(True, "allowed_private_host_under_strict")
            return _record_denial(ctx, "strict_public_host")

        if ctx.scope == EgressScope.PUBLIC_ONLY:
            if classification is False:
                return Decision(True, "allowed_public_host")
            return _record_denial(ctx, "scope_mismatch_public_only")

        if ctx.scope == EgressScope.PRIVATE_ONLY:
            if classification is True:
                return Decision(True, "allowed_private_host")
            return _record_denial(ctx, "scope_mismatch_private_only")

        # BOTH: any classified host is fine.
        if classification is None:
            return _record_denial(ctx, "host_unclassified")
        return Decision(True, "allowed_both_scope")
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception("evaluate_url internal error")
        return Decision(False, "internal_error")


def _record_denial(ctx: EgressContext, reason: str) -> Decision:
    """Increment the denial counter inside the frozen context's mutable dict
    and emit a redacted audit log line. Lock guards the read-modify-write
    against concurrent subagent threads.
    """
    counts_toward_quota = reason not in _NON_QUOTA_DENIAL_REASONS
    # Increment the per-RUN counter (the run's active context, shared across
    # every call-site context) rather than this one context's — see _quota_ctx.
    qctx = _quota_ctx(ctx)
    with qctx._lock:
        if counts_toward_quota:
            qctx._fetch_denial_count["count"] += 1
        count = qctx._fetch_denial_count["count"]
    logger.bind(policy_audit=True).warning(
        "policy denied URL fetch",
        reason=reason,
        scope=ctx.scope.value,
        denial_count=count,
        counted=counts_toward_quota,
    )
    return Decision(False, reason)


def evaluate_retriever(
    retriever_name: str,
    ctx: EgressContext,
    *,
    metadata: Optional[dict] = None,
) -> Decision:
    """Decide whether a registered retriever may be invoked.

    Reads classification (``{"is_local": bool}``) set at registration
    time. Unclassified retrievers fail closed. ``metadata`` may be
    supplied by the caller (e.g. the search-engine factory, which has
    already looked up the retriever in its own registry reference) to
    avoid a second registry lookup and to honor a test-patched registry;
    when ``None`` the global registry is consulted.
    """
    try:
        if metadata is None:
            from ...web_search_engines.retriever_registry import (
                retriever_registry,
            )

            try:
                metadata = retriever_registry.get_metadata(
                    retriever_name, username=ctx.username
                )
            except AttributeError:
                # Older registry without metadata API; treat as unclassified.
                return Decision(False, "retriever_unclassified")

        if not metadata:
            return Decision(False, "retriever_unknown")

        is_local = metadata.get("is_local")
        if is_local is None:
            return Decision(False, "retriever_unclassified")

        if ctx.scope == EgressScope.STRICT:
            # STRICT permits only the user's primary engine. A retriever
            # IS allowed under STRICT when it is itself the primary (the
            # common "research only against my private KB" setup);
            # otherwise it's an expansion STRICT forbids.
            if retriever_name == ctx.primary_engine:
                return Decision(True, "allowed_primary_retriever")
            return Decision(False, "strict_not_primary")
        if ctx.scope == EgressScope.PUBLIC_ONLY and is_local:
            return Decision(False, "scope_mismatch_public_only")
        if ctx.scope == EgressScope.PRIVATE_ONLY and not is_local:
            return Decision(False, "scope_mismatch_private_only")
        return Decision(True, "allowed")
    except Exception:  # pragma: no cover - defensive
        logger.bind(policy_audit=True).exception(
            "evaluate_retriever internal error", retriever=retriever_name
        )
        return Decision(False, "internal_error")


def _retriever_is_local(
    retriever_name: str, username: Optional[str]
) -> Optional[bool]:
    """Return the ``is_local`` classification (True/False) of a registered
    retriever, or ``None`` when it is unknown/unclassified.

    Mirrors the registry lookup in ``evaluate_retriever`` so adaptive-scope
    resolution and enforcement agree on a retriever primary's classification.
    """
    try:
        from ...web_search_engines.retriever_registry import (
            retriever_registry,
        )

        metadata = retriever_registry.get_metadata(
            retriever_name, username=username
        )
    except Exception:  # noqa: silent-exception - unknown → caller falls back
        return None
    if not metadata:
        return None
    is_local = metadata.get("is_local")
    return is_local if isinstance(is_local, bool) else None


def _resolve_adaptive_scope(
    primary_engine: str,
    settings_snapshot: dict,
    *,
    username: Optional[str],
    local_hostnames: tuple,
    allow_dns: bool = True,
) -> EgressScope:
    """Map ADAPTIVE to a concrete scope by classifying the primary engine.

    Uses the engine's **static class flags** (plus ``_engine_bucket``'s
    fail-up URL override) so that e.g. a locally-hosted SearXNG is still
    treated as a public engine, and a remote-hosted Elasticsearch resolves
    public rather than pulling the run into PRIVATE_ONLY.

    private primary => PRIVATE_ONLY, public primary => PUBLIC_ONLY,
    unknown / classification-error => BOTH (adaptive's documented
    permissive fallback, equivalent to pre-policy behavior, so a
    classification hiccup never hard-fails the run).

    A registered LangChain retriever (private KB) is not in ENGINE_REGISTRY,
    so ``_engine_bucket`` returns ``(None, None)`` for it; we then consult the
    retriever registry so a local retriever primary resolves to PRIVATE_ONLY
    (forcing local inference) rather than leaking the corpus to cloud models.
    """
    if not primary_engine:
        return EgressScope.BOTH
    try:
        probe_ctx = EgressContext(
            scope=EgressScope.BOTH,
            primary_engine=primary_engine,
            require_local_llm=False,
            require_local_embeddings=False,
            local_hostnames=local_hostnames,
            username=username,
        )
        is_public, is_local = _engine_bucket(
            primary_engine, probe_ctx, settings_snapshot, allow_dns=allow_dns
        )
    except Exception:  # noqa: silent-exception - adaptive falls back to BOTH
        logger.bind(policy_audit=True).debug(
            "adaptive scope classification failed; falling back to BOTH",
            primary=primary_engine,
        )
        return EgressScope.BOTH

    if is_local is True and is_public is not True:
        return EgressScope.PRIVATE_ONLY
    if is_public is True and is_local is not True:
        return EgressScope.PUBLIC_ONLY

    # Unknown to _engine_bucket — it may be a registered retriever (private
    # KB). Classify via the retriever registry before falling back to BOTH.
    if is_public is None and is_local is None:
        retriever_local = _retriever_is_local(primary_engine, username)
        if retriever_local is True:
            return EgressScope.PRIVATE_ONLY
        if retriever_local is False:
            return EgressScope.PUBLIC_ONLY

    return EgressScope.BOTH


def context_from_snapshot(
    settings_snapshot: dict,
    primary_engine: str,
    *,
    username: Optional[str] = None,
    allow_dns: bool = True,
) -> EgressContext:
    """Construct the frozen ``EgressContext`` for a research run.

    Reads policy settings out of the snapshot exactly once at run-start.
    Default values preserve backward compatibility: missing keys yield
    the most permissive policy (``BOTH``, no local requirements).

    ``allow_dns=False`` makes ADAPTIVE resolution skip the synchronous
    getaddrinfo when classifying a URL-configurable primary engine — used by
    the advisory warning-banner render path so a settings-page load never
    blocks on DNS. Enforcement callers leave it True so classification is
    accurate. (Has no effect for non-ADAPTIVE scopes, which never resolve
    a URL.)
    """
    if settings_snapshot is None:
        raise ValueError("settings_snapshot is required")
    if not isinstance(settings_snapshot, dict):
        # Fail closed with the same ValueError contract callers already
        # handle (llm_config / research_service convert it to a hard policy
        # stop). A non-dict snapshot would otherwise crash deeper in
        # _get_setting_value with a bare AttributeError, which a broad
        # caller-side except could swallow into a permissive default.
        raise ValueError(
            f"settings_snapshot must be a dict, got "
            f"{type(settings_snapshot).__name__}"
        )

    scope_raw = _get_setting_value(
        settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
    )
    try:
        scope = EgressScope(str(scope_raw).lower())
    except ValueError as exc:
        # An unrecognised scope means the saved setting was corrupted
        # or tampered with — silently falling back to BOTH (the most
        # permissive scope) would mask policy violations rather than
        # surface them. Refuse the run instead so the operator notices.
        logger.bind(policy_audit=True).warning(
            "refusing to construct EgressContext from unknown scope",
            value=scope_raw,
        )
        raise PolicyDeniedError(
            Decision(False, "unknown_egress_scope"),
            target=str(scope_raw),
        ) from exc

    require_local_llm = _coerce_bool(
        _get_setting_value(
            settings_snapshot, "llm.require_local_endpoint", False
        )
    )
    require_local_embeddings = _coerce_bool(
        _get_setting_value(settings_snapshot, "embeddings.require_local", False)
    )

    raw_hostnames = _get_setting_value(
        settings_snapshot, "llm.allowed_local_hostnames", ()
    )
    if isinstance(raw_hostnames, (list, tuple)):
        local_hostnames = tuple(h for h in raw_hostnames if isinstance(h, str))
    else:
        local_hostnames = ()

    # ADAPTIVE resolves to a concrete scope by classifying the primary
    # engine: a concrete private primary => PRIVATE_ONLY, a concrete public
    # primary => PUBLIC_ONLY, an unclassifiable primary => BOTH.
    # The resolved scope is what the EgressContext stores and what
    # every downstream PEP enforces. Classification reuses _engine_bucket
    # via a throwaway BOTH-scoped context (its DNS cache is discarded).
    if scope == EgressScope.ADAPTIVE:
        scope = _resolve_adaptive_scope(
            primary_engine,
            settings_snapshot,
            username=username,
            local_hostnames=local_hostnames,
            allow_dns=allow_dns,
        )

    # PRIVATE_ONLY means "my data stays on this box." That guarantee only
    # holds if BOTH inference paths are local — a cloud LLM receives the
    # query + retrieved local chunks, and a cloud embedder receives the
    # whole corpus at ingest. So under PRIVATE_ONLY the require_local_*
    # flags are IMPLIED: scope overrides them, so a user who left them at
    # their default (False) can't silently exfiltrate via inference. This
    # also fires when ADAPTIVE resolved to PRIVATE_ONLY above.
    #
    # STRICT is deliberately NOT coupled here: it restricts the search
    # engine set to the primary only and is orthogonal to where inference
    # runs (a user may legitimately want single-engine search + a cloud
    # LLM). Callers that gate on ctx.require_local_* therefore get
    # scope-correct behaviour without a separate flag read.
    if scope == EgressScope.PRIVATE_ONLY:
        require_local_llm = True
        require_local_embeddings = True

    return EgressContext(
        scope=scope,
        primary_engine=primary_engine,
        require_local_llm=require_local_llm,
        require_local_embeddings=require_local_embeddings,
        local_hostnames=local_hostnames,
        username=username,
    )


def _get_setting_value(snapshot: dict, key: str, default):
    """Read a setting from the snapshot.

    Accepts both flat ({key: value}) and nested ({key: {"value": value}})
    schemas — both shapes appear in LDR's settings infrastructure.
    """
    raw = snapshot.get(key, default)
    return unwrap_setting(raw)


def resolve_run_primary_engine(
    settings_snapshot: Optional[dict], default: Optional[str] = None
) -> str:
    """Resolve a research run's primary search engine id from a snapshot.

    Single source of truth for "which engine drives this run's egress
    scope". Under the default ADAPTIVE scope the concrete scope is derived by
    classifying THIS primary, so run-scoped callers that build an
    ``EgressContext`` should derive the primary here. Otherwise two layers can
    resolve ADAPTIVE to different scopes and one silently under-enforces — the
    LangGraph tool-list filter once derived the primary from the engine CLASS
    name (a collection -> ``"libraryrag"`` -> unclassified -> BOTH) while the
    factory PEP derived it from ``search.tool`` (the real collection key ->
    PRIVATE_ONLY), so a public engine reached the agent and was then
    hard-denied mid-run (``scope_mismatch_private_only``).

    Reads ``search.tool`` (flat or ``{"value": ...}`` shaped). A blank-after-
    strip (``"  "``) or non-string (``5``, list/dict) value is treated as
    MISSING — it must not slip through the truthiness check and classify to the
    permissive BOTH scope.

    A run with NO configured primary is a wiring error, not a normal state.
    Silently substituting a default would set the egress scope from an engine
    the user never chose — and since the system default is public (``searxng``),
    that fails OPEN. So a missing/empty ``search.tool`` raises ``ValueError``
    unless the caller passes an explicit ``default``:

    * run-level callers (research_service, the strategy tool-list filter,
      ``filter_candidates_by_egress``) pass none and fail closed — the worker
      refuses the run; advisory filters degrade to unfiltered with the factory
      PEP still enforcing.
    * the search-engine factory passes ``default=engine_name`` because it is
      evaluating that one specific engine, so deriving the scope from it is
      meaningful rather than arbitrary.

    Raises:
        ValueError: ``search.tool`` is missing/blank/non-string and no truthy
            ``default`` is given.
    """
    primary = (
        _get_setting_value(settings_snapshot, "search.tool", None)
        if settings_snapshot
        else None
    )
    # Blank-after-strip or non-string is not a usable engine id → missing.
    primary = primary.strip() if isinstance(primary, str) else None
    if primary:
        return primary
    if default:
        return default
    raise ValueError(
        "no primary search engine configured: settings 'search.tool' is "
        "missing, blank, or not a string, so the run's egress scope cannot "
        "be resolved"
    )


def _coerce_bool(value) -> bool:
    """Coerce a setting value to bool, defensively.

    Strings ``"true"`` / ``"True"`` are True; anything else string-shaped
    is False. Real booleans pass through. The strict coercion prevents
    type confusion (e.g., the string ``"false"`` being truthy).
    """
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in ("true", "1", "yes", "on")
    return bool(value)
