"""Source-backed Q&A tests.""" import json from dataclasses import replace from datetime import datetime, timedelta, timezone import pytest from coastwise.qna import answer_question, answer_to_dict, interpret_question from coastwise.schemas import ( AnswerStatus, EvidenceType, FreshnessStatus, RequestedFactType, StructuredRegulationFact, ) from coastwise.source_cache import DEFAULT_SEED_CACHE_DIR, SourceCache, load_seed_cache NOW = datetime(2026, 6, 14, 13, tzinfo=timezone.utc) def test_answer_question_returns_structured_lingcod_minimum_size(): answer = answer_question( "what's the min size of lingcod?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, ) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert "22" in answer.direct_answer assert "Lingcod" == answer.species_or_category assert answer.source_links assert answer.freshness_status is FreshnessStatus.CURRENT assert "verify" in answer.next_safe_action.lower() def test_answer_question_uses_snippet_fallback_for_supported_regional_content(): answer = answer_question( "what should I check for ocean rules near Monterey?", "Monterey", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, ) assert answer.status is AnswerStatus.PARTIAL assert answer.evidence_type is EvidenceType.SOURCE_SNIPPET assert answer.supporting_snippets assert any("structured" in note.lower() for note in answer.uncertainty_notes) assert answer.location_context.matched_place == "Monterey" def test_answer_question_declines_unsupported_scope_and_missing_cache(): answer = answer_question( "what is the trout limit in Lake Tahoe?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, ) assert answer.status is AnswerStatus.UNSUPPORTED assert answer.evidence_type is EvidenceType.UNSUPPORTED assert "outside" in answer.direct_answer.lower() assert "official" in answer.next_safe_action.lower() def test_supported_question_without_location_keeps_local_rules_warning(): answer = answer_question( "bag limit for Dungeness crab?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, ) assert answer.status in {AnswerStatus.ANSWERED, AnswerStatus.PARTIAL} assert answer.location_context.confidence == "absent" assert "local" in (answer.location_context.warning or "").lower() def test_offline_stale_and_no_cache_answers_are_conservative(): seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR) stale = SourceCache( sources=seed.sources, snapshots=tuple(replace(snapshot, retrieved_at=NOW - timedelta(hours=25)) for snapshot in seed.snapshots), chunks=seed.chunks, facts=seed.facts, advisory_facts=seed.advisory_facts, ) stale_answer = answer_question("what's the min size of lingcod?", "", stale, NOW) no_cache_answer = answer_question("what's the min size of lingcod?", "", SourceCache((), (), (), ()), NOW) assert stale_answer.freshness_status is FreshnessStatus.STALE assert "stale" in stale_answer.next_safe_action.lower() assert no_cache_answer.status is AnswerStatus.NO_CACHE assert "refresh" in no_cache_answer.next_safe_action.lower() def test_cdph_shellfish_question_keeps_verification_prominent_without_certifying_food_safety(): answer = answer_question( "can I eat mussels from Pacifica today?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, ) lowered = answer.direct_answer.lower() assert answer.status in {AnswerStatus.ANSWERED, AnswerStatus.PARTIAL} assert any("cdph" in link.lower() for link in answer.source_links) assert "cooking does not reliably" in " ".join(answer.supporting_snippets).lower() assert "verify" in answer.next_safe_action.lower() assert "safe to eat" not in lowered assert "certifies" not in lowered def test_cached_evidence_packet_retrieves_crab_season_context_without_exact_fact(): from coastwise.qna import build_cached_evidence_packet, interpret_question cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) interpretation = interpret_question("is dungenese crab in season for fort Bragg", "", cache) packet = build_cached_evidence_packet(interpretation, cache, NOW) assert packet is not None assert packet.species_or_category == "dungeness_crab" assert packet.matched_place == "Fort Bragg" assert packet.requested_fact_type.value == "season" assert any("Dungeness crab season" in source.excerpt for source in packet.sources) def test_dungeness_season_answer_uses_saved_evidence_instead_of_generic_verify_fallback(): answer = answer_question( "is dungenese crab in season for fort Bragg", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, model=lambda _prompt: None, ) assert answer.status is AnswerStatus.PARTIAL assert answer.evidence_type is EvidenceType.SOURCE_SNIPPET assert answer.location_context.matched_place == "Fort Bragg" assert any("clear open or closed season date" in snippet for snippet in answer.supporting_snippets) assert "season is open" not in answer.direct_answer.casefold() assert "Dungeness crab: Verify current CDFW" not in answer.direct_answer def test_structured_lingcod_answer_uses_exact_fact_after_invalid_model_interpretation(): payload = { "species_or_category": "maine_lobster", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } answer = answer_question( "what's the min size of lingcod?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, model=lambda _prompt: json.dumps(payload), ) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert answer.direct_answer == "Lingcod minimum size is 22 inches total length." assert not answer.model_assisted def test_harvest_question_with_typos_answers_with_dated_sf_dungeness_season(): answer = answer_question( "can i harvest dugenese crabs in pacifica today?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, model=lambda _prompt: None, ) # Pacifica resolves to the San Francisco area, which now has a dated season fact. assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert answer.location_context.matched_place == "Pacifica" lowered = answer.direct_answer.casefold() assert "season is open" in lowered assert "november" in lowered # real SF-area Dungeness season dates # States the season; never asserts personal legality to harvest. assert "you may harvest" not in lowered assert "legal to harvest" not in lowered def test_interpret_question_uses_model_for_messy_species_location_and_intent(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season", "closure"], "confidence": "high", "notes": "Normalized misspellings.", } # No explicit species token, so deterministic-first routes this to the model # (recognizable or misspelled species resolve without it). interpretation = interpret_question( "im in mandocino. is it legal to harvest them right now?", "", cache, model=lambda _prompt: json.dumps(payload), ) assert interpretation.model_used assert interpretation.species_or_category == "dungeness_crab" assert interpretation.location_context.matched_place == "Mendocino" assert interpretation.location_context.region_key == "mendocino" assert interpretation.requested_fact_type is RequestedFactType.SEASON assert "permission_status" in (interpretation.model_notes or "") def test_interpret_question_rejects_model_generic_regional_species_term(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "regional", "location_name": "Monterey", "region_key": "central", "intent": "regional_rules", "requested_fact_types": ["general_summary"], "confidence": "high", } interpretation = interpret_question( "what should I check near Monterey?", "Monterey", cache, model=lambda _prompt: json.dumps(payload), ) assert not interpretation.model_used assert interpretation.species_or_category is None def test_interpret_question_rejects_model_unknown_species_and_uses_fallback(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "maine_lobster", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } interpretation = interpret_question( "can I harvest lobster in mendocino?", "", cache, model=lambda _prompt: json.dumps(payload), ) assert not interpretation.model_used assert interpretation.species_or_category is None def test_answer_question_uses_model_path_for_source_backed_dungeness_mendocino(): payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season", "closure"], "confidence": "high", "notes": "Normalized misspellings.", } answer = answer_question( "im in mandocino. is it legal to harvest dungenese crabs?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, model=lambda _prompt: json.dumps(payload), ) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert answer.source_links assert answer.retrieved_at is not None assert answer.freshness_status is FreshnessStatus.CURRENT assert answer.next_safe_action assert "legal to harvest" not in answer.direct_answer.casefold() assert answer.location_context.matched_place == "Mendocino" def test_answer_question_rejects_unknown_model_species_without_lobster_source_answer(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "maine_lobster", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } interpretation = interpret_question( "can I harvest maine lobster in mendocino?", "", cache, model=lambda _prompt: json.dumps(payload), ) answer = answer_question( "can I harvest maine lobster in mendocino?", "", cache, NOW, model=lambda _prompt: json.dumps(payload), ) assert not interpretation.model_used assert interpretation.species_or_category is None assert answer.status is AnswerStatus.UNSUPPORTED assert answer.evidence_type is EvidenceType.UNSUPPORTED assert answer.species_or_category is None assert not answer.source_links def test_permission_question_surfaces_statewide_facts_and_flags_missing_area_season(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season", "closure"], "confidence": "high", } answer = answer_question( "im in mandocino. is it legal to harvest dungenese crabs?", "", cache, NOW, model=lambda _prompt: json.dumps(payload), ) # Statewide bag/size apply in Mendocino and are surfaced; the area-specific # season we lack for Mendocino is flagged, and legality is never asserted. assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert answer.location_context.matched_place == "Mendocino" assert "10 crab" in answer.direct_answer.casefold() notes = " ".join(answer.uncertainty_notes).casefold() assert "season" in notes or "closure" in notes assert "legal to harvest" not in answer.direct_answer.casefold() assert not any("cabezon" in snippet.casefold() for snippet in answer.supporting_snippets) assert any("crab" in snippet.casefold() for snippet in answer.supporting_snippets) def test_answer_question_unsupported_scope_cannot_be_overridden_by_model(): payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } answer = answer_question( "what is the trout limit in freshwater Lake Tahoe?", "", load_seed_cache(DEFAULT_SEED_CACHE_DIR), NOW, model=lambda _prompt: json.dumps(payload), ) assert answer.status is AnswerStatus.UNSUPPORTED assert answer.evidence_type is EvidenceType.UNSUPPORTED assert answer.location_context.confidence == "unsupported" assert answer.location_context.matched_place is None assert not answer.source_links def test_permission_question_answers_directly_when_season_fact_exists(): seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR) season_fact = StructuredRegulationFact( id="dungeness_crab_mendocino_season_status", species_or_category="dungeness_crab", display_name="Dungeness crab", aliases=("dungeness crabs",), area_or_scope="Mendocino California ocean region", fact_type=RequestedFactType.SEASON, value="season status is closed in the cached official source text", units=None, source_id="cdfw_crabs", source_url="https://wildlife.ca.gov/Conservation/Marine/Invertebrates/Crabs", retrieved_at=NOW, supporting_chunk_id="chunk_dungeness_crab_season_context", supporting_snippet="Dungeness crab season status is closed in this test source text.", region_key="mendocino", validation_set=True, ) cache = SourceCache( sources=seed.sources, snapshots=seed.snapshots, chunks=seed.chunks, facts=(*seed.facts, season_fact), advisory_facts=seed.advisory_facts, ) payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season", "closure"], "confidence": "high", } answer = answer_question( "im in mandocino. is it legal to harvest dungenese crabs?", "", cache, NOW, model=lambda _prompt: json.dumps(payload), ) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT # Source is attributed via source_titles, not an inline preamble. assert any("CDFW" in title for title in answer.source_titles) assert "closed" in answer.direct_answer assert "legal to harvest" not in answer.direct_answer.casefold() assert answer.location_context.region_key == "mendocino" def test_answer_to_dict_sanitizes_structured_bundle_forbidden_copy(): """A structured-fact value carrying forbidden copy is sanitized at serialization.""" seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR) poisoned = StructuredRegulationFact( id="dungeness_crab_mendocino_poisoned", species_or_category="dungeness_crab", display_name="Dungeness crab", aliases=("dungeness crabs",), area_or_scope="Mendocino California ocean region", fact_type=RequestedFactType.SEASON, value="legal to harvest now per cached source", units=None, source_id="cdfw_crabs", source_url="https://wildlife.ca.gov/Conservation/Marine/Invertebrates/Crabs", retrieved_at=NOW, supporting_chunk_id="chunk_dungeness_crab_season_context", supporting_snippet="Test poisoned snippet.", region_key="mendocino", validation_set=True, ) cache = SourceCache( sources=seed.sources, snapshots=seed.snapshots, chunks=seed.chunks, facts=(*seed.facts, poisoned), advisory_facts=seed.advisory_facts, ) payload = { "species_or_category": "dungeness_crab", "location_name": "Mendocino", "region_key": "mendocino", "intent": "permission_status", "requested_fact_types": ["season", "closure"], "confidence": "high", } answer = answer_question( "im in mandocino. is it legal to harvest dungenese crabs?", "", cache, NOW, model=lambda _prompt: json.dumps(payload), ) # The raw bundle answer is ANSWERED and carries the forbidden phrase verbatim... assert answer.status is AnswerStatus.ANSWERED assert "legal to harvest" in answer.direct_answer.casefold() # ...but the serialization boundary the UI uses sanitizes and downgrades it. serialized = answer_to_dict(answer) assert serialized["status"] == "partial" assert "legal to harvest" not in serialized["direct_answer"].casefold() def test_cabezon_evidence_excludes_unrelated_species_snippets(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "cabezon", "intent": "general", "requested_fact_types": ["advisory"], "confidence": "high", } answer = answer_question( "is it legal to catch cabazon in pacifica?", "", cache, NOW, model=lambda _prompt: json.dumps(payload), ) joined = " ".join(answer.supporting_snippets).casefold() assert "cabezon" in joined assert "dungeness" not in joined assert "quarantine" not in joined def test_model_path_uses_deterministic_fact_type_not_model_advisory_misclassification(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) payload = { "species_or_category": "cabezon", "intent": "general", "requested_fact_types": ["advisory"], "confidence": "high", } # Underspecified species -> routes to the model, which mislabels the fact type # as advisory; the deterministic fact type must still win. interpretation = interpret_question( "is it legal to catch one in pacifica?", "", cache, model=lambda _prompt: json.dumps(payload), ) assert interpretation.model_used assert interpretation.species_or_category == "cabezon" assert interpretation.location_context.matched_place == "Pacifica" # the model mislabeled a catch question as "advisory"; deterministic wins assert interpretation.requested_fact_type is RequestedFactType.SEASON assert "advisory" not in (interpretation.model_notes or "") def test_permission_question_surfaces_all_available_species_facts(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question("is it legal to catch lingcod?", "", cache, NOW) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT lowered = answer.direct_answer.casefold() assert "22" in answer.direct_answer assert "bag limit" in lowered def test_permission_question_answers_cabezon_with_size_bag_and_season(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question("is it legal to catch cabezon in pacifica?", "", cache, NOW) assert answer.status is AnswerStatus.ANSWERED assert answer.evidence_type is EvidenceType.STRUCTURED_FACT assert answer.species_or_category == "Cabezon" lowered = answer.direct_answer.casefold() # Source-backed cabezon regulation values (CDFW Groundfish Summary). assert "no minimum size limit" in lowered assert "10" in answer.direct_answer # RCG-complex aggregate bag limit assert "april" in lowered # RCG-complex season now encoded for the SF management area assert any("groundfish" in title.casefold() for title in answer.source_titles) # States facts, never a bare legality verdict. assert "you may harvest" not in lowered assert "authorized" not in lowered EXPANDED_SEED_CASES = [ ("what is the minimum size for california halibut?", "22"), ("white seabass size limit", "28"), ("leopard shark bag limit", "3"), ("rock crab bag limit", "35"), ("purple sea urchin limit", "35"), ("can i take red abalone?", "closed"), ("is it legal to catch white sturgeon?", "catch-and-release"), ("kelp harvest limit", "10 pounds"), ("is it legal to catch rockfish in pacifica?", "no minimum size limit"), ("redtail surfperch minimum size", "10"), ] def test_general_rules_question_surfaces_real_facts_not_generic_placeholder(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question("dungeness crab rules", "", cache, NOW) assert answer.status is AnswerStatus.ANSWERED lowered = answer.direct_answer.casefold() # Real values, not the bare "verify current CDFW Dungeness crab season..." placeholder. assert "10 crab" in lowered assert "5¾" in answer.direct_answer assert "season is open" in lowered def test_model_species_guard_overrides_lexically_implausible_resolution(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) # The small model rounds a rare species to a wrong in-enum species (mussels) # that has no lexical basis in the question. payload = { "species_or_category": "mussels", "location_name": None, "region_key": None, "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } model = lambda _prompt: json.dumps(payload) interpretation = interpret_question( "is it legal to catch soupfin shark?", "", cache, model=model ) assert interpretation.species_or_category != "mussels" answer = answer_question( "is it legal to catch soupfin shark?", "", cache, NOW, model=model ) # Must never answer a shark question with the mussel food-safety advisory. assert "mussel" not in answer.direct_answer.casefold() def test_model_species_guard_prefers_specific_alias_over_model_substring_pick(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) # The model rounds "kelp bass" to the plant "kelp"; the specific fact alias # ("kelp bass") must win over the model's substring pick. payload = { "species_or_category": "kelp_plants", "intent": "general", "requested_fact_types": ["minimum_size"], "confidence": "high", } interpretation = interpret_question( "what is the size limit for kelp bass?", "", cache, model=lambda _prompt: json.dumps(payload), ) assert interpretation.species_or_category == "kelp_bass" def test_model_species_guard_keeps_typo_normalization_when_search_is_empty(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) # Deterministic name search can't resolve the typo, so the model's # normalization (cabazon -> cabezon) must be preserved. payload = { "species_or_category": "cabezon", "location_name": "Pacifica", "region_key": "san_francisco", "intent": "permission_status", "requested_fact_types": ["season"], "confidence": "high", } interpretation = interpret_question( "is it legal to catch cabazon in pacifica?", "", cache, model=lambda _prompt: json.dumps(payload), ) assert interpretation.species_or_category == "cabezon" def test_species_without_facts_falls_back_to_reviewed_rule_card_not_wrong_species(): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question( "is it legal to catch california scorpionfish?", "", cache, NOW ) lowered = answer.direct_answer.casefold() # Recognized species with no extracted fact -> reviewed card posture, not a guess. assert answer.status is AnswerStatus.PARTIAL assert "scorpionfish" in answer.species_or_category.casefold() assert "rules apply" in lowered or "verify" in lowered # Must never resolve to an unrelated species' facts. assert "halibut" not in lowered assert "22 inches" not in lowered def test_allowed_species_covers_full_rule_card_taxonomy(): from coastwise.data import all_rule_cards from coastwise.qna import _allowed_species_from_cache cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) allowed = _allowed_species_from_cache(cache) # Every curated species the deterministic path knows must also be offered to # the model, so it cannot round an unknown species to a wrong neighbor. for card in all_rule_cards(): if card.key == "unknown": continue assert card.key in allowed, f"{card.key} missing from allowed species" STATEWIDE_SEED_CASES = [ ("kelp bass size limit", "14"), ("california sheephead bag limit", "2"), ("pacific bonito minimum size", "24"), ("california yellowtail bag limit", "10"), ("ocean whitefish size limit", "no minimum size limit"), ("california spiny lobster season", "october"), ("market squid limit", "no daily bag limit"), ("geoduck clam limit", "3"), ("white croaker limit", "10"), ("is it legal to catch giant sea bass?", "prohibited"), ("is it legal to catch white shark?", "prohibited"), ("california grunion season", "april"), ] @pytest.mark.parametrize("query, must_contain", STATEWIDE_SEED_CASES) def test_statewide_seed_answers_real_values(query, must_contain): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question(query, "", cache, NOW) assert answer.status is AnswerStatus.ANSWERED, f"{query!r} -> {answer.status}: {answer.direct_answer}" assert ( must_contain.casefold() in answer.direct_answer.casefold() ), f"{query!r} -> {answer.direct_answer}" lowered = answer.direct_answer.casefold() assert "you may harvest" not in lowered assert "is safe to eat" not in lowered @pytest.mark.parametrize("query, must_contain", EXPANDED_SEED_CASES) def test_expanded_seed_answers_real_values(query, must_contain): cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) answer = answer_question(query, "", cache, NOW) assert answer.status is AnswerStatus.ANSWERED, f"{query!r} -> {answer.status}: {answer.direct_answer}" assert ( must_contain.casefold() in answer.direct_answer.casefold() ), f"{query!r} -> {answer.direct_answer}" # Source-backed answers never assert legality or food safety. lowered = answer.direct_answer.casefold() assert "you may harvest" not in lowered assert "is safe to eat" not in lowered