diff --git a/src/rag/qa_v1_1_73/soft_gate_policy.py b/src/rag/qa_v1_1_73/soft_gate_policy.py new file mode 100644 index 0000000..bbd67a9 --- /dev/null +++ b/src/rag/qa_v1_1_73/soft_gate_policy.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import os +import re +from collections.abc import Iterable, Mapping +from typing import Any + + +LEGACY = "legacy" +SOFT_COMPAT_V1 = "soft_compat_v1" +_POLICIES = {LEGACY, SOFT_COMPAT_V1} + +_ANSWER_SLOT_TYPES = { + "biochemical_activity", + "chromatin_modification", + "enzyme_activity", + "expression_level", + "expression_pattern", + "expression_relationship", + "function", + "gene_expression", + "regulation", + "regulatory_relationship", + "transporter_activity", +} + +_SYMMETRIC_COMPATIBILITY = ( + {"pathogen", "organism_group"}, + {"compartment", "organelle"}, + {"compartment", "cellular_component"}, + {"gene_group", "biosynthetic_gene_set"}, + {"enzyme", "enzyme_activity"}, +) + +_DOI_INTERNAL_RE = re.compile( + r"^(10\.\d{4,9})_([A-Za-z0-9][A-Za-z0-9.;()/:+\-]*)$", + re.IGNORECASE, +) +_OXFORD_DOI_INTERNAL_RE = re.compile( + r"^(10\.1093)_([A-Za-z0-9][A-Za-z0-9.;()+\-]*)_([A-Za-z0-9][A-Za-z0-9.;()+\-]*)$", + re.IGNORECASE, +) +_DOI_RE = re.compile( + r"^10\.\d{4,9}/[A-Za-z0-9][A-Za-z0-9._;()/:+\-]*$", + re.IGNORECASE, +) + + +def _type_token(value: Any) -> str: + text = re.sub(r"[^a-z0-9]+", "_", str(value or "").strip().lower()) + return text.strip("_") + + +def gate_policy(env: Mapping[str, str] | None = None) -> str: + values = os.environ if env is None else env + value = str(values.get("QA_AGENTIC_GATE_POLICY", LEGACY) or LEGACY).strip().lower() + return value if value in _POLICIES else LEGACY + + +def type_compatibility( + actual_type: Any, + requested_types: Iterable[Any], + *, + relationship_context: bool = False, +) -> tuple[bool, str]: + actual = _type_token(actual_type) + requested = {_type_token(value) for value in requested_types} + requested.discard("") + requested.difference_update({"any", "unknown", "unspecified"}) + + if not requested: + return True, "no_terminal_type_constraint" + if actual in requested: + return True, "exact_type_match" + if requested.intersection(_ANSWER_SLOT_TYPES): + return True, "answer_slot_not_endpoint_type" + for family in _SYMMETRIC_COMPATIBILITY: + if actual in family and family.intersection(requested): + return True, "controlled_type_family" + if actual == "transcription_factor" and requested.intersection({"gene", "protein"}): + return True, "transcription_factor_endpoint" + if ( + relationship_context + and actual == "genetic_variant" + and requested.intersection({"gene", "condition"}) + ): + return True, "genetic_variant_relationship_endpoint" + return False, "terminal_type_mismatch" + + +def doi_from_paper_id(paper_id: Any) -> str: + value = str(paper_id or "").strip() + oxford_match = _OXFORD_DOI_INTERNAL_RE.fullmatch(value) + if oxford_match: + normalized = ( + f"{oxford_match.group(1)}/{oxford_match.group(2)}/{oxford_match.group(3)}" + ) + return normalized if _DOI_RE.fullmatch(normalized) else "" + match = _DOI_INTERNAL_RE.fullmatch(value) + if not match: + return "" + normalized = f"{match.group(1)}/{match.group(2)}" + return normalized if _DOI_RE.fullmatch(normalized) else "" + + +def pmcid_from_paper_id(paper_id: Any) -> str: + value = str(paper_id or "").strip() + match = re.fullmatch(r"(PMC\d+)(?:_[A-Za-z0-9][A-Za-z0-9.\-]*)?", value, re.IGNORECASE) + return match.group(1).upper() if match else "" diff --git a/tests/test_qa_v1_1_73_soft_gate_policy.py b/tests/test_qa_v1_1_73_soft_gate_policy.py new file mode 100644 index 0000000..eea8fe4 --- /dev/null +++ b/tests/test_qa_v1_1_73_soft_gate_policy.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from src.rag.qa_v1_1_73 import soft_gate_policy as module + + +def test_gate_policy_defaults_and_fails_closed_to_legacy(): + assert module.gate_policy({}) == "legacy" + assert ( + module.gate_policy({"QA_AGENTIC_GATE_POLICY": "soft_compat_v1"}) + == "soft_compat_v1" + ) + assert module.gate_policy({"QA_AGENTIC_GATE_POLICY": "unexpected"}) == "legacy" + + +def test_closed_type_compatibility_map_and_answer_slots(): + compatible = [ + ("pathogen", ["organism_group"], False), + ("organism_group", ["pathogen"], False), + ("compartment", ["organelle"], False), + ("compartment", ["cellular_component"], False), + ("organelle", ["compartment"], False), + ("transcription_factor", ["gene"], False), + ("transcription_factor", ["protein"], False), + ("gene_group", ["biosynthetic_gene_set"], False), + ("biosynthetic_gene_set", ["gene_group"], False), + ("enzyme", ["enzyme_activity"], False), + ("enzyme_activity", ["enzyme"], False), + ("genetic_variant", ["gene", "condition"], True), + ("gene", ["expression_pattern"], False), + ("tissue", ["regulatory_relationship"], False), + ] + + for actual, requested, relationship_context in compatible: + accepted, reason = module.type_compatibility( + actual, + requested, + relationship_context=relationship_context, + ) + assert accepted, (actual, requested, reason) + + assert not module.type_compatibility("pathogen", ["phenotype"])[0] + assert not module.type_compatibility("genetic_variant", ["gene"])[0] + assert not module.type_compatibility("unknown_new_type", ["gene"])[0] + + +def test_doi_from_paper_id_accepts_only_unambiguous_internal_form(): + + assert ( + module.doi_from_paper_id("10.3389_fpls.2016.00954") + == "10.3389/fpls.2016.00954" + ) + assert ( + module.doi_from_paper_id("10.1016_j.cj.2021.04.004") + == "10.1016/j.cj.2021.04.004" + ) + assert module.doi_from_paper_id("10.1093_jxb_ery350") == "10.1093/jxb/ery350" + assert ( + module.doi_from_paper_id("10.1093_plphys_kiac335") + == "10.1093/plphys/kiac335" + ) + assert module.doi_from_paper_id("PMC123") == "" + assert module.doi_from_paper_id("12345678") == "" + assert module.doi_from_paper_id("10.3389_fpls_2016_00954") == "" + assert module.doi_from_paper_id("10.3389/") == "" + assert module.doi_from_paper_id("not_a_doi") == "" + + +def test_pmcid_from_paper_id_accepts_only_leading_pmc_accession(): + + assert module.pmcid_from_paper_id("PMC10299364_ijms-24-10165") == "PMC10299364" + assert module.pmcid_from_paper_id("pmc8508941") == "PMC8508941" + assert module.pmcid_from_paper_id("prefix_PMC10299364") == "" + assert module.pmcid_from_paper_id("PMC_10299364") == "" diff --git a/tests/test_qa_v1_1_73_soft_type_gate.py b/tests/test_qa_v1_1_73_soft_type_gate.py new file mode 100644 index 0000000..c684d58 --- /dev/null +++ b/tests/test_qa_v1_1_73_soft_type_gate.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import pytest + +from src.rag.qa_v1_1_73.agentic_controller import AgenticRetrievalController +from src.rag.qa_v1_1_73.schemas import EvidenceItem + + +def _controller(): + return AgenticRetrievalController.__new__(AgenticRetrievalController) + + +def _item( + actual_type: str, + *, + source: str = "ANCHOR", + relation: str = "regulates", + target: str = "ANSWER", + evidence: str = "ANCHOR regulates ANSWER under the reported condition.", + paper_id: str = "paper-1", +) -> EvidenceItem: + return EvidenceItem( + evidence_id="Q0.A1.E1", + kb="V16/V15", + source=source, + relation=relation, + target=target, + evidence=evidence, + paper_id=paper_id, + raw={ + "answer_entity_type": actual_type, + "source_type": "gene", + "target_type": actual_type, + "anchor_name": "ANCHOR", + }, + ) + + +def _group(requested_types: list[str], *, question: str = "How does ANCHOR regulate ANSWER?"): + return { + "subquestion_id": "Q0", + "question": question, + "role": "required", + "anchors": [{"mention": "ANCHOR", "canonical_name": "ANCHOR"}], + "target_entity_types": requested_types, + "acceptance_requirements": { + "requires_structured_assertion": True, + "requires_paper_provenance": True, + "requires_anchor_connection": True, + "requires_terminal_type_match": True, + }, + } + + +@pytest.mark.parametrize( + ("actual_type", "requested_types"), + [ + ("pathogen", ["organism_group"]), + ("compartment", ["organelle", "cellular_component"]), + ("transcription_factor", ["gene", "regulatory_relationship"]), + ("gene_group", ["biosynthetic_gene_set"]), + ("enzyme", ["enzyme_activity"]), + ("genetic_variant", ["gene", "condition", "phenotype"]), + ("tissue", ["gene", "expression_pattern"]), + ("gene", ["process", "pathway", "function"]), + ("genomic_region", ["gene", "chromatin_modification"]), + ], +) +def test_group_type_gate_is_legacy_exact_and_soft_compatible( + monkeypatch, actual_type, requested_types +): + controller = _controller() + item = _item(actual_type) + group = _group(requested_types) + + monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "legacy") + assert not controller._group_evidence_satisfies(group, item) + + monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "soft_compat_v1") + assert controller._group_evidence_satisfies(group, item) + + +def test_plan_evidence_and_chain_type_gates_use_same_policy(monkeypatch): + controller = _controller() + item = _item("transcription_factor") + plan = { + "anchors": [{"mention": "ANCHOR"}], + "graph_plan": {"required_terminal_types": ["gene"]}, + "acceptance_gate": {}, + } + chain = { + "chain_id": "Q0.A1.G1", + "start": "ANCHOR", + "end": "ANSWER", + "anchor_name": "ANCHOR", + "answer_entity_type": "transcription_factor", + "hops": [ + { + "source": "ANCHOR", + "relation": "regulates", + "target": "ANSWER", + "paper_id": "paper-1", + } + ], + } + + monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "legacy") + assert not controller._evidence_satisfies(plan, item) + assert not controller._chain_satisfies(plan, chain) + + monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "soft_compat_v1") + assert controller._evidence_satisfies(plan, item) + assert controller._chain_satisfies(plan, chain) + + +@pytest.mark.parametrize( + ("group", "item"), + [ + (_group(["gene"]), _item("gene", paper_id="")), + ( + _group(["gene"], question="Which genes are associated with ANCHOR?"), + _item( + "gene", + source="OTHER", + evidence="OTHER regulates ANSWER under the reported condition.", + ), + ), + ( + _group(["gene"], question="How does ANCHOR respond to heat stress?"), + _item( + "gene", + evidence="ANCHOR responds only to cold stress and freezing.", + ), + ), + ( + _group(["gene"], question="Which genes does ANCHOR regulate?"), + _item("gene", relation="compared_to"), + ), + ], +) +def test_non_type_safety_gates_remain_active_in_both_policies(monkeypatch, group, item): + controller = _controller() + for policy in ("legacy", "soft_compat_v1"): + monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", policy) + assert not controller._group_evidence_satisfies(group, item)