learningcircuit--local-deep-research/tests/security/test_egress_policy.py

"""Unit tests for the egress policy module (Stage 1a)."""

from __future__ import annotations

from unittest.mock import patch

import pytest
from huggingface_hub import _CACHED_NO_EXIST

from local_deep_research.security.egress.policy import (
    Decision,
    EgressContext,
    EgressScope,
    MAX_DENIED_FETCHES_PER_RUN,
    PolicyDeniedError,
    context_from_snapshot,
    evaluate_engine,
    evaluate_embeddings,
    evaluate_llm_endpoint,
    evaluate_url,
    filter_engines_by_egress,
    resolve_run_primary_engine,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def make_ctx(
    scope: EgressScope = EgressScope.BOTH,
    primary: str = "arxiv",
    require_local_llm: bool = False,
    require_local_embeddings: bool = False,
    local_hostnames=(),
) -> EgressContext:
    return EgressContext(
        scope=scope,
        primary_engine=primary,
        require_local_llm=require_local_llm,
        require_local_embeddings=require_local_embeddings,
        local_hostnames=tuple(local_hostnames),
    )


# ---------------------------------------------------------------------------
# context_from_snapshot
# ---------------------------------------------------------------------------


def test_default_scope_constant_matches_registry():
    """DEFAULT_EGRESS_SCOPE (the code-side fallback every reader imports)
    must equal the registered default in defaults/default_settings.json.
    This single test replaces scattered fallback-string assertions and is
    what prevents the code/registry drift ('both' vs 'adaptive') that
    motivated this fix from recurring."""
    import json

    from local_deep_research.defaults import DEFAULTS_DIR
    from local_deep_research.security.egress.policy import (
        DEFAULT_EGRESS_SCOPE,
    )

    path = DEFAULTS_DIR / "default_settings.json"
    assert path.exists()
    with open(path, encoding="utf-8-sig") as f:
        registry = json.load(f)
    assert registry["policy.egress_scope"]["value"] == DEFAULT_EGRESS_SCOPE
    # The constant must also be a valid enum member (sanity).
    assert EgressScope(DEFAULT_EGRESS_SCOPE)


def test_context_from_snapshot_defaults_to_adaptive():
    """A missing policy.egress_scope falls back to the REGISTERED default
    (adaptive), matching what users with a settings DB already get: a
    public primary resolves PUBLIC_ONLY, a meta-picker primary resolves
    BOTH (the permissive pre-policy behavior)."""
    ctx = context_from_snapshot({}, primary_engine="arxiv")
    assert ctx.scope == EgressScope.PUBLIC_ONLY
    assert ctx.require_local_llm is False
    assert ctx.require_local_embeddings is False
    assert ctx.local_hostnames == ()

    ctx_meta = context_from_snapshot({}, primary_engine="auto")
    assert ctx_meta.scope == EgressScope.BOTH


def _scope_snapshot(scope, tool):
    return {
        "policy.egress_scope": {"value": scope},
        "search.tool": {"value": tool},
    }


def test_adaptive_stray_auto_primary_resolves_to_both():
    # "auto" is no longer a registered engine (meta-pickers were removed); a
    # stray value left in the DB is unclassifiable and falls through to BOTH.
    ctx = context_from_snapshot(
        _scope_snapshot("adaptive", "auto"), primary_engine="auto"
    )
    assert ctx.scope == EgressScope.BOTH
    assert ctx.require_local_llm is False


def test_adaptive_public_primary_resolves_to_public_only():
    ctx = context_from_snapshot(
        _scope_snapshot("adaptive", "arxiv"), primary_engine="arxiv"
    )
    assert ctx.scope == EgressScope.PUBLIC_ONLY
    # public scope does NOT force local inference
    assert ctx.require_local_llm is False


def test_adaptive_private_primary_resolves_to_private_only_and_forces_local():
    # 'library' is the always-private aggregate engine.
    ctx = context_from_snapshot(
        _scope_snapshot("adaptive", "library"), primary_engine="library"
    )
    assert ctx.scope == EgressScope.PRIVATE_ONLY
    # private primary under adaptive must force local inference (coupling)
    assert ctx.require_local_llm is True
    assert ctx.require_local_embeddings is True


def test_adaptive_falls_back_to_both_on_classification_error():
    # An unknown concrete engine can't be classified → BOTH (permissive
    # fallback, never a hard fail).
    ctx = context_from_snapshot(
        _scope_snapshot("adaptive", "totally_unknown_engine"),
        primary_engine="totally_unknown_engine",
    )
    assert ctx.scope == EgressScope.BOTH


def test_public_collection_allowed_under_public_only_via_metadata():
    """A collection flagged public classifies as a public engine: allowed
    under PUBLIC_ONLY, and (mirror) a private one is denied."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY, primary="collection_abc")
    pub = evaluate_engine(
        "collection_abc",
        ctx,
        settings_snapshot={"policy.egress_scope": {"value": "public_only"}},
        metadata={"is_public": True},
    )
    assert pub.allowed is True
    priv = evaluate_engine(
        "collection_abc",
        ctx,
        settings_snapshot={"policy.egress_scope": {"value": "public_only"}},
        metadata={"is_public": False},
    )
    assert priv.allowed is False
    assert priv.reason == "scope_mismatch_public_only"


def test_private_collection_allowed_under_private_only_via_metadata():
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY, primary="collection_abc")
    priv = evaluate_engine(
        "collection_abc",
        ctx,
        settings_snapshot={"policy.egress_scope": {"value": "private_only"}},
        metadata={"is_public": False},
    )
    assert priv.allowed is True
    # A collection is ALWAYS a local KB, so a PUBLIC collection is ALSO usable
    # under PRIVATE_ONLY (is_public is additive, not exclusive — the content is
    # local regardless; the flag only ADDS public-scope/cloud eligibility).
    pub = evaluate_engine(
        "collection_abc",
        ctx,
        settings_snapshot={"policy.egress_scope": {"value": "private_only"}},
        metadata={"is_public": True},
    )
    assert pub.allowed is True


def test_public_collection_is_local_and_public_every_scope():
    """A public collection is classified (is_public=True, is_local=True): it is
    allowed under PUBLIC_ONLY, PRIVATE_ONLY and BOTH. A private collection is
    local-only (excluded from PUBLIC_ONLY)."""
    pub_meta = {"is_public": True}
    priv_meta = {"is_public": False}
    for scope in (
        EgressScope.PUBLIC_ONLY,
        EgressScope.PRIVATE_ONLY,
        EgressScope.BOTH,
    ):
        ctx = make_ctx(scope=scope, primary="collection_abc")
        assert evaluate_engine(
            "collection_abc", ctx, settings_snapshot={}, metadata=pub_meta
        ).allowed, f"public collection should be allowed under {scope}"
    # Private collection: denied only under PUBLIC_ONLY.
    ctx_pub = make_ctx(scope=EgressScope.PUBLIC_ONLY, primary="collection_abc")
    assert not evaluate_engine(
        "collection_abc", ctx_pub, settings_snapshot={}, metadata=priv_meta
    ).allowed
    for scope in (EgressScope.PRIVATE_ONLY, EgressScope.BOTH):
        ctx = make_ctx(scope=scope, primary="collection_abc")
        assert evaluate_engine(
            "collection_abc", ctx, settings_snapshot={}, metadata=priv_meta
        ).allowed, f"private collection should be allowed under {scope}"


def test_collection_without_metadata_defaults_private_via_db_lookup():
    """Without metadata, evaluate_engine resolves the collection's is_public
    from the DB; a lookup failure fails closed to private (local)."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY, primary="collection_xyz")
    # No DB available in this unit test → _resolve_collection_is_public
    # returns False (private) → allowed under PRIVATE_ONLY.
    decision = evaluate_engine(
        "collection_xyz",
        ctx,
        settings_snapshot={"policy.egress_scope": {"value": "private_only"}},
    )
    assert decision.allowed is True


def test_context_from_snapshot_reads_nested_value_dicts():
    snapshot = {
        "policy.egress_scope": {"value": "strict"},
        "llm.require_local_endpoint": {"value": True},
    }
    ctx = context_from_snapshot(snapshot, primary_engine="arxiv")
    assert ctx.scope == EgressScope.STRICT
    assert ctx.require_local_llm is True


def test_context_strict_with_stray_meta_name_builds_strict_context():
    # Meta-pickers were removed: STRICT + a stray "auto" primary no longer
    # raises ValueError at context construction. The context stays STRICT and
    # the stray engine itself is denied downstream (engine_unknown).
    snapshot = {"policy.egress_scope": "strict"}
    ctx = context_from_snapshot(snapshot, primary_engine="auto")
    assert ctx.scope == EgressScope.STRICT
    decision = evaluate_engine("auto", ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "engine_unknown"


def test_context_unknown_scope_raises_policy_denied():
    # N8 (Round 6): unknown scope is fail-closed rather than silently
    # falling back to BOTH. Silent fallback would mask config corruption
    # and effectively disable the policy whenever the saved value was
    # tampered with or migrated incorrectly.
    with pytest.raises(PolicyDeniedError) as excinfo:
        context_from_snapshot(
            {"policy.egress_scope": "nonsense"}, primary_engine="arxiv"
        )
    assert excinfo.value.decision.reason == "unknown_egress_scope"


def test_context_string_false_coerces_correctly():
    # Type-confusion guard: the string "false" must not be truthy.
    ctx = context_from_snapshot(
        {"llm.require_local_endpoint": "false"}, primary_engine="arxiv"
    )
    assert ctx.require_local_llm is False


def test_context_string_true_coerces_to_true():
    ctx = context_from_snapshot(
        {"llm.require_local_endpoint": "true"}, primary_engine="arxiv"
    )
    assert ctx.require_local_llm is True


# ---------------------------------------------------------------------------
# evaluate_engine — STRICT semantics
# ---------------------------------------------------------------------------


def test_evaluate_engine_strict_primary_match():
    ctx = make_ctx(scope=EgressScope.STRICT, primary="arxiv")
    decision = evaluate_engine("arxiv", ctx, settings_snapshot={})
    assert decision.allowed
    assert decision.reason == "allowed"


def test_evaluate_engine_strict_non_primary_denied():
    ctx = make_ctx(scope=EgressScope.STRICT, primary="arxiv")
    decision = evaluate_engine("pubmed", ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "strict_not_primary"


def test_evaluate_engine_strict_with_removed_meta_engine():
    # The removed auto/meta/parallel names can never be permitted under STRICT.
    ctx = make_ctx(scope=EgressScope.STRICT, primary="arxiv")
    decision = evaluate_engine("auto", ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "strict_not_primary"


# ---------------------------------------------------------------------------
# evaluate_engine — public/private bucket
# ---------------------------------------------------------------------------


def test_evaluate_engine_public_only_blocks_local_engine():
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    # paperless is is_local=True with url_setting; with no URL in snapshot
    # the static is_local flag applies, so PUBLIC_ONLY rejects it.
    decision = evaluate_engine("paperless", ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_public_only"


def test_evaluate_engine_private_only_blocks_public_engine():
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    decision = evaluate_engine("arxiv", ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_private_only"


def test_evaluate_engine_both_allows_public():
    ctx = make_ctx(scope=EgressScope.BOTH)
    decision = evaluate_engine("arxiv", ctx, settings_snapshot={})
    assert decision.allowed


def test_evaluate_engine_both_allows_private():
    ctx = make_ctx(scope=EgressScope.BOTH)
    decision = evaluate_engine("paperless", ctx, settings_snapshot={})
    assert decision.allowed


def test_evaluate_engine_no_snapshot_fails_closed():
    ctx = make_ctx()
    decision = evaluate_engine("arxiv", ctx, settings_snapshot=None)
    assert not decision.allowed
    assert decision.reason == "no_snapshot"


def test_evaluate_engine_unknown_name_fails_closed():
    ctx = make_ctx()
    decision = evaluate_engine(
        "totally_made_up_engine", ctx, settings_snapshot={}
    )
    assert not decision.allowed
    assert decision.reason == "engine_unknown"


# ---------------------------------------------------------------------------
# evaluate_engine — newly-classified engines
# ---------------------------------------------------------------------------


def test_evaluate_engine_github_is_public():
    """Regression: github engine must be explicitly classified is_public=True."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    decision = evaluate_engine("github", ctx, settings_snapshot={})
    assert decision.allowed


def test_evaluate_engine_paperless_is_local():
    """Regression: paperless engine must be classified is_local=True."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    # Without a url_setting value, the static is_local flag applies.
    decision = evaluate_engine("paperless", ctx, settings_snapshot={})
    assert decision.allowed


# ---------------------------------------------------------------------------
# evaluate_engine — dynamic URL classification
# ---------------------------------------------------------------------------


def test_evaluate_engine_searxng_localhost_denied_under_private_only():
    """SearXNG is is_public=True regardless of where it's hosted — a local
    SearXNG still queries the internet, so PRIVATE_ONLY must deny it."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    snapshot = {
        "search.engine.web.searxng.default_params.instance_url": "http://localhost:8080"
    }
    decision = evaluate_engine("searxng", ctx, settings_snapshot=snapshot)
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_private_only"


def test_evaluate_engine_searxng_localhost_allowed_under_public_only():
    """SearXNG is is_public=True, so PUBLIC_ONLY allows it even when
    hosted on localhost — the engine-selection gate uses static flags."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    snapshot = {
        "search.engine.web.searxng.default_params.instance_url": "http://localhost:8080"
    }
    decision = evaluate_engine("searxng", ctx, settings_snapshot=snapshot)
    assert decision.allowed


def test_evaluate_engine_searxng_remote_classified_public():
    """SearXNG pointed at a public host should fail PRIVATE_ONLY."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    snapshot = {
        "search.engine.web.searxng.default_params.instance_url": "https://searx.example.com"
    }
    # Patch DNS so the public hostname doesn't actually resolve over the network
    # during tests.
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=False,
    ):
        decision = evaluate_engine("searxng", ctx, settings_snapshot=snapshot)
    assert not decision.allowed


# ---------------------------------------------------------------------------
# Engine nature classification — static class flags are authoritative
# ---------------------------------------------------------------------------


# Engines that can be loaded in CI (no optional deps needed).
_PUBLIC_ENGINES = [
    "brave",
    "ddg",
    "exa",
    "github",
    "google_pse",
    "guardian",
    "gutenberg",
    "mojeek",
    "nasa_ads",
    "openalex",
    "openlibrary",
    "pubchem",
    "pubmed",
    "scaleserp",
    "searxng",
    "semantic_scholar",
    "serpapi",
    "serper",
    "stackexchange",
    "tavily",
    "wayback",
    "wikinews",
    "wikipedia",
    "zenodo",
]

_LOCAL_ENGINES = [
    "paperless",
]


@pytest.mark.parametrize("engine_name", _PUBLIC_ENGINES)
def test_public_engine_denied_under_private_only(engine_name):
    """Every is_public=True engine must be denied under PRIVATE_ONLY,
    regardless of its configured URL.  Engine nature (queries the
    internet) is determined by the Python class flag, not by where the
    engine happens to be hosted."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_private_only"


@pytest.mark.parametrize("engine_name", _PUBLIC_ENGINES)
def test_public_engine_allowed_under_public_only(engine_name):
    """Every is_public=True engine must pass PUBLIC_ONLY."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert decision.allowed


@pytest.mark.parametrize("engine_name", _PUBLIC_ENGINES)
def test_public_engine_allowed_under_both(engine_name):
    """Every is_public=True engine must pass BOTH."""
    ctx = make_ctx(scope=EgressScope.BOTH)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert decision.allowed


@pytest.mark.parametrize("engine_name", _LOCAL_ENGINES)
def test_local_engine_allowed_under_private_only(engine_name):
    """Every is_local=True engine must pass PRIVATE_ONLY."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert decision.allowed


@pytest.mark.parametrize("engine_name", _LOCAL_ENGINES)
def test_local_engine_denied_under_public_only(engine_name):
    """Every is_local=True engine (without is_public) must be denied under
    PUBLIC_ONLY."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_public_only"


@pytest.mark.parametrize("engine_name", _LOCAL_ENGINES)
def test_local_engine_allowed_under_both(engine_name):
    """Every is_local=True engine must pass BOTH."""
    ctx = make_ctx(scope=EgressScope.BOTH)
    decision = evaluate_engine(engine_name, ctx, settings_snapshot={})
    assert decision.allowed


def test_searxng_local_url_still_public_nature():
    """SearXNG with a localhost URL must still be classified as public —
    it proxies to internet search engines regardless of where it's hosted."""
    ctx_pub = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    ctx_priv = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    snapshot = {
        "search.engine.web.searxng.default_params.instance_url": "http://localhost:8080"
    }
    assert evaluate_engine(
        "searxng", ctx_pub, settings_snapshot=snapshot
    ).allowed
    assert not evaluate_engine(
        "searxng", ctx_priv, settings_snapshot=snapshot
    ).allowed


def test_paperless_local_url_still_local_nature():
    """Paperless with a localhost URL is is_local=True — allowed under
    PRIVATE_ONLY, denied under PUBLIC_ONLY."""
    ctx_pub = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    ctx_priv = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    snapshot = {
        "search.engine.web.paperless.default_params.api_url": "http://localhost:8930"
    }
    assert not evaluate_engine(
        "paperless", ctx_pub, settings_snapshot=snapshot
    ).allowed
    assert evaluate_engine(
        "paperless", ctx_priv, settings_snapshot=snapshot
    ).allowed


def test_paperless_public_host_denied_under_private_only():
    """Fail-up URL override: a local-nature engine whose configured URL
    points at a PUBLIC host is reclassified public — querying it sends the
    user's queries off the box, so PRIVATE_ONLY denies it at selection
    time (not just at the audit-hook socket net)."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    snapshot = {
        # Public literal IP so classification needs no real DNS in CI.
        "search.engine.web.paperless.default_params.api_url": "http://93.184.216.34:8930"
    }
    decision = evaluate_engine("paperless", ctx, settings_snapshot=snapshot)
    assert not decision.allowed
    assert decision.reason == "scope_mismatch_private_only"


def test_paperless_public_host_allowed_under_public_only():
    """The fail-up reclassification makes a remote-hosted local-data
    engine eligible under PUBLIC_ONLY (pre-static-flags behavior)."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    snapshot = {
        "search.engine.web.paperless.default_params.api_url": "http://93.184.216.34:8930"
    }
    decision = evaluate_engine("paperless", ctx, settings_snapshot=snapshot)
    assert decision.allowed


def test_paperless_public_host_adaptive_resolves_public_only():
    """ADAPTIVE with a remote-hosted Paperless primary must NOT resolve to
    PRIVATE_ONLY (which would imply 'nothing leaves the box' while every
    query goes to a public host)."""
    snap = {
        "policy.egress_scope": {"value": "adaptive"},
        "search.tool": {"value": "paperless"},
        "search.engine.web.paperless.default_params.api_url": "http://93.184.216.34:8930",
    }
    ctx = context_from_snapshot(snap, primary_engine="paperless")
    assert ctx.scope == EgressScope.PUBLIC_ONLY


def test_searxng_adaptive_resolves_to_public_only():
    """ADAPTIVE with SearXNG (is_public=True) as primary must resolve to
    PUBLIC_ONLY even when the instance URL points to localhost."""
    snap = {
        "policy.egress_scope": {"value": "adaptive"},
        "search.tool": {"value": "searxng"},
        "search.engine.web.searxng.default_params.instance_url": "http://localhost:8080",
    }
    ctx = context_from_snapshot(snap, primary_engine="searxng")
    assert ctx.scope == EgressScope.PUBLIC_ONLY
    assert ctx.require_local_llm is False


def test_paperless_adaptive_resolves_to_private_only():
    """ADAPTIVE with Paperless (is_local=True) as primary must resolve to
    PRIVATE_ONLY and force local inference."""
    snap = {
        "policy.egress_scope": {"value": "adaptive"},
        "search.tool": {"value": "paperless"},
    }
    ctx = context_from_snapshot(snap, primary_engine="paperless")
    assert ctx.scope == EgressScope.PRIVATE_ONLY
    assert ctx.require_local_llm is True
    assert ctx.require_local_embeddings is True


# ---------------------------------------------------------------------------
# filter_engines_by_egress
# ---------------------------------------------------------------------------


def test_filter_engines_removes_public_under_private_only():
    """filter_engines_by_egress must strip all public engines under
    PRIVATE_ONLY."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    result = filter_engines_by_egress(
        ["wikipedia", "github", "searxng", "paperless"],
        ctx,
        settings_snapshot={},
    )
    assert result == ["paperless"]


def test_filter_engines_removes_local_under_public_only():
    """filter_engines_by_egress must strip local engines under PUBLIC_ONLY."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    result = filter_engines_by_egress(
        ["wikipedia", "github", "searxng", "paperless"],
        ctx,
        settings_snapshot={},
    )
    assert "paperless" not in result
    assert "wikipedia" in result


def test_filter_engines_keeps_all_under_both():
    """filter_engines_by_egress keeps everything under BOTH."""
    ctx = make_ctx(scope=EgressScope.BOTH)
    result = filter_engines_by_egress(
        ["wikipedia", "paperless"],
        ctx,
        settings_snapshot={},
    )
    assert "wikipedia" in result
    assert "paperless" in result


def test_filter_engines_unknown_engine_kept():
    """Names unknown to the static registry are KEPT by the advisory
    pre-filter: they may be retriever-backed or dynamically injected
    engines that the factory PEP evaluates via its own path. The
    pre-filter must never be stricter than the enforcement point it
    fronts — the factory still denies anything truly disallowed."""
    ctx = make_ctx(scope=EgressScope.BOTH)
    result = filter_engines_by_egress(
        ["wikipedia", "totally_made_up"],
        ctx,
        settings_snapshot={},
    )
    assert result == ["wikipedia", "totally_made_up"]


def test_filter_engines_strict_keeps_only_primary_and_unknown():
    """Under STRICT, non-primary registry engines are stripped
    (strict_not_primary, matching the factory); the primary survives.
    NB: under STRICT an unknown name is also stripped — the STRICT gate
    fires before the registry lookup, exactly as it does in the factory."""
    ctx = make_ctx(scope=EgressScope.STRICT, primary="paperless")
    result = filter_engines_by_egress(
        ["paperless", "wikipedia", "totally_made_up"],
        ctx,
        settings_snapshot={},
    )
    assert result == ["paperless"]


def test_filter_candidates_helper_strips_by_snapshot_scope():
    """filter_candidates_by_egress does the full snapshot plumbing:
    scope/primary extraction, context build, filter."""
    from local_deep_research.security.egress.policy import (
        filter_candidates_by_egress,
    )

    snap = {
        "policy.egress_scope": {"value": "private_only"},
        "search.tool": {"value": "paperless"},
    }
    result = filter_candidates_by_egress(
        ["wikipedia", "paperless", "totally_made_up"], snap
    )
    assert "wikipedia" not in result
    assert "paperless" in result
    assert "totally_made_up" in result


def test_filter_candidates_helper_noop_without_snapshot_or_under_both():
    from local_deep_research.security.egress.policy import (
        filter_candidates_by_egress,
    )

    names = ["wikipedia", "paperless"]
    assert filter_candidates_by_egress(names, None) == names
    assert filter_candidates_by_egress(names, {}) == names
    snap = {"policy.egress_scope": {"value": "both"}}
    assert filter_candidates_by_egress(names, snap) == names


def test_filter_candidates_helper_failopen_on_corrupt_scope():
    """A corrupted scope string must not break engine selection — the
    helper returns the list unchanged (the unrecognized value falls out
    of the scope gate before any context is built) and the factory PEP
    remains the enforcement point."""
    from local_deep_research.security.egress.policy import (
        filter_candidates_by_egress,
    )

    names = ["wikipedia", "paperless"]
    snap = {"policy.egress_scope": {"value": "garbage_scope"}}
    assert filter_candidates_by_egress(names, snap) == names


def test_filter_engines_strips_scope_denials_keeps_unknown():
    """Under PRIVATE_ONLY a public engine is stripped (active scope
    denial) while an unknown name survives for the factory to judge."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    result = filter_engines_by_egress(
        ["wikipedia", "totally_made_up", "paperless"],
        ctx,
        settings_snapshot={},
    )
    assert "wikipedia" not in result
    assert "totally_made_up" in result
    assert "paperless" in result


# ---------------------------------------------------------------------------
# evaluate_llm_endpoint
# ---------------------------------------------------------------------------


def test_evaluate_llm_endpoint_no_local_requirement_allows_cloud():
    ctx = make_ctx(require_local_llm=False)
    decision = evaluate_llm_endpoint("openai", ctx, settings_snapshot={})
    assert decision.allowed


def test_evaluate_llm_endpoint_require_local_blocks_cloud_providers():
    ctx = make_ctx(require_local_llm=True)
    for provider in ("openai", "anthropic", "google", "openrouter"):
        decision = evaluate_llm_endpoint(provider, ctx, settings_snapshot={})
        assert not decision.allowed, (
            f"{provider} should be blocked under require_local_llm"
        )


def test_evaluate_llm_endpoint_local_ollama_allowed():
    ctx = make_ctx(require_local_llm=True)
    decision = evaluate_llm_endpoint("ollama", ctx, settings_snapshot={})
    # No URL override → assumes localhost default.
    assert decision.allowed


def test_evaluate_llm_endpoint_user_registered_llm_allowed():
    """A user-registered in-process LLM (programmatic API ``llms={...}``)
    must be allowed under require_local_llm — it has no endpoint to
    classify and the audit hook backstops stray sockets. Regression test
    for the mock-LLM example breaking after ADAPTIVE retriever-primary
    runs began resolving to PRIVATE_ONLY."""
    from local_deep_research.llm.llm_registry import (
        register_llm,
        unregister_llm,
    )

    ctx = make_ctx(require_local_llm=True)
    register_llm("egress_test_mock_llm", lambda **kwargs: None)
    try:
        decision = evaluate_llm_endpoint(
            "egress_test_mock_llm", ctx, settings_snapshot={}
        )
        assert decision.allowed
        assert decision.reason == "user_registered_llm"
    finally:
        unregister_llm("egress_test_mock_llm")


def test_evaluate_llm_endpoint_registered_name_shadowing_cloud_still_blocked():
    """Registering a custom LLM under a built-in cloud name must NOT
    bypass the cloud-provider gate."""
    from local_deep_research.llm.llm_registry import (
        get_llm_from_registry,
        register_llm,
        unregister_llm,
    )

    ctx = make_ctx(require_local_llm=True)
    original = get_llm_from_registry("openai")
    register_llm("openai", lambda **kwargs: None)
    try:
        decision = evaluate_llm_endpoint("openai", ctx, settings_snapshot={})
        assert not decision.allowed
        assert decision.reason == "provider_cloud_only"
    finally:
        # Restore the auto-registered built-in entry rather than leaving
        # the registry polluted for later tests in this process.
        if original is not None:
            register_llm("openai", original)
        else:
            unregister_llm("openai")


def test_evaluate_llm_endpoint_unregistered_unknown_provider_still_blocked():
    """An unknown provider that is NOT in the registry keeps failing
    closed with provider_url_unset."""
    ctx = make_ctx(require_local_llm=True)
    decision = evaluate_llm_endpoint(
        "totally_unknown_provider", ctx, settings_snapshot={}
    )
    assert not decision.allowed
    assert decision.reason == "provider_url_unset"


def test_evaluate_llm_endpoint_ollama_pointed_remote_blocked():
    ctx = make_ctx(require_local_llm=True)
    snapshot = {"llm.ollama.url": "https://remote-ollama.example.com"}
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=False,
    ):
        decision = evaluate_llm_endpoint(
            "ollama", ctx, settings_snapshot=snapshot
        )
    assert not decision.allowed


# ---------------------------------------------------------------------------
# evaluate_embeddings
# ---------------------------------------------------------------------------


def test_evaluate_embeddings_no_requirement_allows_openai():
    ctx = make_ctx(require_local_embeddings=False)
    decision = evaluate_embeddings("openai", ctx, settings_snapshot={})
    assert decision.allowed


def test_evaluate_embeddings_require_local_allows_sentence_transformers():
    ctx = make_ctx(require_local_embeddings=True)
    decision = evaluate_embeddings(
        "sentence_transformers", ctx, settings_snapshot={}
    )
    assert decision.allowed


def test_evaluate_embeddings_require_local_blocks_openai_cloud():
    ctx = make_ctx(require_local_embeddings=True)
    decision = evaluate_embeddings("openai", ctx, settings_snapshot={})
    assert not decision.allowed


def test_evaluate_embeddings_openai_with_local_base_url_allowed():
    ctx = make_ctx(require_local_embeddings=True)
    snapshot = {"embeddings.openai.base_url": "http://localhost:1234/v1"}
    decision = evaluate_embeddings("openai", ctx, settings_snapshot=snapshot)
    assert decision.allowed


# ---------------------------------------------------------------------------
# evaluate_url
# ---------------------------------------------------------------------------


def test_evaluate_url_rejects_dangerous_scheme():
    ctx = make_ctx()
    assert not evaluate_url("javascript:alert(1)", ctx).allowed
    assert not evaluate_url("data:text/html,<script>", ctx).allowed


def test_evaluate_url_rejects_malformed():
    ctx = make_ctx()
    assert not evaluate_url("", ctx).allowed
    assert not evaluate_url("not-a-url", ctx).allowed


def test_evaluate_url_strict_allows_private_host():
    ctx = make_ctx(scope=EgressScope.STRICT)
    assert evaluate_url("http://127.0.0.1:8080/", ctx).allowed


def test_evaluate_url_strict_blocks_public_host():
    ctx = make_ctx(scope=EgressScope.STRICT)
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=False,
    ):
        decision = evaluate_url("https://example.com/", ctx)
    assert not decision.allowed


def test_evaluate_url_private_only_blocks_public():
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=False,
    ):
        decision = evaluate_url("https://example.com/", ctx)
    assert not decision.allowed


def test_evaluate_url_denial_quota_exhaustion():
    """After MAX_DENIED_FETCHES_PER_RUN denials, even a legal URL fails closed.

    This blocks a malicious indexed document from looping the agent through
    hundreds of denied fetches and inflating the audit log.
    """
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    # Pre-fill the counter to the threshold.
    ctx._fetch_denial_count["count"] = MAX_DENIED_FETCHES_PER_RUN
    decision = evaluate_url("http://127.0.0.1/", ctx)
    assert not decision.allowed
    assert decision.reason == "denial_quota_exceeded"


# ---------------------------------------------------------------------------
# EgressContext: frozen + mutable internals
# ---------------------------------------------------------------------------


def test_egress_context_is_frozen_for_user_fields():
    ctx = make_ctx()
    with pytest.raises(Exception):
        ctx.scope = EgressScope.STRICT  # type: ignore[misc]


def test_egress_context_mutable_internals_work():
    """The DNS cache and denial counter live inside dict wrappers so they
    can mutate even on a frozen=True dataclass. This is the R9-01 pattern.
    """
    ctx = make_ctx()
    ctx._dns_cache["example.com"] = False
    assert ctx._dns_cache["example.com"] is False
    ctx._fetch_denial_count["count"] += 1
    assert ctx._fetch_denial_count["count"] == 1


# ---------------------------------------------------------------------------
# PolicyDeniedError
# ---------------------------------------------------------------------------


def test_policy_denied_error_carries_decision_and_target():
    err = PolicyDeniedError(
        decision=Decision(allowed=False, reason="strict_not_primary"),
        target="pubmed",
    )
    assert err.decision.reason == "strict_not_primary"
    assert err.target == "pubmed"
    assert "policy_denied" in str(err)


# ---------------------------------------------------------------------------
# Factory PEP integration (Stage 1a's end-to-end check)
# ---------------------------------------------------------------------------


def test_factory_raises_policy_denied_under_strict_non_primary():
    """End-to-end: factory gate blocks an engine that doesn't match the
    user's STRICT primary. This is the original LangGraph-silent-expansion
    bug closed at its source.
    """
    from local_deep_research.web_search_engines.search_engine_factory import (
        create_search_engine,
    )

    # User picks arxiv as primary, scope=STRICT. The factory must reject any
    # attempt to instantiate pubmed.
    snapshot = {
        "policy.egress_scope": "strict",
        "search.tool": "arxiv",
        # Minimal search_engines_config entry so the engine resolves.
        "search.engine.web.pubmed.use_in_auto_search": True,
    }
    with pytest.raises(PolicyDeniedError) as exc_info:
        create_search_engine("pubmed", llm=None, settings_snapshot=snapshot)
    assert exc_info.value.decision.reason == "strict_not_primary"


def test_get_llm_blocks_cloud_under_require_local():
    """End-to-end: get_llm() must raise PolicyDeniedError when the user
    has require_local_endpoint=true and asks for a cloud provider.
    """
    from local_deep_research.config.llm_config import get_llm

    snapshot = {
        "llm.require_local_endpoint": True,
        "llm.provider": "openai",
        "llm.model": "gpt-4o",
        "search.tool": "arxiv",
        # API key needed so we get past openai-base's own validation —
        # the policy PEP fires before that anyway.
        "llm.openai.api_key": "sk-test-not-real",
    }
    with pytest.raises(PolicyDeniedError) as exc_info:
        get_llm(
            provider="openai", model_name="gpt-4o", settings_snapshot=snapshot
        )
    assert exc_info.value.target == "openai"
    assert "cloud" in exc_info.value.decision.reason


def test_get_llm_allows_cloud_without_require_local():
    """Without require_local_endpoint, cloud providers are permitted
    (policy is a separate orthogonal control — the default is permissive).
    """
    from local_deep_research.config.llm_config import get_llm

    snapshot = {
        "llm.require_local_endpoint": False,
        "llm.provider": "openai",
        "llm.model": "gpt-4o",
        "search.tool": "arxiv",
        "llm.openai.api_key": "sk-test-not-real",
    }
    # Don't assert on successful return — OpenAI client construction
    # might fail for unrelated reasons. We only care that PolicyDeniedError
    # is NOT raised by our gate.
    try:
        get_llm(
            provider="openai",
            model_name="gpt-4o",
            settings_snapshot=snapshot,
        )
    except PolicyDeniedError:
        pytest.fail(
            "policy should permit cloud LLM when require_local_endpoint=False"
        )
    except Exception:
        pass  # unrelated failure is fine


@pytest.mark.parametrize(
    "provider",
    sorted(
        __import__(
            "local_deep_research.security.egress.policy",
            fromlist=["_CLOUD_LLM_PROVIDERS"],
        )._CLOUD_LLM_PROVIDERS
    )
    + ["openai_endpoint", "groq", "mistral", "cohere"],
)
def test_get_llm_no_snapshot_blocks_non_local_providers(provider):
    """Snapshot-less callers must not silently instantiate any provider
    outside the known-local allow-list. Covers all enumerated cloud
    providers PLUS ambiguous (openai_endpoint) and hypothetical
    future ones (groq/mistral/cohere) — the allow-list approach is
    deliberately tight so a new cloud provider is refused until it's
    explicitly classified.
    """
    from local_deep_research.config.llm_config import get_llm

    with pytest.raises(PolicyDeniedError) as exc_info:
        get_llm(provider=provider, model_name="x")
    assert exc_info.value.target == provider
    assert exc_info.value.decision.reason == "no_snapshot_for_provider"


@pytest.mark.parametrize("provider", ["ollama", "lmstudio", "llamacpp"])
def test_get_llm_no_snapshot_allows_local_providers(provider):
    """The snapshot-less gate is an allow-list. Known-local providers
    pass through; downstream failures are unrelated to policy.
    """
    from local_deep_research.config.llm_config import get_llm

    try:
        get_llm(provider=provider, model_name="x")
    except PolicyDeniedError as exc:
        pytest.fail(
            f"no-snapshot policy should permit {provider} (got "
            f"{exc.decision.reason})"
        )
    except Exception:
        # Non-policy failures (missing model config, no reachable
        # endpoint, etc.) are out of scope for this test.
        pass


def test_get_llm_no_snapshot_no_provider_is_noop():
    """When neither snapshot nor provider is set, the gate does nothing
    (preserves legacy "look up provider from settings" path).
    """
    from local_deep_research.config.llm_config import get_llm

    try:
        get_llm(provider=None, model_name=None)
    except PolicyDeniedError:
        pytest.fail("gate must not fire when no provider is given")
    except Exception:
        pass  # downstream config errors are fine


# ---------------------------------------------------------------------------
# Warning banners
# ---------------------------------------------------------------------------


def test_warning_public_egress_fires_when_scope_is_both():
    from local_deep_research.security.egress.warnings import (
        check_public_egress_enabled,
    )

    warning = check_public_egress_enabled("both", acknowledged=False)
    assert warning is not None
    assert warning["type"] == "public_egress_enabled"


def test_warning_public_egress_suppressed_under_private_only():
    from local_deep_research.security.egress.warnings import (
        check_public_egress_enabled,
    )

    assert (
        check_public_egress_enabled("private_only", acknowledged=False) is None
    )


def test_warning_public_egress_suppressed_when_acknowledged():
    """The fresh-install ack flag silences the banner triplet."""
    from local_deep_research.security.egress.warnings import (
        check_public_egress_enabled,
    )

    assert check_public_egress_enabled("both", acknowledged=True) is None


def test_warning_cloud_llm_fires_for_openai_without_require_local():
    from local_deep_research.security.egress.warnings import (
        check_cloud_llm_enabled,
    )

    warning = check_cloud_llm_enabled(
        "openai", require_local_endpoint=False, acknowledged=False
    )
    assert warning is not None
    assert warning["type"] == "cloud_llm_enabled"


def test_warning_cloud_llm_suppressed_for_local_provider():
    from local_deep_research.security.egress.warnings import (
        check_cloud_llm_enabled,
    )

    # Ollama is local-default — no warning even without the toggle.
    assert (
        check_cloud_llm_enabled(
            "ollama", require_local_endpoint=False, acknowledged=False
        )
        is None
    )


def test_warning_cloud_llm_suppressed_when_require_local_set():
    from local_deep_research.security.egress.warnings import (
        check_cloud_llm_enabled,
    )

    assert (
        check_cloud_llm_enabled(
            "openai", require_local_endpoint=True, acknowledged=False
        )
        is None
    )


def test_warning_cloud_embeddings_fires_for_openai_cloud():
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
    )

    warning = check_cloud_embeddings_enabled(
        embeddings_provider="openai",
        embeddings_base_url="",
        require_local_embeddings=False,
        acknowledged=False,
    )
    assert warning is not None
    assert warning["type"] == "cloud_embeddings_enabled"


def test_warning_cloud_embeddings_suppressed_for_local_base_url():
    """OpenAI provider type pointed at a local endpoint (LM Studio, vLLM)
    should not trigger the cloud-embeddings warning.
    """
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
    )

    assert (
        check_cloud_embeddings_enabled(
            embeddings_provider="openai",
            embeddings_base_url="http://localhost:1234/v1",
            require_local_embeddings=False,
            acknowledged=False,
        )
        is None
    )


def test_warning_cloud_embeddings_suppressed_for_sentence_transformers():
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
    )

    assert (
        check_cloud_embeddings_enabled(
            embeddings_provider="sentence_transformers",
            embeddings_base_url="",
            require_local_embeddings=False,
            acknowledged=False,
        )
        is None
    )


# ---------------------------------------------------------------------------
# Embeddings pre-flight
# ---------------------------------------------------------------------------


def test_rag_factory_blocks_openai_embeddings_under_require_local():
    """End-to-end: get_rag_service() must fail pre-flight when the user
    has embeddings.require_local=True and the configured embedding
    provider is OpenAI (cloud). This closes the silent corpus-upload
    leak the audit identified as the highest-impact embeddings vector.
    """
    from unittest.mock import MagicMock
    from local_deep_research.research_library.services import (
        rag_service_factory,
    )

    # Fake SettingsManager that returns openai as the provider + require_local=True.
    fake_settings = MagicMock()
    fake_settings.get_setting.side_effect = lambda key, *args, **kwargs: {
        "embeddings.require_local": True,
        "embeddings.openai.base_url": "",  # cloud endpoint
        "embeddings.ollama.url": "",
    }.get(key)

    with pytest.raises(PolicyDeniedError) as exc_info:
        rag_service_factory._enforce_embeddings_policy(
            "openai", fake_settings, username="alice"
        )
    assert exc_info.value.target == "openai"
    assert exc_info.value.decision.reason == "provider_cloud"


def test_rag_factory_allows_sentence_transformers_under_require_local():
    """sentence_transformers is always local; the pre-flight passes."""
    from unittest.mock import MagicMock
    from local_deep_research.research_library.services import (
        rag_service_factory,
    )

    fake_settings = MagicMock()
    fake_settings.get_setting.side_effect = lambda key, *args, **kwargs: {
        "embeddings.require_local": True,
        "embeddings.openai.base_url": "",
        "embeddings.ollama.url": "",
    }.get(key)

    # Should not raise.
    rag_service_factory._enforce_embeddings_policy(
        "sentence_transformers", fake_settings, username="alice"
    )


def test_rag_factory_allows_openai_pointed_at_local_endpoint():
    """OpenAI provider becomes "local" when base_url is loopback/RFC1918."""
    from unittest.mock import MagicMock
    from local_deep_research.research_library.services import (
        rag_service_factory,
    )

    fake_settings = MagicMock()
    fake_settings.get_setting.side_effect = lambda key, *args, **kwargs: {
        "embeddings.require_local": True,
        # OpenAI-compatible local endpoint (e.g. LM Studio).
        "embeddings.openai.base_url": "http://127.0.0.1:1234/v1",
        "embeddings.ollama.url": "",
    }.get(key)

    # Should not raise.
    rag_service_factory._enforce_embeddings_policy(
        "openai", fake_settings, username="alice"
    )


def test_rag_factory_denies_cloud_when_retired_both_coerces_to_adaptive():
    """`both` is retired (ADR-0007) and coerces to ADAPTIVE. The RAG gate
    classifies against the library corpus (sensitive) -> PRIVATE_ONLY -> local
    embeddings forced -> a cloud provider is DENIED even with require_local
    left False."""
    import pytest
    from unittest.mock import MagicMock
    from local_deep_research.research_library.services import (
        rag_service_factory,
    )
    from local_deep_research.security.egress.policy import PolicyDeniedError

    fake_settings = MagicMock()
    fake_settings.get_setting.side_effect = lambda key, *args, **kwargs: {
        "policy.egress_scope": "both",
        "embeddings.require_local": False,
        "embeddings.openai.base_url": "",
        "embeddings.ollama.url": "",
    }.get(key)

    with pytest.raises(PolicyDeniedError):
        rag_service_factory._enforce_embeddings_policy(
            "openai", fake_settings, username="alice"
        )


def test_rag_factory_denies_cloud_under_adaptive_default():
    """A missing scope falls back to the registered default ADAPTIVE. The
    gate classifies against primary_engine='library' (the corpus being
    embedded is local-nature), so adaptive resolves PRIVATE_ONLY and
    forces local embeddings — a cloud provider is denied even with
    embeddings.require_local at its default False. This matches what a
    real settings DB (default value 'adaptive') already produced."""
    from unittest.mock import MagicMock
    from local_deep_research.research_library.services import (
        rag_service_factory,
    )

    fake_settings = MagicMock()
    fake_settings.get_setting.side_effect = lambda key, *args, **kwargs: {
        "embeddings.require_local": False,
        "embeddings.openai.base_url": "",
        "embeddings.ollama.url": "",
    }.get(key)

    with pytest.raises(PolicyDeniedError) as exc_info:
        rag_service_factory._enforce_embeddings_policy(
            "openai", fake_settings, username="alice"
        )
    assert exc_info.value.decision.reason == "provider_cloud"


def test_factory_allows_primary_under_strict():
    """Factory permits the user's primary engine under STRICT."""
    from local_deep_research.web_search_engines.search_engine_factory import (
        create_search_engine,
    )

    snapshot = {
        "policy.egress_scope": "strict",
        "search.tool": "arxiv",
    }
    # We're not asserting on the returned engine (it would require full
    # arxiv setup), only that the policy gate doesn't raise.
    try:
        create_search_engine("arxiv", llm=None, settings_snapshot=snapshot)
    except PolicyDeniedError:
        pytest.fail("policy should allow the primary engine under STRICT")
    except Exception:
        # Engine instantiation may fail for unrelated reasons (e.g. config);
        # we only care that PolicyDeniedError was not raised.
        pass


# ---------------------------------------------------------------------------
# Round 6 fix-coverage regression tests (T1-T8 + new findings)
# ---------------------------------------------------------------------------


def test_t1_strict_blocks_public_engine_no_http_called():
    """T1: under STRICT + primary=arxiv, an attempt to construct a public
    engine (wikipedia) is refused and no HTTP socket fires.

    Catches a regression where the factory PEP was bypassed and an LLM-
    generated tool call could silently reach the network.
    """
    from local_deep_research.web_search_engines.search_engine_factory import (
        create_search_engine,
    )

    snapshot = {
        "policy.egress_scope": "strict",
        "search.tool": "arxiv",
    }
    with (
        patch("socket.getaddrinfo") as mock_getaddr,
        patch("socket.socket") as mock_socket,
    ):
        try:
            create_search_engine(
                "wikipedia", llm=None, settings_snapshot=snapshot
            )
        except PolicyDeniedError:
            pass
        except Exception:
            pass
        mock_getaddr.assert_not_called()
        mock_socket.assert_not_called()


def test_t2_list_typed_url_setting_any_public_wins():
    """T2: list-typed url_setting (e.g. Elasticsearch hosts) must
    classify as PUBLIC if any entry classifies as public — the safer
    fail-up direction. Without this an attacker could hide a public
    host inside a list of local hosts and bypass STRICT.
    """
    from local_deep_research.security.egress.policy import (
        _classify_engine_url,
    )

    ctx = make_ctx(scope=EgressScope.BOTH)
    snapshot = {
        "elastic.hosts": [
            "http://192.168.1.5:9200",  # local
            "http://elastic-cloud.example.com:9200",  # public
        ]
    }
    with patch(
        "local_deep_research.security.egress.policy._classify_host"
    ) as mock_cls:
        mock_cls.side_effect = lambda host, ctx, allow_dns=True: (
            host.startswith("192.")
        )
        result = _classify_engine_url("elastic.hosts", snapshot, ctx)
    # any public present => return False (public)
    assert result is False


def test_t3_classify_host_uses_dns_when_not_an_ip():
    """T3: exercise the REAL _classify_host (not patched out) with
    mocked socket.getaddrinfo. Catches regressions in the DNS code
    path that mock-heavy tests would miss.
    """
    from local_deep_research.security.egress.policy import _classify_host

    ctx = make_ctx()
    fake_addrinfo = [
        (None, None, None, None, ("192.168.1.42", 0)),
    ]
    with patch(
        "local_deep_research.security.egress.policy._resolve_with_timeout",
        return_value=fake_addrinfo,
    ):
        assert _classify_host("example.lab", ctx) is True


def test_t4_ipv6_literal_url_classifies():
    """T4: IPv6-literal URLs (with square brackets) classify
    correctly. Catches regressions in the bracket-strip logic.
    """
    from local_deep_research.security.egress.policy import _classify_host

    ctx = make_ctx()
    assert _classify_host("[::1]", ctx) is True


def test_t5_dns_cache_avoids_repeat_lookups():
    """T5: A second classification of the same hostname uses the
    cache — DO NOT call getaddrinfo twice.
    """
    from local_deep_research.security.egress.policy import _classify_host

    ctx = make_ctx()
    with patch(
        "local_deep_research.security.egress.policy._resolve_with_timeout",
        return_value=[(None, None, None, None, ("8.8.8.8", 0))],
    ) as mock_resolve:
        _classify_host("dns.example.com", ctx)
        _classify_host("dns.example.com", ctx)
        assert mock_resolve.call_count == 1


def test_t7_dns_timeout_fail_safe_treats_as_public():
    """T7: when DNS lookup times out, _classify_host falls back to
    public (False) so PRIVATE_ONLY doesn't accidentally allow an
    unreachable host that might resolve to anything later.
    """
    from local_deep_research.security.egress.policy import _classify_host

    ctx = make_ctx()
    with patch(
        "local_deep_research.security.egress.policy._resolve_with_timeout",
        return_value=None,
    ):
        assert _classify_host("timed-out.example.com", ctx) is False


def test_t7b_resolve_with_timeout_actually_bounds_a_hung_lookup():
    """Regression: a hung getaddrinfo must NOT block past the timeout.

    The earlier implementation drove the lookup from a
    ``with ThreadPoolExecutor(...)`` block; the context manager's __exit__
    calls shutdown(wait=True), which re-blocked on the abandoned worker and
    defeated the timeout entirely (the call returned only after the OS DNS
    timeout). This asserts the bound actually holds.
    """
    import time

    from local_deep_research.security.egress import policy as egress_policy

    # Sleep kept modest (1.2s): the orphaned worker is joined by the
    # concurrent.futures atexit handler at interpreter shutdown, so a large
    # sleep would slow pytest teardown. 1.2s is still 6x the 0.2s timeout —
    # enough to distinguish "bounded" from "waited out the worker".
    def _hang(*_a, **_k):
        # Simulates a hung getaddrinfo in an abandoned worker; the test
        # itself returns in ~0.2s (the timeout), so it is not slow and must
        # not be skipped under -m 'not slow'.
        time.sleep(1.2)  # allow: unmarked-sleep - mocked hang, test is fast
        return [(2, 1, 6, "", ("127.0.0.1", 0))]

    with (
        patch.object(egress_policy, "_DNS_TIMEOUT_SEC", 0.2),
        patch("socket.getaddrinfo", side_effect=_hang),
    ):
        start = time.monotonic()
        result = egress_policy._resolve_with_timeout("hung.example.com")
        elapsed = time.monotonic() - start

    assert result is None
    # Ceiling = timeout 0.2s + scheduling slack, well below the 1.2s hang —
    # proves the timeout is effective rather than waiting out the worker.
    assert elapsed < 0.9, f"resolve did not bound the hang: {elapsed:.2f}s"


@pytest.mark.parametrize(
    "scope",
    [
        EgressScope.STRICT,
        EgressScope.PRIVATE_ONLY,
        EgressScope.BOTH,
        EgressScope.PUBLIC_ONLY,
    ],
)
@pytest.mark.parametrize(
    "url",
    [
        "http://169.254.169.254/latest/meta-data/",  # AWS/GCE IMDS (IPv4)
        "http://[::ffff:169.254.169.254]/",  # IPv4-mapped IPv6 form
        "http://169.254.170.2/",  # AWS ECS task metadata
    ],
)
def test_cloud_metadata_ip_blocked_under_every_scope(scope, url):
    """Cloud-metadata endpoints classify as link-local/private, so STRICT and
    PRIVATE_ONLY would otherwise ALLOW them — a credential-theft path,
    especially via the audit-hook net which calls evaluate_url directly on
    raw socket.connect targets (bypassing the SSRF validator the explicit
    fetch PEPs run first). They must be refused regardless of scope.
    """
    ctx = make_ctx(scope=scope, primary="arxiv")
    decision = evaluate_url(url, ctx)
    assert decision.allowed is False
    assert decision.reason == "blocked_metadata_ip"


def test_hostname_resolving_to_metadata_ip_is_refused_under_private_scopes():
    """A hostname whose DNS A-record points at a cloud-metadata IP
    (169.254.169.254) must NOT classify as local — link-local metadata IPs
    pass is_private_ip, which would let STRICT/PRIVATE_ONLY fetch IMDS via a
    hostname. _classify_host now rejects resolved metadata IPs.
    """
    for scope in (EgressScope.STRICT, EgressScope.PRIVATE_ONLY):
        ctx = make_ctx(scope=scope, primary="library")
        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("169.254.169.254", 0))],
        ):
            decision = evaluate_url(
                "http://imds.attacker.example/latest/meta-data/", ctx
            )
        assert decision.allowed is False, f"leaked under {scope}"


