feat: add evidence verification agent for scholarly claims

citations.py

@@ -153,6 +153,23 @@ def score(self) -> float | None:
         return round(len(self.citations) / self.attempted, 4)
 
 
+class ClaimVerification(BaseModel):
+    """Verification result for a single extracted claim."""
+
+    claim: str
+    matched_citations: list[Citation] = []
+    supported: bool = False
+    strength: Literal["strong", "moderate", "weak", "unsupported"] = "unsupported"
+    reasons: list[str] = []
+
+
+class EvidenceVerificationReport(BaseModel):
+    """Claims checked and the citations that supported them."""
+
+    claims: list[ClaimVerification] = []
+    overall_score: float | None = None
+
+
 def _coerce_int(value: Any) -> int | None:
     """Accept 2 and "2" alike; reject everything else without raising."""
     if isinstance(value, bool):
@@ -369,6 +386,73 @@ def extract_citations(text: str | None) -> tuple[str, CitationExtraction]:
     return text, CitationExtraction()
 
 
+_CLAIM_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")
+
+
+def _extract_claims(prose: str) -> list[str]:
+    """Split prose into candidate claims, keeping sentences that cite anything."""
+    claims = []
+    for sentence in _CLAIM_SENTENCE_RE.split(prose):
+        sentence = sentence.strip()
+        if not sentence:
+            continue
+        if re.search(r"(Qur'?an|hadith|surah|Surah|reported|narrated|said|wrote|states?)\b", sentence):
+            claims.append(sentence)
+    return claims
+
+
+def _match_citations(claim: str, citations: list[Citation]) -> list[Citation]:
+    """Link a claim to the citations whose references appear in its text."""
+    matched = []
+    for citation in citations:
+        ref = citation.reference if isinstance(citation, QuranCitation) else None
+        if ref and ref in claim:
+            matched.append(citation)
+            continue
+        if isinstance(citation, HadithCitation) and citation.collection and citation.collection in claim:
+            matched.append(citation)
+            continue
+        if isinstance(citation, ScholarlyReference) and citation.work and citation.work in claim:
+            matched.append(citation)
+            continue
+    return matched
+
+
+def verify_evidence(text: str | None) -> EvidenceVerificationReport:
+    """Verify claims in *text* against the citations it carries."""
+    prose, extraction = extract_citations(text)
+    report = EvidenceVerificationReport()
+    for claim in _extract_claims(prose):
+        matched = _match_citations(claim, extraction.citations)
+        if not matched:
+            report.claims.append(
+                ClaimVerification(
+                    claim=claim,
+                    supported=False,
+                    reasons=["no structured citation found for this claim"],
+                )
+            )
+            continue
+        weak = any(
+            isinstance(c, HadithCitation) and c.grading and "sahih" not in c.grading
+            for c in matched
+        )
+        report.claims.append(
+            ClaimVerification(
+                claim=claim,
+                matched_citations=matched,
+                supported=True,
+                strength="moderate" if weak else "strong",
+                reasons=[f"matched {len(matched)} structured citation(s)"],
+            )
+        )
+    if report.claims:
+        report.overall_score = round(
+            sum(1 for c in report.claims if c.supported) / len(report.claims), 4
+        )
+    return report
+
+
 class CitationStreamFilter:
     """Keep the citation block out of SSE deltas.
 

evidence_agent.py

@@ -0,0 +1,4 @@
+from verifier import extract_ana_verify_all
+
+def run_evidence_agent(text: str) -> dict:
+    return {'results': extract_and_verify_all(text)}

hadith.py

@@ -503,6 +503,87 @@ def build_caution_note(text: str, references: list[HadithReference]) -> str | No
     return "\n".join(lines)
 
 
+# ---------------------------------------------------------------------------
+# Evidence verification agent
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class EvidenceVerdict:
+    claim: str
+    references: list[HadithReference]
+    supported: bool
+    strength: Strength
+    notes: list[str] = field(default_factory=list)
+
+
+class EvidenceVerificationAgent:
+    """Validates scholarly claims by tracing cited hadith evidence to its
+    bundled primary-source grading record.
+    """
+
+    def __init__(self, source: GradingSource | None = None):
+        self._source = source or get_default_source()
+
+    def extract_claims(self, text: str) -> list[str]:
+        """Extract candidate scholarly claims from generated response text."""
+        claims = []
+        for sentence in re.split(r"(?<=[.!?])\s+", text.strip()):
+            sentence = sentence.strip()
+            if sentence and re.search(
+                r"\b(narrated|reported|said|states|according to)\b",
+                sentence,
+                re.IGNORECASE,
+            ):
+                claims.append(sentence)
+        return claims
+
+    def verify_claims(self, text: str) -> list[EvidenceVerdict]:
+        """Verify each extracted claim, falling back to the full response."""
+        claims = self.extract_claims(text) or [text.strip()]
+        return [self._verify(claim, text) for claim in claims]
+
+    def _verify(self, claim: str, full_text: str) -> EvidenceVerdict:
+        refs = annotate(claim, self._source) or annotate(full_text, self._source)
+        notes: list[str] = []
+        if not refs:
+            return EvidenceVerdict(claim, [], False, Strength.UNKNOWN, ["no primary-source evidence cited"])
+        if not any(ref.raw.lower() in claim.lower() for ref in refs):
+            notes.append("cited evidence is not explicitly connected to this claim")
+        verified = [ref for ref in refs if ref.verified]
+        if not verified:
+            notes.append("cited references could not be verified against primary sources")
+            return EvidenceVerdict(claim, refs, False, Strength.UNKNOWN, notes)
+        strengths = [Strength(ref.grade) for ref in verified]
+        strength = aggregate_strength(strengths)
+        if strength in (Strength.DAIF, Strength.MAWDU):
+            notes.append("cited evidence is weak or fabricated; do not rely on it")
+        if any(ref.flagged for ref in verified):
+            notes.append("cited evidence carries an unstated authenticity caveat")
+        supported = (
+            bool(refs)
+            and strength not in (Strength.DAIF, Strength.MAWDU, Strength.UNKNOWN)
+            and not any(ref.flagged for ref in verified)
+            and not any("not explicitly connected" in note for note in notes)
+        )
+        return EvidenceVerdict(claim, refs, supported, strength, notes)
+
+    def trace_provenance(self, text: str) -> list[dict[str, str]]:
+        """Return a verification audit trail for every citation in *text*."""
+        return [
+            {
+                "collection": ref.collection or "",
+                "hadith_number": str(ref.hadith_number or ""),
+                "grade": ref.grade,
+                "grader": ref.grader or "",
+                "chain_type": ref.chain_type or "",
+                "verified": str(ref.verified).lower(),
+                "flagged": str(ref.flagged).lower(),
+            }
+            for ref in annotate(text, self._source)
+        ]
+
+
 HADITH_ADAB_CONTEXT = """
 
 HADITH CITATION RULES:

