tests/test_real_user_queries.py

"""Real public-user source-backed query tests."""

from __future__ import annotations

import json
import re
from pathlib import Path
from datetime import datetime, timezone

import pytest

from coastwise.qna import answer_question, answer_to_dict
from coastwise.source_cache import DEFAULT_SEED_CACHE_DIR, load_seed_cache

FIXTURE_PATH = Path(__file__).parent / "fixtures" / "real_user_queries.json"
NOW = datetime(2026, 6, 14, 13, tzinfo=timezone.utc)
NEGATED_FORBIDDEN_PREFIX_PATTERNS = (
    r"(?:^|\W)not(?:\s+\w+){0,3}\s*$",
    r"(?:^|\W)does\s+not(?:\s+\w+){0,6}\s*$",
    r"(?:^|\W)do\s+not(?:\s+\w+){0,6}\s*$",
    r"(?:^|\W)cannot(?:\s+\w+){0,6}\s*$",
    r"(?:^|\W)can't(?:\s+\w+){0,6}\s*$",
    r"(?:^|\W)no(?:\s+\w+){0,3}\s*$",
    r"(?:^|\W)without(?:\s+\w+){0,3}\s*$",
    r"(?:^|\W)avoid\s+implying(?:\s+\w+){0,6}\s*$",
)
DUNGENESS_GENERIC_FALLBACK = "dungeness_generic_fallback"
CONTRASTIVE_BOUNDARY_PATTERN = re.compile(r",?\s+(?:but|however|yet|nevertheless)\s+")


def _cases() -> list[dict[str, object]]:
    return json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))


def _contains_forbidden_answer_claim(answer_copy: str, forbidden_term: str) -> bool:
    lowered_answer = answer_copy.casefold()
    lowered_term = forbidden_term.casefold()
    term_pattern = re.compile(rf"(?<!\w){re.escape(lowered_term)}(?!\w)")
    for match in term_pattern.finditer(lowered_answer):
        index = match.start()
        segment_start = max(lowered_answer.rfind(separator, 0, index) for separator in ".!?;\n")
        scope_start = segment_start + 1
        segment_before_term = lowered_answer[scope_start:index]
        for boundary in CONTRASTIVE_BOUNDARY_PATTERN.finditer(segment_before_term):
            scope_start = segment_start + 1 + boundary.end()
        segment_prefix = lowered_answer[scope_start:index].strip()
        if not any(re.search(pattern, segment_prefix) for pattern in NEGATED_FORBIDDEN_PREFIX_PATTERNS):
            return True
    return False


def _has_concrete_dungeness_result(lowered_answer: str) -> bool:
    conditional_prefix_pattern = re.compile(
        r"\b(?:check|verify|confirm)\s+whether\b|\b(?:if|whether|when)\b"
    )
    for pattern in (
        r"\bharvest\s+is\s+(?:open|closed)\b",
        r"\b(?:season\s+)?is\s+(?:open|closed)\b",
        r"\bseason\s+(?:opens|closes)\s+on\s+\S+",
    ):
        for match in re.finditer(pattern, lowered_answer):
            segment_start = max(
                lowered_answer.rfind(separator, 0, match.start())
                for separator in ".!?;\n"
            )
            segment_prefix = lowered_answer[segment_start + 1 : match.start()]
            segment_suffix = lowered_answer[match.end() :]
            if re.match(r"\s+or\s+(?:open|closed)\b", segment_suffix):
                continue
            if not re.search(r"\b(?:if|whether|when)\b", segment_prefix):
                return True

    # Numeric plausibility belongs to source-backed Q&A assertions; this guard
    # only distinguishes answer-shaped facts from generic topics or years.
    for pattern in (
        r"\bbag\s+limit\s+(?:is|:)\s*\d+\b",
        r"\bpossession\s+limit\s+(?:is|:)\s*\d+\b",
        r"\b(?:minimum\s+)?size\s+limit\s+(?:is|:)\s*\d+(?:\.\d+)?",
        r"\bminimum\s+size\s+(?:is|:)\s*\d+(?:\.\d+)?",
    ):
        for match in re.finditer(pattern, lowered_answer):
            segment_start = max(
                lowered_answer.rfind(separator, 0, match.start())
                for separator in ".!?;\n"
            )
            segment_prefix = lowered_answer[segment_start + 1 : match.start()]
            if not conditional_prefix_pattern.search(segment_prefix):
                return True
    return False