def test_private_host_still_allowed_under_strict_and_private():
    """The metadata guard must NOT over-block legitimate private hosts
    (local Ollama/SearXNG) under STRICT / PRIVATE_ONLY.
    """
    for scope in (EgressScope.STRICT, EgressScope.PRIVATE_ONLY):
        ctx = make_ctx(scope=scope, primary="arxiv")
        decision = evaluate_url("http://127.0.0.1:11434/", ctx)
        assert decision.allowed is True


def test_t8_denial_quota_off_by_one():
    """T8: exactly MAX_DENIED_FETCHES_PER_RUN denials are processed
    individually; the next call hits the quota cap.
    """
    ctx = make_ctx(scope=EgressScope.STRICT, primary="arxiv")
    # Force every URL to be denied by using PUBLIC URLs under STRICT.
    for _ in range(MAX_DENIED_FETCHES_PER_RUN):
        decision = evaluate_url("https://public-host.example.com/", ctx)
        assert decision.allowed is False
        assert decision.reason != "denial_quota_exceeded"
    # The next denial hits the quota.
    decision = evaluate_url("https://public-host.example.com/", ctx)
    assert decision.reason == "denial_quota_exceeded"


def test_nat64_metadata_classifies_as_public():
    """H2: NAT64-wrapped cloud-metadata IPs are not silently
    classified as private. Catches the original H2 regression.
    """
    from local_deep_research.security.egress.policy import _classify_host

    ctx = make_ctx()
    # 64:ff9b::169.254.169.254 — NAT64 wrapping the AWS metadata IP.
    result = _classify_host("64:ff9b::a9fe:a9fe", ctx)
    assert result is False, "NAT64 metadata IP must classify as public"


