"""Hard ambiguity and conflict rules for routing decisions. These rules check structural markers in extraction output that indicate the document *requires* semantic reasoning by the 9B adjudicator. Any triggered rule forces ADJUDICATION regardless of confidence scores. """ from __future__ import annotations from typing import Any from services.intelligence_pipeline_v3.routing.reasons import RoutingReason def evaluate_hard_rules( confidence_features: dict[str, Any], ambiguity_markers: dict[str, Any], ) -> list[RoutingReason]: """Evaluate hard ambiguity/conflict rules against extraction output. Parameters ---------- confidence_features: Field-level confidence features from the confidence pipeline. Expected keys include: - ``evidence_coverage``: float 0-1 - ``material_fields_present``: bool - ``cross_chunk_relations``: bool (relations span multiple chunks) ambiguity_markers: Structural ambiguity markers from candidate generation and resolution. Expected keys include: - ``unresolved_aliases``: int (count of unresolved entity aliases) - ``primary_company_count``: int (number of primary companies detected) - ``contradictory_numeric_facts``: bool - ``conflicting_sentiment``: bool - ``implied_causal_impact``: bool - ``guidance_vs_consensus``: bool - ``long_document_cross_chunk``: bool Returns ------- list[RoutingReason] List of triggered reasons. Empty list means no hard rules triggered. """ triggered: list[RoutingReason] = [] # Unresolved entity aliases require contextual disambiguation if ambiguity_markers.get("unresolved_aliases", 0) > 0: triggered.append(RoutingReason.UNRESOLVED_ALIAS) # Multiple primary companies need reasoning about which is the subject if ambiguity_markers.get("primary_company_count", 0) > 1: triggered.append(RoutingReason.MULTIPLE_PRIMARY_COMPANIES) # Contradictory numeric facts (e.g., conflicting revenue figures) if ambiguity_markers.get("contradictory_numeric_facts", False): triggered.append(RoutingReason.CONTRADICTORY_NUMERIC_FACTS) # Conflicting sentiment across evidence groups for the same company if ambiguity_markers.get("conflicting_sentiment", False): triggered.append(RoutingReason.CONFLICTING_SENTIMENT) # Implied causal impact requiring reasoning (not explicit statement) if ambiguity_markers.get("implied_causal_impact", False): triggered.append(RoutingReason.IMPLIED_CAUSAL_IMPACT) # Guidance vs consensus comparison requires model reasoning if ambiguity_markers.get("guidance_vs_consensus", False): triggered.append(RoutingReason.GUIDANCE_VS_CONSENSUS_REQUIRES_REASONING) # Material fields missing from extraction output if not confidence_features.get("material_fields_present", True): triggered.append(RoutingReason.MATERIAL_FIELD_MISSING) # Cross-chunk relations in long documents need broader context if ambiguity_markers.get("long_document_cross_chunk", False): triggered.append(RoutingReason.LONG_DOCUMENT_CROSS_CHUNK_RELATION) return triggered