main.py

@@ -148,6 +148,8 @@
 
 genai.configure(api_key=GEMINI_API_KEY)
 
+EVIDENCE_VERIFICATION_ENABLED = os.getenv("EVIDENCE_VERIFICATION_ENABLED", "true").lower() not in {"0", "false", "off"}
+
 app = FastAPI(title="DeenBridge AI API")
 
 # --- Service API-key authentication ---
@@ -344,6 +346,26 @@ class CitationVerificationResult(BaseModel):
     reason: str | None = None
 
 
+class EvidenceClaim(BaseModel):
+    claim: str
+    evidence: list[str] = Field(default_factory=list)
+    source_type: str | None = None
+    verification_status: str = "unverified"
+    confidence: float | None = None
+    reason: str | None = None
+    alternative_evidence: list[str] = Field(default_factory=list)
+
+
+class EvidenceVerificationResult(BaseModel):
+    claims: list[EvidenceClaim] = Field(default_factory=list)
+    overall_score: float = 0.0
+    audit_trail: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class EvidenceVerifyRequest(BaseModel):
+    text: str = Field(..., min_length=1, max_length=30000)
+
+
 class ChatRequest(BaseModel):
     prompt: str = Field(..., max_length=CHAT_PROMPT_MAX_LENGTH)
     chat_id: str | None = None