def test_h3_nested_dict_url_setting_unwraps():
    """H3: production snapshots wrap each setting in
    {"value": X, "type": ..., ...}. The PDP MUST unwrap before
    parsing the URL, or it sees a dict and silently skips
    classification.
    """
    from local_deep_research.security.egress.policy import (
        _classify_engine_url,
    )

    ctx = make_ctx()
    snapshot = {
        "search_url": {"value": "http://192.168.10.10/search", "type": "str"}
    }
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=True,
    ):
        result = _classify_engine_url("search_url", snapshot, ctx)
    assert result is True, "nested-dict url_setting must unwrap"


def test_c2_policy_aware_validate_url_private_only_allows_private():
    """C2: under PRIVATE_ONLY policy, the SSRF wrapper permits private
    IPs so local lab deployments work without SSRF_ALLOW_PRIVATE_IPS=1.
    """
    from local_deep_research.security.egress.fetch import (
        policy_aware_validate_url,
    )

    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    assert policy_aware_validate_url("http://127.0.0.1:11434", ctx) is True
    assert policy_aware_validate_url("http://192.168.1.5/api", ctx) is True


def test_c2_policy_aware_validate_url_metadata_still_blocked():
    """C2: even under PRIVATE_ONLY, cloud-metadata IPs are blocked
    — they're in ALWAYS_BLOCKED regardless of the scope flag.
    """
    from local_deep_research.security.egress.fetch import (
        policy_aware_validate_url,
    )

    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    assert policy_aware_validate_url("http://169.254.169.254/", ctx) is False