def _matches_forbidden_direct_answer_concept(direct_answer: str, concept: str) -> bool:
    lowered_answer = direct_answer.casefold()
    if concept != DUNGENESS_GENERIC_FALLBACK:
        raise AssertionError(f"Unknown forbidden direct answer concept: {concept}")
    if _has_concrete_dungeness_result(lowered_answer):
        return False

    has_dungeness_subject = re.search(r"\bdungeness\s+crabs?\b", lowered_answer) is not None
    has_subject = (
        has_dungeness_subject or re.search(r"\bcrabs?\b", lowered_answer) is not None
    )
    has_bare_topic_fallback = has_dungeness_subject and any(
        re.search(pattern, lowered_answer)
        for pattern in (
            r"\bdungeness\s+crabs?\s+season\s*[.!?]?\s*$",
            r"\bdungeness\s+crabs?\s+trap\s+rules?\s*[.!?]?\s*$",
            r"\bdungeness\s+crabs?\s+season\s+and\s+(?:trap\s+)?rules?\b",
            r"\b\d{4}\s+dungeness\s+crabs?\s+season\b",
            r"\bdungeness\s+crabs?\s+harvest\s+status\b",
        )
    )
    if has_bare_topic_fallback:
        return True

    has_action = (
        re.search(r"\b(?:verify|check|confirm|consult|review)\b", lowered_answer)
        is not None
    )
    has_authority = re.search(r"\b(?:cdfw|official|state)\b", lowered_answer) is not None
    rule_matches = [
        pattern
        for pattern in (
            r"\bseason\b",
            r"\brules?\b",
            r"\bregulations?\b",
            r"\bgear\b",
            r"\btraps?\b",
            r"\bhoop[- ]nets?\b",
        )
        if re.search(pattern, lowered_answer)
    ]
    has_harvest_context = re.search(
        r"\b(?:harvest|today|open|closed)\b|before\s+harvest", lowered_answer
    ) is not None
    has_harvest_open_closed_context = (
        re.search(r"\bharvest\b", lowered_answer) is not None
        and re.search(r"\b(?:open\s+or\s+closed|closed\s+or\s+open)\b", lowered_answer)
        is not None
    )
    has_open_closed_alternative_fallback = (
        has_subject
        and re.search(r"\b(?:season|harvest)\b", lowered_answer) is not None
        and re.search(r"\b(?:open\s+or\s+closed|closed\s+or\s+open)\b", lowered_answer)
        is not None
    )
    has_harvest_status_context = (
        re.search(r"\bharvest(?:ing)?\b", lowered_answer) is not None
        and re.search(r"\bstatus\b", lowered_answer) is not None
    )
    has_unavailable_status_context = has_harvest_status_context and (
        re.search(
            r"\b(?:unavailable|unknown|unclear|undetermined)\b|"
            r"\bnot\s+(?:available|known)\b|"
            r"\bno\b.{0,80}\bstatus\s+is\s+available\b|"
            r"\b(?:cannot|can't)\s+(?:be\s+)?determin(?:e|ed)\b",
            lowered_answer,
        )
        is not None
    )
    has_harvest_intent_context = (
        has_harvest_status_context
        or (
            re.search(r"\bharvest(?:ing)?\b", lowered_answer) is not None
            and re.search(
                r"\b(?:allowed|today|whether)\b|before\s+harvest(?:ing)?",
                lowered_answer,
            )
            is not None
        )
    )
    has_check_or_topic_fallback = (
        has_dungeness_subject
        and has_harvest_context
        and re.search(r"\b(?:check|verify|confirm)\b|\b\d{4}\b", lowered_answer)
        is not None
    )
    has_rule_context = (
        len(rule_matches) >= 2
        or (bool(rule_matches) and has_harvest_context)
        or has_harvest_open_closed_context
        or has_harvest_intent_context
    )
    has_action_fallback = all(
        (
            has_action,
            has_authority or has_harvest_status_context or has_check_or_topic_fallback,
            has_subject,
            has_rule_context,
        )
    )
    has_unavailable_status_fallback = (
        has_subject and has_harvest_status_context and has_unavailable_status_context
    )
    return (
        has_action_fallback
        or has_unavailable_status_fallback
        or has_open_closed_alternative_fallback
    )