@@ -387,6 +409,8 @@ class ChatResponse(BaseModel):
     # Structured references parsed out of the answer (#15). Empty when the
     # answer cited nothing, or when nothing it cited could be validated.
     citations: list[Citation] = []
+    # Evidence verification agent audit (#evidence-verification).
+    evidence_verification: EvidenceVerificationResult | None = None
 
 
 class FeedbackRequest(BaseModel):
@@ -781,6 +805,87 @@ async def run_strict_corrective_loop(
     return safe_text or original_text
 
 
+def _citation_to_dict(item: Any) -> dict[str, Any]:
+    if hasattr(item, "model_dump"):
+        return item.model_dump()
+    if hasattr(item, "dict"):
+        return item.dict()
+    return {}
+
+
+def _evidence_label(d: dict[str, Any]) -> str:
+    if d.get("source") == "quran" and d.get("surah") is not None:
+        return f"Quran {d['surah']}:{d.get('ayah', '')}"
+    if d.get("collection"):
+        return f"{d['collection']} {d.get('number', '')}".strip()
+    return str(d.get("source") or "cited source")
+
+
+def verify_evidence_claims(
+    response_text: str,
+    citation_extraction: CitationExtraction,
+    hadith_refs: list[HadithReference],
+) -> EvidenceVerificationResult:
+    """Deterministic claim-evidence audit using already-verified citations."""
+    result = EvidenceVerificationResult()
+    claims = [
+        s.strip()
+        for s in response_text.replace("\n", " ").split(".")
+        if len(s.strip()) >= 25
+    ]
+    evidence: list[tuple[str, str, bool]] = []
+
+    for c in citation_extraction.citations:
+        d = _citation_to_dict(c)
+        label = _evidence_label(d)
+        status = str(d.get("status") or d.get("verification") or "unverified")
+        evidence.append((label, status, status in {"verified", "authentic", "sahih", "hasan", "mutawatir"}))
+
+    for h in hadith_refs:
+        d = _citation_to_dict(h)
+        label = _evidence_label(d)
+        grade = str(d.get("grade") or d.get("authenticity") or d.get("status") or "unverified").lower()
+        evidence.append((label, grade, grade not in {"daif", "weak", "munkar", "fabricated", "mawdu", "unverified"}))
+
+    for claim in claims:
+        matched = [e for e in evidence if e[0].lower() in claim.lower()]
+        if not matched and len(claims) == 1 and len(evidence) == 1:
+            matched = evidence[:]
+
+        if not matched:
+            result.claims.append(
+                EvidenceClaim(
+                    claim=claim,
+                    verification_status="unsupported",
+                    confidence=0.1,
+                    reason="No verifiable citation is present for this claim.",
+                    alternative_evidence=["Provide a precise Quranic/Hadith citation for this claim."],
+                )
+            )
+            result.audit_trail.append({"claim": claim, "status": "unsupported"})
+            continue
+
+        supported = any(e[2] for e in matched)
+        status = "supported" if supported else "weak"
+        result.claims.append(
+            EvidenceClaim(
+                claim=claim,
+                evidence=[e[0] for e in matched],
+                verification_status=status,
+                confidence=0.9 if supported else 0.4,
+                reason="Evidence linked and verified." if supported else "Evidence cited but could not be verified.",
+            )
+        )
+        result.audit_trail.append({"claim": claim, "evidence": [e[0] for e in matched], "status": status})
+
+    result.overall_score = (
+        sum(1 for c in result.claims if c.verification_status == "supported") / len(result.claims)
+        if result.claims
+        else 0.0
+    )
+    return result
+
+
 @app.post("/chat", response_model=ChatResponse)
 @limiter.limit(f"{CHAT_RATE_LIMIT_MAX}/{CHAT_RATE_LIMIT_WINDOW_SECONDS} seconds")
 async def chat(body: ChatRequest, request: Request, fastapi_response: Response) -> ChatResponse:
@@ -1080,6 +1185,13 @@ async def generate(safety_prompt: str) -> str:
         # Baked into response_text *before* the cache write so a cached hit
         # replays the same caution the user originally saw.
         hadith_refs = annotate_hadith(response_text)
+        # --- Evidence verification agent ---
+        evidence_verification = None
+        if EVIDENCE_VERIFICATION_ENABLED:
+            try:
+                evidence_verification = verify_evidence_claims(response_text, citation_extraction, hadith_refs)
+            except Exception as exc:  # noqa: BLE001 - verification is best-effort
+                logger.warning("Evidence verification failed; continuing without it: %s", exc)
         caution = build_caution_note(response_text, hadith_refs)
         if caution:
             response_text = f"{response_text.rstrip()}\n\n{caution}"
@@ -1202,6 +1314,7 @@ async def generate(safety_prompt: str) -> str:
             purchases=purchase_info,
             language=effective_language,
             citations=citation_extraction.citations,
+            evidence_verification=evidence_verification,
         )
         _finalize()
         _succeeded = True