def test_n1_retriever_registry_get_metadata():
    """retriever_registry.get_metadata returns the classification dict
    for a registered retriever (default {"is_local": True}), None for
    unknown.
    """
    from unittest.mock import MagicMock

    from local_deep_research.web_search_engines.retriever_registry import (
        retriever_registry,
    )

    name = "_test_retriever_n1"
    retriever_registry.register(name, MagicMock())
    try:
        assert retriever_registry.get_metadata(name) == {"is_local": True}
        assert retriever_registry.get_metadata("does-not-exist") is None
    finally:
        retriever_registry.unregister(name)


def test_classified_retriever_is_local_flag():
    """register(is_local=...) records classification; register_multiple
    applies it to all; unregister/clear drop the metadata."""
    from unittest.mock import MagicMock

    from local_deep_research.web_search_engines.retriever_registry import (
        RetrieverRegistry,
    )

    reg = RetrieverRegistry()
    reg.register("local_kb", MagicMock(), is_local=True)
    reg.register("public_idx", MagicMock(), is_local=False)
    assert reg.get_metadata("local_kb") == {"is_local": True}
    assert reg.get_metadata("public_idx") == {"is_local": False}

    reg.register_multiple({"a": MagicMock(), "b": MagicMock()}, is_local=False)
    assert reg.get_metadata("a") == {"is_local": False}
    assert reg.get_metadata("b") == {"is_local": False}

    reg.unregister("a")
    assert reg.get_metadata("a") is None
    reg.clear()
    assert reg.get_metadata("local_kb") is None