def test_forbidden_answer_claim_helper_blocks_affirmative_claims():
    assert _contains_forbidden_answer_claim("CDFW authorized harvest today.", "authorized")
    assert _contains_forbidden_answer_claim("CDPH certified safe shellfish.", "certified safe")
    assert _contains_forbidden_answer_claim("Mussels are safe to eat.", "are safe to eat")


def test_forbidden_answer_claim_helper_allows_negated_or_cautious_claims():
    assert not _contains_forbidden_answer_claim("CDFW does not authorize harvest.", "authorize")
    assert not _contains_forbidden_answer_claim("This is not authorized guidance.", "authorized")
    assert not _contains_forbidden_answer_claim("CDPH does not certify harvest is safe.", "certify")
    assert not _contains_forbidden_answer_claim(
        "CDPH does not certify that shellfish are safe to eat.", "safe to eat"
    )
    assert not _contains_forbidden_answer_claim(
        "CDPH does not certify that shellfish are safe for consumption.", "safe for consumption"
    )
    assert not _contains_forbidden_answer_claim("Shellfish are not certified safe.", "certified safe")
    assert not _contains_forbidden_answer_claim("Mussels are not currently safe to eat.", "safe to eat")
    assert not _contains_forbidden_answer_claim("Mussels are not safe to eat today.", "are safe to eat")
    assert not _contains_forbidden_answer_claim("Mussels are not safe to eat today.", "safe to eat")
    assert not _contains_forbidden_answer_claim("Cannot determine safe to eat from cached evidence.", "safe to eat")
    assert not _contains_forbidden_answer_claim("Avoid implying authorized harvest.", "authorized")
    assert not _contains_forbidden_answer_claim("No authorized harvest is shown.", "authorized")
    assert not _contains_forbidden_answer_claim("This is unauthorized guidance.", "authorized")


def test_forbidden_answer_claim_helper_blocks_unnegated_repeated_claims():
    answer_copy = "This is not authorized guidance. You are authorized to harvest."
    assert _contains_forbidden_answer_claim(answer_copy, "authorized")
    assert _contains_forbidden_answer_claim(
        "This is not authorized guidance, but you are authorized to harvest.", "authorized"
    )
    safety_copy = "CDPH does not certify that mussels are safe to eat, but they are safe to eat today."
    assert _contains_forbidden_answer_claim(safety_copy, "safe to eat")
    assert _contains_forbidden_answer_claim(safety_copy, "are safe to eat")
    assert _contains_forbidden_answer_claim(
        "CDPH does not certify shellfish, but mussels are safe to eat.", "safe to eat"
    )