@@ -1526,6 +1639,12 @@ async def event_generator() -> AsyncGenerator[str, None]:
 
                 # --- Hadith authenticity grading ---
                 hadith_refs = annotate_hadith(combined_text)
+                evidence_verification = None
+                if EVIDENCE_VERIFICATION_ENABLED:
+                    try:
+                        evidence_verification = verify_evidence_claims(combined_text, citation_extraction, hadith_refs)
+                    except Exception as exc:  # noqa: BLE001 - verification is best-effort
+                        logger.warning("Streaming evidence verification failed: %s", exc)
                 caution = build_caution_note(combined_text, hadith_refs)
                 if caution:
                     combined_text = f"{combined_text.rstrip()}\n\n{caution}"
@@ -1604,6 +1723,7 @@ async def event_generator() -> AsyncGenerator[str, None]:
                         "tafsir": tafsir_info.model_dump() if tafsir_info else None,
                         "zakat": zakat_info.model_dump() if zakat_info else None,
                         "citations": [c.model_dump() for c in citation_extraction.citations],
+                        "evidence_verification": evidence_verification.model_dump() if evidence_verification else None,
                     },
                     ensure_ascii=False,
                 )
@@ -1787,6 +1907,26 @@ async def delete_chat(chat_id: str, user_id: str | None = None) -> dict[str, str
         ) from e
 
 
+@app.post("/evidence/verify", response_model=EvidenceVerificationResult)
+async def verify_evidence_endpoint(body: EvidenceVerifyRequest) -> EvidenceVerificationResult:
+    """Run the evidence verification agent on a generated answer.
+
+    Extracts any citation block, parses Quran/Hadith references, links them to
+    claims, and returns a structured audit of support strength.
+    """
+    try:
+        cleaned_text, citation_extraction = extract_citations(body.text)
+        hadith_refs = annotate_hadith(cleaned_text)
+        return verify_evidence_claims(cleaned_text, citation_extraction, hadith_refs)
+    except Exception as exc:
+        logger.error("Evidence verification endpoint failed: %s", exc)
+        raise APIException(
+            status_code=500,
+            detail="Evidence verification failed.",
+            hint="The evidence verification agent could not process the supplied text. Verify the text is well-formed and retry.",
+        ) from exc
+
+
 # ---------------------------------------------------------------------------
 # Feedback: capture and admin views
 # ---------------------------------------------------------------------------