def test_evaluate_retriever_scopes():
    """evaluate_retriever honors scope: local retriever allowed under
    BOTH/PRIVATE_ONLY, denied under PUBLIC_ONLY, and under STRICT only
    when it is the primary engine."""
    from unittest.mock import MagicMock

    from local_deep_research.security.egress.policy import evaluate_retriever
    from local_deep_research.web_search_engines.retriever_registry import (
        retriever_registry,
    )

    name = "_test_local_kb"
    retriever_registry.register(name, MagicMock(), is_local=True)
    try:
        # BOTH → allowed
        assert evaluate_retriever(
            name, make_ctx(scope=EgressScope.BOTH)
        ).allowed
        # PRIVATE_ONLY → allowed (it's local)
        assert evaluate_retriever(
            name, make_ctx(scope=EgressScope.PRIVATE_ONLY)
        ).allowed
        # PUBLIC_ONLY → denied (it's local)
        d = evaluate_retriever(name, make_ctx(scope=EgressScope.PUBLIC_ONLY))
        assert not d.allowed
        assert d.reason == "scope_mismatch_public_only"
        # STRICT + retriever is the primary → allowed
        assert evaluate_retriever(
            name, make_ctx(scope=EgressScope.STRICT, primary=name)
        ).allowed
        # STRICT + retriever is NOT the primary → denied
        d = evaluate_retriever(
            name, make_ctx(scope=EgressScope.STRICT, primary="arxiv")
        )
        assert not d.allowed
        assert d.reason == "strict_not_primary"
        # Unknown retriever → None metadata → retriever_unknown
        d = evaluate_retriever("nope", make_ctx(scope=EgressScope.BOTH))
        assert not d.allowed
    finally:
        retriever_registry.unregister(name)


def test_n8_unknown_scope_raises_after_context_from_snapshot():
    """N8: unknown scope is fail-closed at context build time.
    Verified separately above; this test pins the error reason code.
    """
    with pytest.raises(PolicyDeniedError) as excinfo:
        context_from_snapshot(
            {"policy.egress_scope": "nope_thx"}, primary_engine="arxiv"
        )
    assert excinfo.value.decision.reason == "unknown_egress_scope"


def test_c8_banner_suppressed_for_private_base_url():
    """C8: cloud-embeddings banner suppressed when base_url points to
    any private IP (RFC1918, CGNAT, link-local, IPv6 private). The old
    substring-match version missed legitimate private endpoints.
    """
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
    )

    # RFC1918 host should NOT fire a banner (was the false-positive case).
    result = check_cloud_embeddings_enabled(
        embeddings_provider="openai",
        embeddings_base_url="http://172.16.0.5/v1",
        require_local_embeddings=False,
        acknowledged=False,
    )
    assert result is None


def test_c8_banner_fires_for_public_base_url():
    """C8: cloud-embeddings banner fires for genuine cloud base_url."""
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
    )

    result = check_cloud_embeddings_enabled(
        embeddings_provider="openai",
        embeddings_base_url="https://api.openai.com/v1",
        require_local_embeddings=False,
        acknowledged=False,
    )
    assert result is not None
    assert result["type"] == "cloud_embeddings_enabled"


def test_dismiss_key_uses_new_convention():
    """N18: the egress-policy banner's dismissKey matches the
    app.warnings.dismiss_* convention used by every other warning.
    """
    from local_deep_research.security.egress.warnings import (
        check_public_egress_enabled,
    )

    result = check_public_egress_enabled(
        egress_scope="both", acknowledged=False
    )
    assert result is not None
    assert result["dismissKey"] == "app.warnings.dismiss_egress_policy"


def test_every_llm_provider_is_classified_by_egress_policy():
    """Every auto-discovered LLM provider must be classified by the egress
    policy — cloud (_CLOUD_LLM_PROVIDERS), local-default
    (_LOCAL_DEFAULT_LLM_PROVIDERS), or the URL-classified ``openai_endpoint``.
    A newly-added provider that's in none of these would silently bypass the
    require_local / PRIVATE_ONLY LLM gate, so this test fails loudly until it's
    classified.
    """
    from local_deep_research.llm.providers.auto_discovery import (
        discover_providers,
    )
    from local_deep_research.llm.providers.base import normalize_provider
    from local_deep_research.security.egress.policy import (
        _CLOUD_LLM_PROVIDERS,
        _LOCAL_DEFAULT_LLM_PROVIDERS,
    )

    # openai_endpoint and anthropic_endpoint are classified by their configured
    # URL (llm.<provider>.url), not a static set.
    classified = (
        _CLOUD_LLM_PROVIDERS
        | _LOCAL_DEFAULT_LLM_PROVIDERS
        | {"openai_endpoint", "anthropic_endpoint"}
    )
    providers = discover_providers()
    assert providers, "no providers discovered — test wiring issue"
    unclassified = sorted(
        {
            normalize_provider(key)
            for key in providers
            if normalize_provider(key) not in classified
        }
    )
    assert not unclassified, (
        "These LLM providers are not classified by the egress policy and "
        "would bypass the require_local gate — add them to "
        "_CLOUD_LLM_PROVIDERS or _LOCAL_DEFAULT_LLM_PROVIDERS in "
        f"security/egress/policy.py: {unclassified}"
    )