def test_direct_answer_concept_helper_blocks_reworded_dungeness_generic_fallback():
    direct_answer = "Verify current CDFW Dungeness crab season and trap rules before harvest."
    assert _matches_forbidden_direct_answer_concept(direct_answer, "dungeness_generic_fallback")
    assert _matches_forbidden_direct_answer_concept(
        "Check CDFW Dungeness crab season and rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Confirm the latest official Dungeness crab season and gear rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check the 2025 CDFW Dungeness crab season and trap rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check whether Dungeness crab bag limit is 10 before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check the 2025 Dungeness crab season before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Verify current CDFW Dungeness crab size limit and season rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Verify current CDFW Dungeness crab season is open or closed before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check whether the CDFW Dungeness crab season is open or closed before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check CDFW to confirm whether Dungeness crab harvest is open or closed before harvest.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check CDFW before harvesting Dungeness crab today.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Confirm with CDFW whether Dungeness crab harvest is allowed today.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check current official Dungeness crab harvest status.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check current Dungeness crab harvest status.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Check Dungeness crab harvest status.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest status is unavailable.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Current Dungeness crab harvest status is unknown.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest status cannot be determined.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest status is not known.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "No Dungeness crab harvest status is available.",
        "dungeness_generic_fallback",
    )


def test_direct_answer_concept_helper_blocks_bare_dungeness_open_closed_alternatives():
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab season is open or closed.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest is open or closed.",
        "dungeness_generic_fallback",
    )


def test_direct_answer_concept_helper_blocks_bare_dungeness_topic_fallbacks():
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab season.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab trap rules.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab season and trap rules.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "2025 Dungeness crab season.",
        "dungeness_generic_fallback",
    )
    assert _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest status.",
        "dungeness_generic_fallback",
    )


def test_direct_answer_concept_helper_allows_concrete_dungeness_results():
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab season is closed; verify current CDFW rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab season is open according to cached CDFW evidence; "
        "check current official rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest is closed; verify current CDFW rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab harvest is open according to cached CDFW evidence; "
        "check current official rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab bag limit is 10; check current CDFW rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab bag limit is 10; verify current CDFW rules before harvest.",
        "dungeness_generic_fallback",
    )
    assert not _matches_forbidden_direct_answer_concept(
        "Dungeness crab minimum size is 22 inches; verify current official rules.",
        "dungeness_generic_fallback",
    )


@pytest.mark.parametrize("case", _cases(), ids=lambda case: case["query"])
def test_real_user_queries_use_saved_evidence_without_generic_answers(case):
    model_payload = case.get("model_interpretation")
    model = (lambda _prompt, payload=model_payload: json.dumps(payload)) if model_payload else (lambda _prompt: None)
    answer = answer_question(
        str(case["query"]),
        "",
        load_seed_cache(DEFAULT_SEED_CACHE_DIR),
        NOW,
        model=model,
    )
    payload = answer_to_dict(answer)
    source_material = " ".join(
        [
            " ".join(payload["source_titles"]),
            " ".join(payload["source_links"]),
            " ".join(payload["supporting_snippets"]),
        ]
    ).casefold()
    answer_copy = " ".join(
        [
            payload["direct_answer"] or "",
            " ".join(payload["uncertainty_notes"]),
            payload["next_safe_action"] or "",
        ]
    ).casefold()
    direct_answer = (payload["direct_answer"] or "").casefold()
    expected_species_variants = {
        str(case["expected_species"]),
        str(case["expected_species"]).replace("_", " ").title(),
        str(case["expected_species"]).replace("_", " "),
    }
    assert str(payload["species_or_category"]).casefold() in {
        species.casefold() for species in expected_species_variants
    }
    if case["expected_place"] is not None:
        assert payload["matched_place"] == case["expected_place"]
    else:
        assert payload["matched_place"] is None
    assert payload["status"] in case["expected_statuses"]
    assert payload["source_links"]
    assert payload["supporting_snippets"]
    assert payload["retrieved_at"]
    for term in case["required_source_terms"]:
        assert term.casefold() in source_material
    for term in case["required_answer_terms"]:
        assert term.casefold() in answer_copy
    for term in case["forbidden_answer_terms"]:
        assert not _contains_forbidden_answer_claim(answer_copy, term)
    for concept in case.get("forbidden_direct_answer_concepts", []):
        assert not _matches_forbidden_direct_answer_concept(direct_answer, concept)