def test_check_effective_scope_states_adaptive_resolution():
    """The informational banner makes ADAPTIVE's effective posture explicit
    (public / private / both) and only fires for adaptive."""
    from local_deep_research.security.egress.warnings import (
        check_effective_scope,
    )

    pub = check_effective_scope("adaptive", "public_only", "searxng", False)
    assert pub is not None and "Public searches enabled" in pub["title"]
    priv = check_effective_scope(
        "adaptive", "private_only", "collection_x", False
    )
    assert priv is not None and "Private only" in priv["title"]
    both = check_effective_scope("adaptive", "both", "auto", False)
    assert both is not None and "Public + private" in both["title"]

    # Only adaptive — explicit scopes are self-describing in the dropdown.
    assert check_effective_scope("both", "both", "searxng", False) is None
    assert (
        check_effective_scope("private_only", "private_only", "x", False)
        is None
    )
    # Dismissed → suppressed; separate dismiss key from the risk banners.
    assert check_effective_scope("adaptive", "public_only", "x", True) is None
    assert pub["dismissKey"] == "app.warnings.dismiss_adaptive_scope_info"


def test_egress_banners_have_distinct_dismiss_keys():
    """The three egress banners must use DISTINCT dismiss keys so dismissing
    the fresh-install public-egress notice can't silently suppress the
    critical cloud-LLM / cloud-embeddings warnings (false-safety trap)."""
    from local_deep_research.security.egress.warnings import (
        check_cloud_embeddings_enabled,
        check_cloud_llm_enabled,
        check_public_egress_enabled,
    )

    pub = check_public_egress_enabled(egress_scope="both", acknowledged=False)
    llm = check_cloud_llm_enabled(
        provider="openai", require_local_endpoint=False, acknowledged=False
    )
    emb = check_cloud_embeddings_enabled(
        embeddings_provider="openai",
        embeddings_base_url="",
        require_local_embeddings=False,
        acknowledged=False,
    )
    keys = {pub["dismissKey"], llm["dismissKey"], emb["dismissKey"]}
    assert len(keys) == 3, f"dismiss keys not distinct: {keys}"
    # Dismissing public-egress alone leaves the critical banners firing.
    assert (
        check_cloud_llm_enabled(
            provider="openai",
            require_local_endpoint=False,
            acknowledged=False,
        )
        is not None
    )


def test_n16_policy_setting_triggers_audit_logger():
    """N16: writing a policy.* key emits a policy_audit log line so
    admins can audit changes. Without it, key changes are silent.
    """
    from local_deep_research.settings.manager import _is_policy_setting

    assert _is_policy_setting("policy.egress_scope") is True
    assert _is_policy_setting("llm.require_local_endpoint") is True
    assert _is_policy_setting("embeddings.require_local") is True
    assert _is_policy_setting("llm.temperature") is False
    assert _is_policy_setting("search.iterations") is False


def test_n17_research_context_carries_settings_snapshot():
    """Every site that calls ``set_search_context`` must populate
    ``settings_snapshot`` — the live egress-policy scope, engine config,
    and per-user resolution are all read from it, so a constructor that
    drops it silently degrades them to defaults. (Originally caught as the
    N17 cache-key wiring bug; the search cache has since been removed, but
    the contract still matters for the live egress path.) Pins the contract
    by inspecting the source of each known call site so a future regression
    fails at unit-test time, not at runtime.
    """
    import inspect

    from local_deep_research.api import research_functions
    from local_deep_research.web.services import research_service

    for module in (research_functions, research_service):
        src = inspect.getsource(module)
        construct_count = src.count("search_context = {") + src.count(
            "shared_research_context = {"
        )
        snapshot_in_context = src.count('"settings_snapshot":')
        assert snapshot_in_context >= construct_count, (
            f"{module.__name__}: every search_context constructor must "
            f"include 'settings_snapshot' "
            f"(constructions={construct_count}, "
            f"with-snapshot={snapshot_in_context})"
        )


def test_sbert_cache_check_probes_both_forms_for_bare_sbert_name():
    """Bare SentenceTransformer model names (e.g. "all-MiniLM-L6-v2")
    are downloaded under the sentence-transformers org, so the loader
    requests the namespaced repo_id and the HF hub cache is keyed on
    that. The check probes BOTH the bare and the namespaced forms so
    it also covers names in the loader's ``basic_transformer_models``
    allowlist (which use the bare key) — but the namespaced form is
    the one that hits for shipped SBERT defaults.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    probed_repo_ids = []

    def fake_try_to_load_from_cache(repo_id, filename):
        probed_repo_ids.append(repo_id)
        if repo_id == "sentence-transformers/all-MiniLM-L6-v2":
            return "/fake/path/config.json"
        return None

    with patch(
        "huggingface_hub.try_to_load_from_cache",
        side_effect=fake_try_to_load_from_cache,
    ):
        assert (
            SentenceTransformersProvider._is_model_cached_locally(
                "all-MiniLM-L6-v2"
            )
            is True
        )
    # The namespaced form is the hit for SBERT-curated defaults; assert
    # it was probed (order-independent — the bare-key probe is covered
    # by the basic-transformer test below).
    assert "sentence-transformers/all-MiniLM-L6-v2" in probed_repo_ids


def test_sbert_cache_check_probes_bare_form_for_basic_transformer_name():
    """Vanilla transformer models in SentenceTransformer's
    ``basic_transformer_models`` allowlist (bert-base-uncased, gpt2,
    t5-base, ...) are loaded from the BARE repo_id, not prefixed. The
    check must find them under the bare key — probed first — so a
    correctly cached vanilla transformer is admitted under
    require_local instead of false-negativing.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    probed_repo_ids = []

    def fake_try_to_load_from_cache(repo_id, filename):
        probed_repo_ids.append(repo_id)
        if repo_id == "bert-base-uncased":
            return "/fake/path/config.json"
        return None

    with patch(
        "huggingface_hub.try_to_load_from_cache",
        side_effect=fake_try_to_load_from_cache,
    ):
        assert (
            SentenceTransformersProvider._is_model_cached_locally(
                "bert-base-uncased"
            )
            is True
        )
    assert "bert-base-uncased" in probed_repo_ids


def test_sbert_cache_check_does_not_double_prefix_namespaced_input():
    """Already-namespaced inputs (e.g. 'BAAI/bge-small-en') must be
    probed verbatim — not re-prefixed to
    'sentence-transformers/BAAI/bge-small-en', which would always miss.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    probed_repo_ids = []

    def fake_try_to_load_from_cache(repo_id, filename):
        probed_repo_ids.append(repo_id)
        return

    with patch(
        "huggingface_hub.try_to_load_from_cache",
        side_effect=fake_try_to_load_from_cache,
    ):
        assert (
            SentenceTransformersProvider._is_model_cached_locally(
                "BAAI/bge-small-en"
            )
            is False
        )
    assert probed_repo_ids == ["BAAI/bge-small-en"]


@pytest.mark.parametrize(
    "miss_value",
    [None, _CACHED_NO_EXIST],
    ids=["never_probed", "known_absent"],
)
def test_sbert_cache_check_treats_non_string_returns_as_misses(miss_value):
    """try_to_load_from_cache returns None for never-probed files and
    the _CACHED_NO_EXIST sentinel for confirmed-absent ones. Both must
    be treated as cache misses so require_local keeps blocking — a
    known-absent file is not a cached file.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    with patch(
        "huggingface_hub.try_to_load_from_cache",
        return_value=miss_value,
    ):
        assert (
            SentenceTransformersProvider._is_model_cached_locally(
                "all-MiniLM-L6-v2"
            )
            is False
        )


def test_sbert_cache_check_admits_existing_local_path(tmp_path):
    """A model configured as an existing local directory is already on
    disk — the loader takes its ``os.path.exists`` branch and never hits
    the network, so it must be admitted under require_local without an
    HF cache probe. Before this guard the path was probed as an HF
    repo_id, raised HFValidationError, was swallowed, and false-denied
    with ``embeddings_model_not_cached``. Differential: returns False on
    the pre-guard code.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    # tmp_path is an existing directory. The admit checks only that the
    # path exists (no config.json / model-structure validation), so a
    # bare existing dir is sufficient — mirroring the loader's
    # os.path.exists guard, which likewise does not inspect contents.
    with patch("huggingface_hub.try_to_load_from_cache") as mock_cache:
        assert (
            SentenceTransformersProvider._is_model_cached_locally(str(tmp_path))
            is True
        )
        # Local-path admission short-circuits before any HF cache probe.
        mock_cache.assert_not_called()


def test_sbert_cache_check_blank_name_not_treated_as_existing_path():
    """Regression guard for the ``Path("")`` quirk: an empty model_name
    normalises to ``.`` (the cwd, which exists), so a naive
    ``Path(model_name).exists()`` would wrongly admit a blank config.
    The ``model_name and`` truthiness guard keeps it fail-closed, in
    parity with the loader's ``os.path.exists("") is False``.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    assert SentenceTransformersProvider._is_model_cached_locally("") is False
    # Whitespace is truthy, so unlike "" it DOES reach Path.exists() — it
    # must still miss (no such path) and fall through to fail-closed,
    # rather than being admitted as a local path.
    with patch("huggingface_hub.try_to_load_from_cache", return_value=None):
        assert (
            SentenceTransformersProvider._is_model_cached_locally("   ")
            is False
        )


def test_sentence_transformers_model_hub_organization_pin():
    """Pin the private ``__MODEL_HUB_ORGANIZATION__`` constant that
    ``_is_model_cached_locally`` uses to build the namespaced cache key
    (``sentence-transformers/<name>``) for bare SBERT model names.

    If upstream renames or removes it, the import inside the cache
    check raises ``ImportError``, the broad ``except`` swallows it, and
    every ``require_local=True`` run silently fail-closes with a
    misleading ``embeddings_model_not_cached``. Importing it here
    surfaces removal in CI; the value assert catches silent cache-key
    drift if upstream ever ships a different org string.
    """
    from sentence_transformers import __MODEL_HUB_ORGANIZATION__

    assert __MODEL_HUB_ORGANIZATION__ == "sentence-transformers"


def test_create_embeddings_admits_cached_default_model_under_require_local():
    """End-to-end admit path — the exact regression the fix targets.

    Under require_local, a shipped-default model that IS cached (only
    under the ``sentence-transformers/`` namespaced HF key) must be
    admitted through the gate: no PolicyDeniedError, and the load forced
    offline with ``local_files_only=True``. Before the fix, the gate's
    cache probe used the bare name, missed the namespaced key, and
    raised ``embeddings_model_not_cached`` for a model that was present
    — so this test fails on the old single-probe code and passes with
    the fix. The unit test pins ``_is_model_cached_locally``; this pins
    the wiring from that check through ``create_embeddings`` to admission.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    def fake_cache(repo_id, filename):
        # Only the namespaced key hits — what the loader actually
        # requests for a bare SBERT default, and what the old code missed.
        if repo_id == "sentence-transformers/all-MiniLM-L6-v2":
            return "/fake/path/config.json"
        return None

    stub = object()
    # Empty snapshot: resolve_run_primary_engine raises (no primary), so
    # create_embeddings fails closed to require_local=True and the gate
    # actually runs — no need to synthesise a PRIVATE_ONLY scope.
    with (
        patch("huggingface_hub.try_to_load_from_cache", side_effect=fake_cache),
        patch(
            "langchain_community.embeddings.SentenceTransformerEmbeddings",
            return_value=stub,
        ) as mock_st,
    ):
        result = SentenceTransformersProvider.create_embeddings(
            model="all-MiniLM-L6-v2",
            settings_snapshot={},
            device="cpu",
        )

    assert result is stub
    # Admitted, and forced offline for defence in depth.
    assert mock_st.call_args.kwargs["model_kwargs"]["local_files_only"] is True


def test_create_embeddings_denies_uncached_model_under_require_local():
    """End-to-end deny path through the same gate: a genuinely uncached
    model under require_local raises ``embeddings_model_not_cached`` and
    the SentenceTransformer loader is never invoked — so no HuggingFace
    download is triggered. Guards the fail-closed direction against a
    future change that widens admission too far.
    """
    from local_deep_research.embeddings.providers.implementations.sentence_transformers import (
        SentenceTransformersProvider,
    )

    with (
        patch("huggingface_hub.try_to_load_from_cache", return_value=None),
        patch(
            "langchain_community.embeddings.SentenceTransformerEmbeddings"
        ) as mock_st,
    ):
        with pytest.raises(PolicyDeniedError) as excinfo:
            SentenceTransformersProvider.create_embeddings(
                model="definitely-not-cached-zzz",
                settings_snapshot={},
                device="cpu",
            )

    assert excinfo.value.decision.reason == "embeddings_model_not_cached"
    mock_st.assert_not_called()


def test_local_embedding_manager_propagates_policy_denied():
    """LocalEmbeddingManager._initialize_embeddings catches Exception
    broadly and falls back to HuggingFaceEmbeddings. Without an
    explicit PolicyDeniedError re-raise, my N6 SentenceTransformer
    pre-flight gets silently swallowed AND the fallback then tries to
    download from HF — re-opening the very hole N6 was meant to close.
    """
    from local_deep_research.web_search_engines.engines.local_embedding_manager import (
        LocalEmbeddingManager,
    )

    mgr = LocalEmbeddingManager.__new__(LocalEmbeddingManager)
    mgr.embedding_model_type = "sentence_transformers"
    mgr.embedding_model = "all-MiniLM-L6-v2"
    mgr.embedding_device = "cpu"
    mgr.ollama_base_url = None
    mgr.settings_snapshot = {}

    fake_denial = PolicyDeniedError(
        Decision(False, "embeddings_model_not_cached"), target="fake-model"
    )
    with patch(
        "local_deep_research.embeddings.get_embeddings",
        side_effect=fake_denial,
    ):
        with pytest.raises(PolicyDeniedError) as excinfo:
            mgr._initialize_embeddings()
    assert excinfo.value.decision.reason == "embeddings_model_not_cached"


def test_n14_subscription_policy_rejects_forbidden_engine():
    """N14: _validate_subscription_policy returns a rejection reason when
    the subscription's engine violates the user's egress policy, and None
    when it's allowed. Uses a mocked settings manager so no real DB is
    needed."""
    from unittest.mock import MagicMock, patch

    from local_deep_research.news import api as news_api

    snapshot = {"policy.egress_scope": "strict", "search.tool": "arxiv"}
    fake_sm = MagicMock()
    fake_sm.get_settings_snapshot.return_value = snapshot
    fake_sm.get_setting.side_effect = lambda key, default=None: (
        "arxiv" if key == "search.tool" else default
    )

    with patch(
        "local_deep_research.utilities.db_utils.get_settings_manager",
        return_value=fake_sm,
    ):
        # Under STRICT+arxiv, a non-primary engine is rejected.
        reason = news_api._validate_subscription_policy(
            db_session=MagicMock(),
            user_id="alice",
            search_engine="pubmed",
            model_provider=None,
        )
        assert reason is not None
        assert "pubmed" in reason

        # The primary engine itself is allowed.
        assert (
            news_api._validate_subscription_policy(
                db_session=MagicMock(),
                user_id="alice",
                search_engine="arxiv",
                model_provider=None,
            )
            is None
        )


def test_n14_subscription_policy_skips_without_settings():
    """N14: when the settings backend is unavailable, validation is
    skipped (best-effort) — the execution-time factory PEP backstops."""
    from unittest.mock import MagicMock, patch

    from local_deep_research.news import api as news_api

    with patch(
        "local_deep_research.utilities.db_utils.get_settings_manager",
        side_effect=RuntimeError("no settings DB"),
    ):
        assert (
            news_api._validate_subscription_policy(
                db_session=MagicMock(),
                user_id="alice",
                search_engine="pubmed",
                model_provider="openai",
            )
            is None
        )


def test_n15_subscription_policy_rejects_stray_meta_primary_config():
    """R3 #8 follow-up: a stray removed meta-picker primary ("auto") under
    STRICT no longer raises at context construction — it must STILL reject a
    non-primary subscription engine (fail closed), never silently allow it."""
    from unittest.mock import MagicMock, patch

    from local_deep_research.news import api as news_api

    snapshot = {"policy.egress_scope": "strict", "search.tool": "auto"}
    fake_sm = MagicMock()
    fake_sm.get_settings_snapshot.return_value = snapshot
    fake_sm.get_setting.side_effect = lambda key, default=None: (
        "auto" if key == "search.tool" else default
    )

    with patch(
        "local_deep_research.utilities.db_utils.get_settings_manager",
        return_value=fake_sm,
    ):
        reason = news_api._validate_subscription_policy(
            db_session=MagicMock(),
            user_id="alice",
            search_engine="pubmed",
            model_provider=None,
        )
        assert reason is not None
        assert "strict_not_primary" in reason


# ---------------------------------------------------------------------------
# Review round 2 — regression tests for confirmed findings
# ---------------------------------------------------------------------------

from local_deep_research.security.egress.policy import (  # noqa: E402
    _classify_host,
)


def test_classify_host_literal_metadata_ip_is_public():
    """Literal cloud-metadata IPs must classify as PUBLIC on the literal-IP
    branch too (not just the DNS branch), so evaluate_llm_endpoint /
    evaluate_embeddings / _classify_engine_url (which call _classify_host
    directly, with no metadata pre-check) can't treat 169.254.169.254 as a
    'local' inference/search target. Regression for the IMDS-SSRF gap."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    assert _classify_host("169.254.169.254", ctx) is False


def test_classify_host_literal_private_ip_still_local():
    """The metadata fix must NOT regress normal literal private IPs."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    assert _classify_host("10.0.0.5", ctx) is True
    assert _classify_host("127.0.0.1", ctx) is True


def test_llm_endpoint_at_metadata_ip_denied_under_require_local():
    """End-to-end: an LLM endpoint pointed at the metadata IP must be denied
    under require_local (it previously slipped through as provider_local)."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY, require_local_llm=True)
    decision = evaluate_llm_endpoint(
        "openai_endpoint",
        ctx,
        settings_snapshot={
            "llm.openai_endpoint.url": "http://169.254.169.254/v1"
        },
    )
    assert not decision.allowed


def test_adaptive_local_retriever_primary_resolves_private():
    """ADAPTIVE with a registered LOCAL retriever as the primary must resolve
    to PRIVATE_ONLY (forcing local inference), not the permissive BOTH.
    Regression for the private-corpus-to-cloud leak."""
    from local_deep_research.web_search_engines.retriever_registry import (
        retriever_registry,
    )

    class _FakeRetriever:
        pass

    retriever_registry.register("mykb_local", _FakeRetriever(), is_local=True)
    try:
        ctx = context_from_snapshot(
            {
                "policy.egress_scope": {"value": "adaptive"},
                "search.tool": {"value": "mykb_local"},
            },
            primary_engine="mykb_local",
        )
        assert ctx.scope == EgressScope.PRIVATE_ONLY
        assert ctx.require_local_llm is True
        assert ctx.require_local_embeddings is True
    finally:
        retriever_registry.unregister("mykb_local")


def test_adaptive_public_retriever_primary_resolves_public():
    """ADAPTIVE with a registered PUBLIC retriever primary resolves to
    PUBLIC_ONLY (symmetry check)."""
    from local_deep_research.web_search_engines.retriever_registry import (
        retriever_registry,
    )

    class _FakeRetriever:
        pass

    retriever_registry.register("mykb_public", _FakeRetriever(), is_local=False)
    try:
        ctx = context_from_snapshot(
            {
                "policy.egress_scope": {"value": "adaptive"},
                "search.tool": {"value": "mykb_public"},
            },
            primary_engine="mykb_public",
        )
        assert ctx.scope == EgressScope.PUBLIC_ONLY
    finally:
        retriever_registry.unregister("mykb_public")


def test_benign_denials_do_not_exhaust_fetch_quota():
    """Benign parse failures (mailto:/ftp: links scraped from pages) must NOT
    count toward the per-run denied-fetch quota — otherwise a long PUBLIC_ONLY
    run starts refusing legitimate public URLs mid-run."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    for _ in range(MAX_DENIED_FETCHES_PER_RUN + 10):
        d = evaluate_url("mailto:someone@example.com", ctx)
        assert not d.allowed
        assert d.reason == "unsupported_scheme"
    # The quota was never consumed, so a legitimate public URL still passes.
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=False,
    ):
        decision = evaluate_url("https://example.com/page", ctx)
    assert decision.allowed


def test_security_denials_still_exhaust_fetch_quota():
    """Security-relevant denials DO still count, preserving the exhaust-attack
    guard."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    with patch(
        "local_deep_research.security.egress.policy._classify_host",
        return_value=True,  # private host → scope_mismatch_public_only
    ):
        for _ in range(MAX_DENIED_FETCHES_PER_RUN):
            evaluate_url("https://private.example/page", ctx)
        decision = evaluate_url("https://private.example/page", ctx)
    assert decision.reason == "denial_quota_exceeded"


def test_allowed_local_hostnames_accepts_unresolvable():
    """A legitimate intranet/VPN hostname that does not resolve (DNS down /
    split-horizon) must be ACCEPTED on save (fail-open), not rejected — the
    documented behavior. Regression for the over-block."""
    from local_deep_research.security.egress.validators import (
        validate_allowed_local_hostnames,
    )

    result = validate_allowed_local_hostnames(
        {"llm.allowed_local_hostnames": ["box.invalid.nonexistent.local"]},
        {},
    )
    assert result is None  # no rejection


def test_elasticsearch_cloud_id_rejected_under_private_scope():
    """Elasticsearch cloud_id (public Elastic Cloud) must be refused under
    PRIVATE_ONLY at construction. Regression for the cloud_id bypass."""
    from local_deep_research.web_search_engines.engines.search_engine_elasticsearch import (  # noqa: E501
        ElasticsearchSearchEngine,
    )

    snapshot = {"policy.egress_scope": {"value": "private_only"}}
    with pytest.raises(PolicyDeniedError):
        ElasticsearchSearchEngine(
            cloud_id="my-deployment:dXMtY2VudHJhbDE=",
            settings_snapshot=snapshot,
        )


def test_elasticsearch_cloud_id_allowed_under_both_scope():
    """cloud_id is fine under BOTH (no over-block). Construction may still fail
    on the actual connect, but NOT with a PolicyDeniedError."""
    from local_deep_research.web_search_engines.engines.search_engine_elasticsearch import (  # noqa: E501
        ElasticsearchSearchEngine,
    )

    snapshot = {"policy.egress_scope": {"value": "both"}}
    assert (
        ElasticsearchSearchEngine._cloud_id_forbidden_by_scope(snapshot)
        is False
    )


# ---------------------------------------------------------------------------
# Review round 3 — regression tests for confirmed findings
# ---------------------------------------------------------------------------

from local_deep_research.security.egress.policy import (  # noqa: E402
    _normalize_alt_ipv4,
)


def test_normalize_alt_ipv4_encodings():
    """Octal / hex / integer IPv4 forms normalize to the dotted-quad the
    resolver would connect to; real hostnames and IPv6 return None."""
    assert _normalize_alt_ipv4("0251.0376.0251.0376") == "169.254.169.254"
    assert _normalize_alt_ipv4("2852039166") == "169.254.169.254"
    assert _normalize_alt_ipv4("0xa9fea9fe") == "169.254.169.254"
    assert _normalize_alt_ipv4("example.com") is None
    assert _normalize_alt_ipv4("::1") is None


def test_evaluate_url_blocks_alt_encoded_metadata_every_scope():
    """R3 #1/#3: octal/integer-encoded cloud-metadata literals are blocked
    under EVERY scope with the explicit blocked_metadata_ip reason — closing
    the evaluate_url 'never permitted' gap that ipaddress-only parsing left
    open, and unifying the denial reason across scopes (no scope_mismatch
    leak for the octal form)."""
    for scope in (
        EgressScope.PUBLIC_ONLY,
        EgressScope.PRIVATE_ONLY,
        EgressScope.BOTH,
        EgressScope.STRICT,
    ):
        ctx = make_ctx(scope=scope)
        for form in ("0251.0376.0251.0376", "2852039166", "0xa9fea9fe"):
            d = evaluate_url(f"http://{form}/", ctx)
            assert not d.allowed, f"{form} allowed under {scope}"
            assert d.reason == "blocked_metadata_ip", (
                f"{form} under {scope} got {d.reason}"
            )


def test_evaluate_url_blocks_metadata_hostnames_and_trailing_dot():
    """R4 #4/#11: complete the metadata invariant — GCP metadata HOSTNAMES
    (metadata.google.internal / .goog, any case, trailing dot) and a
    trailing-dot metadata IP must all deny with blocked_metadata_ip under
    every scope. is_ip_blocked can't see hostnames and inet_aton rejects
    trailing dots, so these dodged the round-1 fix."""
    forms = [
        "metadata.google.internal",
        "METADATA.GOOGLE.INTERNAL.",
        "metadata.goog",
        "169.254.169.254.",
        "0251.0376.0251.0376.",
    ]
    for scope in (
        EgressScope.PUBLIC_ONLY,
        EgressScope.BOTH,
        EgressScope.PRIVATE_ONLY,
        EgressScope.STRICT,
    ):
        ctx = make_ctx(scope=scope)
        for host in forms:
            d = evaluate_url(f"http://{host}/", ctx)
            assert not d.allowed, f"{host} allowed under {scope}"
            assert d.reason == "blocked_metadata_ip", (
                f"{host} under {scope} got {d.reason}"
            )


def test_evaluate_url_metadata_hostname_fix_no_overblock():
    """The metadata-hostname/trailing-dot fix must not over-block a legit
    public host (incl. a fully-qualified trailing dot) or a legit internal
    host that merely contains 'metadata' in its name."""
    ctx_pub = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    assert evaluate_url("http://example.com./", ctx_pub).allowed
    # An unrelated internal host named like metadata.* is NOT the GCP endpoint;
    # it classifies normally (private under PUBLIC_ONLY => scope mismatch, not
    # a metadata block).
    d = evaluate_url("http://metadata.mycorp.local/", ctx_pub)
    assert d.reason != "blocked_metadata_ip"


def test_evaluate_url_alt_encoding_does_not_block_legit_public():
    """The alt-encoding normalization must not over-block a normal public
    IP (it is not a metadata/private address)."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    d = evaluate_url("http://93.184.216.34/", ctx)
    assert d.allowed and d.reason == "allowed_public_host"


def test_evaluate_url_both_scope_allows_public_and_private():
    """R5 #7: BOTH scope allows any classified host — both a public and a
    private host return allowed_both_scope (the deny-side is covered
    elsewhere; this pins the allow-side for the default-ish permissive
    scope)."""
    ctx = make_ctx(scope=EgressScope.BOTH)
    pub = evaluate_url("http://93.184.216.34/", ctx)
    priv = evaluate_url("http://10.0.0.5/", ctx)
    assert pub.allowed and pub.reason == "allowed_both_scope"
    assert priv.allowed and priv.reason == "allowed_both_scope"


def test_context_from_snapshot_allow_dns_false_skips_resolution():
    """R5 #5: allow_dns=False makes ADAPTIVE resolution skip the synchronous
    getaddrinfo for a URL-engine primary (warning-banner render path). With a
    hostname that would need DNS, it falls back to BOTH instead of blocking on
    a lookup — and crucially does NOT call _resolve_with_timeout."""
    from unittest.mock import patch

    snapshot = {
        "policy.egress_scope": "adaptive",
        "search.tool": "searxng",
        "search.engine.web.searxng.default_params.instance_url": (
            "http://searx.example.test/"
        ),
    }
    with patch(
        "local_deep_research.security.egress.policy._resolve_with_timeout"
    ) as mock_dns:
        ctx = context_from_snapshot(snapshot, "searxng", allow_dns=False)
        # The whole point: NO synchronous DNS on the render path.
        mock_dns.assert_not_called()
    # Resolution still completes to a concrete scope (from the engine's static
    # classification), never left as ADAPTIVE and never blocked on DNS.
    assert ctx.scope in (EgressScope.PUBLIC_ONLY, EgressScope.BOTH)


def test_dangerous_scheme_does_not_exhaust_fetch_quota():
    """R3 #9: javascript:/data:/file: hrefs are non-fetchable and common in
    scraped HTML; like unsupported_scheme they must not tick the anti-loop
    quota, so a doc full of data: URIs can't starve later legit fetches."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    for _ in range(MAX_DENIED_FETCHES_PER_RUN + 10):
        evaluate_url("javascript:alert(1)", ctx)
        evaluate_url("data:text/html,<b>x</b>", ctx)
    assert ctx._fetch_denial_count["count"] == 0
    # A legitimate public URL still goes through afterwards.
    d = evaluate_url("http://93.184.216.34/", ctx)
    assert d.allowed


def test_security_denials_still_exhaust_quota_after_dangerous_scheme_change():
    """The quota still fires for genuinely security-relevant denials
    (scope mismatches), so the #9 fix didn't disarm the exhaustion guard."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    for _ in range(MAX_DENIED_FETCHES_PER_RUN + 1):
        evaluate_url("http://10.0.0.5/", ctx)  # private host, public scope
    assert evaluate_url("http://93.184.216.34/", ctx).reason == (
        "denial_quota_exceeded"
    )


def test_evaluate_embeddings_percent_encoded_local_allowed():
    """R3 #5/#11: a percent-encoded local embeddings endpoint must be allowed
    under require_local_embeddings — the HTTP client decodes it before
    connect, so the policy must classify the decoded host (consistency with
    evaluate_url / _classify_engine_url)."""
    ctx = make_ctx(
        scope=EgressScope.PRIVATE_ONLY, require_local_embeddings=True
    )
    snapshot = {"embeddings.openai.base_url": "http://127%2e0%2e0%2e1:1234/v1"}
    d = evaluate_embeddings("openai", ctx, settings_snapshot=snapshot)
    assert d.allowed and d.reason == "provider_local_endpoint"


def test_evaluate_llm_endpoint_percent_encoded_local_allowed():
    """R3 #6/#11: same as above for the LLM endpoint gate."""
    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY, require_local_llm=True)
    snapshot = {"llm.ollama.url": "http://127%2e0%2e0%2e1:11434"}
    d = evaluate_llm_endpoint("ollama", ctx, settings_snapshot=snapshot)
    assert d.allowed and d.reason == "provider_local"


def test_evaluate_embeddings_percent_encoded_cloud_still_blocked():
    """The unquote fix must not let a percent-encoded CLOUD host through
    under require-local."""
    ctx = make_ctx(
        scope=EgressScope.PRIVATE_ONLY, require_local_embeddings=True
    )
    # api.openai.com percent-encoded — decodes to a public host, still denied.
    snapshot = {"embeddings.openai.base_url": "http://api%2eopenai%2ecom/v1"}
    d = evaluate_embeddings("openai", ctx, settings_snapshot=snapshot)
    assert not d.allowed


def test_context_from_snapshot_rejects_non_dict():
    """R3 #10: a non-dict snapshot fails closed with ValueError (the contract
    callers already convert to a hard policy stop), not a bare AttributeError
    that a broad except could swallow into a permissive default."""
    import pytest

    for bad in ("string", 123, ["list"], object()):
        with pytest.raises(ValueError):
            context_from_snapshot(bad, "auto")


def test_dns_cache_first_writer_wins():
    """R3 #2: concurrent disagreeing classifications converge on the first
    writer's value, so a hostname's run classification is stable (no
    last-writer-wins flip that could relax PRIVATE_ONLY on a later fetch)."""
    from local_deep_research.security.egress.policy import (
        _cache_classification,
    )

    ctx = make_ctx(scope=EgressScope.PRIVATE_ONLY)
    assert _cache_classification(ctx, "rr.example", True) is True
    # A later disagreeing writer does not overwrite the pinned value.
    assert _cache_classification(ctx, "rr.example", False) is True
    assert ctx._dns_cache["rr.example"] is True


def test_fetch_quota_is_per_run_not_per_context():
    """Completeness fix: the denied-fetch quota aggregates per RUN (the armed
    active context) so an attacker can't reset the budget by causing a fresh
    EgressContext to be built per engine/fetch. Denials routed through two
    different call-site contexts must share the run budget and exhaust it."""
    from local_deep_research.security.egress.audit_hook import (
        clear_active_context,
        set_active_context,
    )

    run_ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    ctx_a = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    ctx_b = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    set_active_context(run_ctx)
    try:
        half = MAX_DENIED_FETCHES_PER_RUN // 2 + 1
        for _ in range(half):
            evaluate_url("http://10.0.0.5/", ctx_a)  # private host = scope deny
        for _ in range(half):
            evaluate_url("http://10.0.0.6/", ctx_b)
        # Run budget is exhausted even though neither per-call-site context
        # alone issued MAX denials.
        assert run_ctx._fetch_denial_count["count"] >= (
            MAX_DENIED_FETCHES_PER_RUN
        )
        d = evaluate_url("http://93.184.216.34/", ctx_a)
        assert d.reason == "denial_quota_exceeded"
    finally:
        clear_active_context()


def test_fetch_quota_falls_back_to_local_ctx_without_run_context():
    """With no armed run context (snapshot-less / programmatic call), the quota
    stays per-context (unchanged behavior)."""
    ctx = make_ctx(scope=EgressScope.PUBLIC_ONLY)
    for _ in range(5):
        evaluate_url("http://10.0.0.5/", ctx)
    assert ctx._fetch_denial_count["count"] == 5


# ---------------------------------------------------------------------------
# resolve_run_primary_engine: the single source of truth for a run's primary
# engine (what ADAPTIVE classifies). Every run-scoped EgressContext builder
# routes through this so the scope can't be resolved inconsistently across
# layers (factory PEP vs. LangGraph tool-list pre-filter).
# ---------------------------------------------------------------------------


class TestResolveRunPrimaryEngine:
    def test_reads_flat_search_tool(self):
        assert resolve_run_primary_engine({"search.tool": "pubmed"}) == "pubmed"

    def test_reads_nested_value_shape(self):
        # The {"value": ...} settings shape is unwrapped like everywhere else.
        assert (
            resolve_run_primary_engine({"search.tool": {"value": "paperless"}})
            == "paperless"
        )

    # --- Fail closed: a missing/empty primary must NOT silently become a
    # default (public) engine, which would set the egress scope from an engine
    # the user never chose. Run-level callers pass no default and must raise.

    def test_missing_key_raises(self):
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({})

    def test_none_snapshot_raises(self):
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine(None)

    def test_empty_value_raises(self):
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({"search.tool": ""})

    def test_blank_nested_value_raises(self):
        # A {"value": ""} shape unwraps to "" — still no configured primary.
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({"search.tool": {"value": ""}})

    # --- Explicit default escape hatch: the factory passes default=engine_name
    # because it is evaluating that one specific engine.

    def test_custom_default_used_when_missing(self):
        assert resolve_run_primary_engine({}, default="arxiv") == "arxiv"

    def test_custom_default_used_when_empty(self):
        assert (
            resolve_run_primary_engine({"search.tool": ""}, default="arxiv")
            == "arxiv"
        )

    def test_explicit_value_wins_over_custom_default(self):
        assert (
            resolve_run_primary_engine(
                {"search.tool": "wikipedia"}, default="arxiv"
            )
            == "wikipedia"
        )

    # --- A truthy-but-unusable primary (whitespace / non-string) must NOT
    # slip past the fail-closed guard and classify to the permissive BOTH.

    def test_whitespace_only_raises(self):
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({"search.tool": "   "})

    def test_whitespace_is_stripped(self):
        assert (
            resolve_run_primary_engine({"search.tool": "  pubmed  "})
            == "pubmed"
        )

    def test_non_string_raises(self):
        for bad in (5, True, ["arxiv"], {"nested": 1}):
            with pytest.raises(ValueError, match="no primary search engine"):
                resolve_run_primary_engine({"search.tool": bad})

    def test_nested_whitespace_value_raises(self):
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({"search.tool": {"value": "  "}})

    def test_empty_default_is_treated_as_no_default(self):
        # default="" must not satisfy the fail-closed guard (the old
        # `if default is not None` would have returned "").
        with pytest.raises(ValueError, match="no primary search engine"):
            resolve_run_primary_engine({}, default="")

    def test_whitespace_default_falls_through_to_value(self):
        # A blank default doesn't mask a real primary.
        assert (
            resolve_run_primary_engine({"search.tool": "arxiv"}, default="   ")
            == "arxiv"
        )