Celes Renata
a72f336ad1
Multi-stage evidence-grounded inference architecture replacing the monolithic 9B model extraction pipeline. CPU-first specialist services handle routine extraction while the 9B vLLM model is preserved for semantic adjudication of ambiguous cases. Key components: - Capability-aware inference gateway (OpenAI-compatible + Ollama) - Endpoint registry with DB migrations and REST API - Sentence-aware document segmenter (property tests) - Deterministic financial parsing with offset integrity - Symbol resolution with ambiguity detection - Specialist service (GLiNER2, dynamic batching, K8s deployment) - Company-specific sentiment (FinBERT, calibration) - Retrieval-based novelty and duplicate detection - Confidence calibration pipeline - Deterministic routing engine (property tests) - 9B adjudication layer with VRAM gating - Stock-specific impact model (features, labels, baseline, trained) - Pipeline orchestrator (state machine, queues, leases, feature flags) - Bounded parallelism (async workers, semaphore, load shedding) - Observability (tracing, metrics, alerts) - Compatibility adapter (v3→v2 golden mapping tests) - Shadow/canary promotion framework - Active learning and fine-tuning pipeline Test results: 1,161 tests pass, ruff lint clean. All 282 spec tasks completed.
617 lines
23 KiB
Python
617 lines
23 KiB
Python
"""Tests for the deterministic routing engine.
|
|
|
|
Covers:
|
|
- Hard rules trigger adjudication
|
|
- Confidence below threshold triggers adjudication
|
|
- Confidence above threshold triggers fast path
|
|
- All reasons are assigned correctly
|
|
- Property test: same inputs always produce same route (determinism)
|
|
- Property test: confidence at exact threshold boundary has deterministic behavior
|
|
- Decision storage captures features
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from hypothesis import given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.intelligence_pipeline_v3.routing.reasons import (
|
|
RouteDecision,
|
|
RoutingReason,
|
|
)
|
|
from services.intelligence_pipeline_v3.routing.router import (
|
|
RoutingEngine,
|
|
)
|
|
from services.intelligence_pipeline_v3.routing.rules import evaluate_hard_rules
|
|
from services.intelligence_pipeline_v3.routing.store import RoutingDecisionStore
|
|
from services.intelligence_pipeline_v3.routing.thresholds import (
|
|
DEFAULT_DOCUMENT_THRESHOLDS,
|
|
DEFAULT_EVENT_THRESHOLDS,
|
|
DEFAULT_FALLBACK_THRESHOLD,
|
|
FastPathThresholds,
|
|
evaluate_thresholds,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def engine() -> RoutingEngine:
|
|
return RoutingEngine()
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_features() -> dict:
|
|
"""Confidence features with no issues — should pass fast path."""
|
|
return {
|
|
"calibrated_confidence": 0.90,
|
|
"evidence_coverage": 0.95,
|
|
"material_fields_present": True,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_markers() -> dict:
|
|
"""Ambiguity markers with no issues."""
|
|
return {
|
|
"unresolved_aliases": 0,
|
|
"primary_company_count": 1,
|
|
"contradictory_numeric_facts": False,
|
|
"conflicting_sentiment": False,
|
|
"implied_causal_impact": False,
|
|
"guidance_vs_consensus": False,
|
|
"long_document_cross_chunk": False,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 31.1 — Routing reason enums
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRoutingReasonEnums:
|
|
"""Test that all required routing reason enums exist and are correct."""
|
|
|
|
def test_all_reasons_defined(self):
|
|
expected = {
|
|
"UNRESOLVED_ALIAS",
|
|
"MULTIPLE_PRIMARY_COMPANIES",
|
|
"CONTRADICTORY_NUMERIC_FACTS",
|
|
"CONFLICTING_SENTIMENT",
|
|
"IMPLIED_CAUSAL_IMPACT",
|
|
"GUIDANCE_VS_CONSENSUS_REQUIRES_REASONING",
|
|
"MATERIAL_FIELD_MISSING",
|
|
"EVIDENCE_COVERAGE_BELOW_THRESHOLD",
|
|
"CALIBRATED_CONFIDENCE_BELOW_THRESHOLD",
|
|
"LONG_DOCUMENT_CROSS_CHUNK_RELATION",
|
|
"FAST_PATH_ACCEPTED",
|
|
}
|
|
actual = {r.name for r in RoutingReason}
|
|
assert actual == expected
|
|
|
|
def test_route_decision_values(self):
|
|
assert RouteDecision.FAST_PATH.value == "fast_path"
|
|
assert RouteDecision.ADJUDICATION.value == "adjudication"
|
|
|
|
def test_reason_string_values_match_names(self):
|
|
"""Reason values should be their name for database storage."""
|
|
for reason in RoutingReason:
|
|
assert reason.value == reason.name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 31.2 — Hard ambiguity/conflict rules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHardRules:
|
|
"""Test that hard rules correctly trigger adjudication reasons."""
|
|
|
|
def test_no_triggers_returns_empty(self, clean_features, clean_markers):
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert result == []
|
|
|
|
def test_unresolved_alias_triggers(self, clean_features, clean_markers):
|
|
clean_markers["unresolved_aliases"] = 2
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.UNRESOLVED_ALIAS in result
|
|
|
|
def test_multiple_primary_companies_triggers(self, clean_features, clean_markers):
|
|
clean_markers["primary_company_count"] = 3
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.MULTIPLE_PRIMARY_COMPANIES in result
|
|
|
|
def test_contradictory_numeric_facts_triggers(self, clean_features, clean_markers):
|
|
clean_markers["contradictory_numeric_facts"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.CONTRADICTORY_NUMERIC_FACTS in result
|
|
|
|
def test_conflicting_sentiment_triggers(self, clean_features, clean_markers):
|
|
clean_markers["conflicting_sentiment"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.CONFLICTING_SENTIMENT in result
|
|
|
|
def test_implied_causal_impact_triggers(self, clean_features, clean_markers):
|
|
clean_markers["implied_causal_impact"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.IMPLIED_CAUSAL_IMPACT in result
|
|
|
|
def test_guidance_vs_consensus_triggers(self, clean_features, clean_markers):
|
|
clean_markers["guidance_vs_consensus"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.GUIDANCE_VS_CONSENSUS_REQUIRES_REASONING in result
|
|
|
|
def test_material_field_missing_triggers(self, clean_features, clean_markers):
|
|
clean_features["material_fields_present"] = False
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.MATERIAL_FIELD_MISSING in result
|
|
|
|
def test_long_document_cross_chunk_triggers(self, clean_features, clean_markers):
|
|
clean_markers["long_document_cross_chunk"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert RoutingReason.LONG_DOCUMENT_CROSS_CHUNK_RELATION in result
|
|
|
|
def test_multiple_triggers_accumulate(self, clean_features, clean_markers):
|
|
clean_markers["unresolved_aliases"] = 1
|
|
clean_markers["conflicting_sentiment"] = True
|
|
clean_markers["implied_causal_impact"] = True
|
|
result = evaluate_hard_rules(clean_features, clean_markers)
|
|
assert len(result) == 3
|
|
assert RoutingReason.UNRESOLVED_ALIAS in result
|
|
assert RoutingReason.CONFLICTING_SENTIMENT in result
|
|
assert RoutingReason.IMPLIED_CAUSAL_IMPACT in result
|
|
|
|
def test_hard_rules_override_high_confidence(self, clean_markers):
|
|
"""Even with perfect confidence, hard rules force adjudication."""
|
|
features = {
|
|
"calibrated_confidence": 1.0,
|
|
"evidence_coverage": 1.0,
|
|
"material_fields_present": True,
|
|
}
|
|
clean_markers["contradictory_numeric_facts"] = True
|
|
result = evaluate_hard_rules(features, clean_markers)
|
|
assert RoutingReason.CONTRADICTORY_NUMERIC_FACTS in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 31.3 — Calibrated fast-path thresholds
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestThresholds:
|
|
"""Test threshold evaluation by document and event type."""
|
|
|
|
def test_default_article_threshold(self):
|
|
assert DEFAULT_DOCUMENT_THRESHOLDS["article"] == 0.80
|
|
|
|
def test_default_filing_threshold(self):
|
|
assert DEFAULT_DOCUMENT_THRESHOLDS["filing"] == 0.70
|
|
|
|
def test_default_transcript_threshold(self):
|
|
assert DEFAULT_DOCUMENT_THRESHOLDS["transcript"] == 0.75
|
|
|
|
def test_confidence_above_threshold_is_fast_path(self):
|
|
thresholds = FastPathThresholds()
|
|
result = evaluate_thresholds(0.85, "article", None, thresholds)
|
|
assert result == RouteDecision.FAST_PATH
|
|
|
|
def test_confidence_below_threshold_is_adjudication(self):
|
|
thresholds = FastPathThresholds()
|
|
result = evaluate_thresholds(0.75, "article", None, thresholds)
|
|
assert result == RouteDecision.ADJUDICATION
|
|
|
|
def test_confidence_at_exact_threshold_is_fast_path(self):
|
|
"""Boundary: confidence == threshold passes fast path."""
|
|
thresholds = FastPathThresholds()
|
|
result = evaluate_thresholds(0.80, "article", None, thresholds)
|
|
assert result == RouteDecision.FAST_PATH
|
|
|
|
def test_event_type_overrides_document_type(self):
|
|
thresholds = FastPathThresholds()
|
|
# guidance_change has threshold 0.65, article has 0.80
|
|
# With event_type, the event threshold should apply
|
|
result = evaluate_thresholds(0.70, "article", "guidance_change", thresholds)
|
|
assert result == RouteDecision.FAST_PATH
|
|
|
|
def test_unknown_document_type_uses_fallback(self):
|
|
thresholds = FastPathThresholds()
|
|
result = evaluate_thresholds(0.79, "unknown_type", None, thresholds)
|
|
assert result == RouteDecision.ADJUDICATION # fallback is 0.80
|
|
|
|
def test_unknown_event_type_falls_through_to_document(self):
|
|
thresholds = FastPathThresholds()
|
|
# Unknown event, known document type
|
|
result = evaluate_thresholds(0.72, "filing", "unknown_event", thresholds)
|
|
assert result == RouteDecision.FAST_PATH # filing threshold is 0.70
|
|
|
|
def test_custom_thresholds(self):
|
|
thresholds = FastPathThresholds(
|
|
document_thresholds={"custom_doc": 0.50},
|
|
event_thresholds={"custom_event": 0.30},
|
|
fallback_threshold=0.90,
|
|
)
|
|
assert evaluate_thresholds(0.50, "custom_doc", None, thresholds) == RouteDecision.FAST_PATH
|
|
assert evaluate_thresholds(0.49, "custom_doc", None, thresholds) == RouteDecision.ADJUDICATION
|
|
assert evaluate_thresholds(0.30, "other", "custom_event", thresholds) == RouteDecision.FAST_PATH
|
|
|
|
def test_resolve_threshold_priority(self):
|
|
thresholds = FastPathThresholds()
|
|
# Event type takes priority
|
|
threshold = thresholds.resolve_threshold("article", "earnings_beat")
|
|
assert threshold == DEFAULT_EVENT_THRESHOLDS["earnings_beat"]
|
|
|
|
# Document type when no event
|
|
threshold = thresholds.resolve_threshold("article", None)
|
|
assert threshold == DEFAULT_DOCUMENT_THRESHOLDS["article"]
|
|
|
|
# Fallback for unknown
|
|
threshold = thresholds.resolve_threshold("mystery", None)
|
|
assert threshold == DEFAULT_FALLBACK_THRESHOLD
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 31.4 — Store every route decision and feature snapshot
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRoutingDecisionStore:
|
|
"""Test that decisions are stored with full feature snapshots."""
|
|
|
|
def test_store_and_retrieve_by_pipeline_run(self, engine, clean_features, clean_markers):
|
|
store = RoutingDecisionStore()
|
|
run_id = uuid4()
|
|
doc_id = uuid4()
|
|
|
|
decision = engine.route(
|
|
pipeline_run_id=run_id,
|
|
document_id=doc_id,
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
store.store(decision)
|
|
|
|
retrieved = store.get_by_pipeline_run(run_id)
|
|
assert len(retrieved) == 1
|
|
assert retrieved[0].id == decision.id
|
|
|
|
def test_decision_captures_confidence_snapshot(self, engine, clean_features, clean_markers):
|
|
run_id = uuid4()
|
|
doc_id = uuid4()
|
|
|
|
decision = engine.route(
|
|
pipeline_run_id=run_id,
|
|
document_id=doc_id,
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
|
|
assert "confidence_features" in decision.confidence_snapshot
|
|
assert "ambiguity_markers" in decision.confidence_snapshot
|
|
assert "thresholds_version" in decision.confidence_snapshot
|
|
assert decision.confidence_snapshot["confidence_features"] == clean_features
|
|
assert decision.confidence_snapshot["ambiguity_markers"] == clean_markers
|
|
|
|
def test_store_multiple_decisions_same_run(self, engine, clean_features, clean_markers):
|
|
store = RoutingDecisionStore()
|
|
run_id = uuid4()
|
|
|
|
for _ in range(3):
|
|
decision = engine.route(
|
|
pipeline_run_id=run_id,
|
|
document_id=uuid4(),
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
store.store(decision)
|
|
|
|
assert len(store.get_by_pipeline_run(run_id)) == 3
|
|
assert store.count() == 3
|
|
|
|
def test_get_by_unknown_run_returns_empty(self):
|
|
store = RoutingDecisionStore()
|
|
assert store.get_by_pipeline_run(uuid4()) == []
|
|
|
|
def test_decision_has_timestamp(self, engine, clean_features, clean_markers):
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.decided_at is not None
|
|
assert decision.decided_at.tzinfo is not None # UTC-aware
|
|
|
|
def test_decision_is_immutable(self, engine, clean_features, clean_markers):
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
with pytest.raises(Exception): # frozen dataclass
|
|
decision.route = RouteDecision.ADJUDICATION # type: ignore[misc]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration: full routing engine
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRoutingEngine:
|
|
"""Integration tests for the full routing path."""
|
|
|
|
def test_clean_document_gets_fast_path(self, engine, clean_features, clean_markers):
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.route == RouteDecision.FAST_PATH
|
|
assert RoutingReason.FAST_PATH_ACCEPTED in decision.reasons
|
|
|
|
def test_hard_rule_forces_adjudication(self, engine, clean_features, clean_markers):
|
|
clean_markers["contradictory_numeric_facts"] = True
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=clean_features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.route == RouteDecision.ADJUDICATION
|
|
assert RoutingReason.CONTRADICTORY_NUMERIC_FACTS in decision.reasons
|
|
|
|
def test_low_confidence_triggers_adjudication(self, engine, clean_markers):
|
|
features = {
|
|
"calibrated_confidence": 0.50,
|
|
"evidence_coverage": 0.95,
|
|
"material_fields_present": True,
|
|
}
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.route == RouteDecision.ADJUDICATION
|
|
assert RoutingReason.CALIBRATED_CONFIDENCE_BELOW_THRESHOLD in decision.reasons
|
|
|
|
def test_low_evidence_coverage_triggers_adjudication(self, engine, clean_markers):
|
|
features = {
|
|
"calibrated_confidence": 0.95,
|
|
"evidence_coverage": 0.30,
|
|
"material_fields_present": True,
|
|
}
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.route == RouteDecision.ADJUDICATION
|
|
assert RoutingReason.EVIDENCE_COVERAGE_BELOW_THRESHOLD in decision.reasons
|
|
|
|
def test_hard_rules_take_priority_over_threshold(self, engine):
|
|
"""Hard rules short-circuit — threshold is not even evaluated."""
|
|
features = {
|
|
"calibrated_confidence": 0.95,
|
|
"evidence_coverage": 0.95,
|
|
"material_fields_present": True,
|
|
}
|
|
markers = {
|
|
"unresolved_aliases": 1,
|
|
"primary_company_count": 1,
|
|
"contradictory_numeric_facts": False,
|
|
"conflicting_sentiment": False,
|
|
"implied_causal_impact": False,
|
|
"guidance_vs_consensus": False,
|
|
"long_document_cross_chunk": False,
|
|
}
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=features,
|
|
ambiguity_markers=markers,
|
|
document_type="article",
|
|
)
|
|
assert decision.route == RouteDecision.ADJUDICATION
|
|
assert RoutingReason.UNRESOLVED_ALIAS in decision.reasons
|
|
# Should NOT contain threshold reason since hard rules short-circuited
|
|
assert RoutingReason.CALIBRATED_CONFIDENCE_BELOW_THRESHOLD not in decision.reasons
|
|
|
|
def test_event_type_affects_threshold(self, engine, clean_markers):
|
|
"""Filing with merger event gets easier threshold (0.60)."""
|
|
features = {
|
|
"calibrated_confidence": 0.62,
|
|
"evidence_coverage": 0.80,
|
|
"material_fields_present": True,
|
|
}
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=features,
|
|
ambiguity_markers=clean_markers,
|
|
document_type="filing",
|
|
event_type="merger_acquisition",
|
|
)
|
|
assert decision.route == RouteDecision.FAST_PATH
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 31.5 — Property tests for determinism and threshold boundaries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
# Strategy for generating valid confidence features
|
|
confidence_features_strategy = st.fixed_dictionaries({
|
|
"calibrated_confidence": st.floats(min_value=0.0, max_value=1.0),
|
|
"evidence_coverage": st.floats(min_value=0.0, max_value=1.0),
|
|
"material_fields_present": st.booleans(),
|
|
})
|
|
|
|
# Strategy for generating ambiguity markers
|
|
ambiguity_markers_strategy = st.fixed_dictionaries({
|
|
"unresolved_aliases": st.integers(min_value=0, max_value=10),
|
|
"primary_company_count": st.integers(min_value=0, max_value=5),
|
|
"contradictory_numeric_facts": st.booleans(),
|
|
"conflicting_sentiment": st.booleans(),
|
|
"implied_causal_impact": st.booleans(),
|
|
"guidance_vs_consensus": st.booleans(),
|
|
"long_document_cross_chunk": st.booleans(),
|
|
})
|
|
|
|
document_type_strategy = st.sampled_from(
|
|
["article", "filing", "transcript", "press_release", "macro_event", "unknown"]
|
|
)
|
|
|
|
event_type_strategy = st.one_of(
|
|
st.none(),
|
|
st.sampled_from([
|
|
"earnings_beat", "earnings_miss", "guidance_change",
|
|
"management_change", "merger_acquisition", "regulatory_action",
|
|
"product_launch", "legal_action", "rating_change", "supply_chain",
|
|
"unknown_event",
|
|
]),
|
|
)
|
|
|
|
|
|
class TestDeterminismProperty:
|
|
"""Property test: same inputs always produce the same route.
|
|
|
|
**Validates: Requirements 10.5**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
confidence_features=confidence_features_strategy,
|
|
ambiguity_markers=ambiguity_markers_strategy,
|
|
document_type=document_type_strategy,
|
|
event_type=event_type_strategy,
|
|
)
|
|
def test_same_inputs_always_same_route(
|
|
self,
|
|
confidence_features: dict,
|
|
ambiguity_markers: dict,
|
|
document_type: str,
|
|
event_type: str | None,
|
|
):
|
|
"""Route decisions are deterministic: same inputs → same output."""
|
|
engine = RoutingEngine()
|
|
run_id = uuid4()
|
|
doc_id = uuid4()
|
|
|
|
decision_1 = engine.route(
|
|
pipeline_run_id=run_id,
|
|
document_id=doc_id,
|
|
confidence_features=confidence_features,
|
|
ambiguity_markers=ambiguity_markers,
|
|
document_type=document_type,
|
|
event_type=event_type,
|
|
)
|
|
decision_2 = engine.route(
|
|
pipeline_run_id=run_id,
|
|
document_id=doc_id,
|
|
confidence_features=confidence_features,
|
|
ambiguity_markers=ambiguity_markers,
|
|
document_type=document_type,
|
|
event_type=event_type,
|
|
)
|
|
|
|
assert decision_1.route == decision_2.route
|
|
assert decision_1.reasons == decision_2.reasons
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
confidence_features=confidence_features_strategy,
|
|
ambiguity_markers=ambiguity_markers_strategy,
|
|
document_type=document_type_strategy,
|
|
event_type=event_type_strategy,
|
|
)
|
|
def test_route_is_always_valid_enum(
|
|
self,
|
|
confidence_features: dict,
|
|
ambiguity_markers: dict,
|
|
document_type: str,
|
|
event_type: str | None,
|
|
):
|
|
"""Route decision is always a valid RouteDecision enum value."""
|
|
engine = RoutingEngine()
|
|
decision = engine.route(
|
|
pipeline_run_id=uuid4(),
|
|
document_id=uuid4(),
|
|
confidence_features=confidence_features,
|
|
ambiguity_markers=ambiguity_markers,
|
|
document_type=document_type,
|
|
event_type=event_type,
|
|
)
|
|
assert decision.route in (RouteDecision.FAST_PATH, RouteDecision.ADJUDICATION)
|
|
assert len(decision.reasons) > 0
|
|
|
|
|
|
class TestThresholdBoundaryProperty:
|
|
"""Property test: confidence at exact threshold boundary is deterministic.
|
|
|
|
**Validates: Requirements 10.5, 11.6**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
document_type=st.sampled_from(list(DEFAULT_DOCUMENT_THRESHOLDS.keys())),
|
|
)
|
|
def test_at_threshold_is_always_fast_path(self, document_type: str):
|
|
"""Confidence exactly at threshold always results in fast path."""
|
|
thresholds = FastPathThresholds()
|
|
threshold_value = thresholds.resolve_threshold(document_type, None)
|
|
|
|
# At the boundary
|
|
result = evaluate_thresholds(threshold_value, document_type, None, thresholds)
|
|
assert result == RouteDecision.FAST_PATH
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
document_type=st.sampled_from(list(DEFAULT_DOCUMENT_THRESHOLDS.keys())),
|
|
epsilon=st.floats(min_value=1e-15, max_value=0.1),
|
|
)
|
|
def test_below_threshold_is_always_adjudication(
|
|
self, document_type: str, epsilon: float
|
|
):
|
|
"""Confidence below threshold always results in adjudication."""
|
|
thresholds = FastPathThresholds()
|
|
threshold_value = thresholds.resolve_threshold(document_type, None)
|
|
below = threshold_value - epsilon
|
|
|
|
if below >= 0.0:
|
|
result = evaluate_thresholds(below, document_type, None, thresholds)
|
|
assert result == RouteDecision.ADJUDICATION
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
document_type=st.sampled_from(list(DEFAULT_DOCUMENT_THRESHOLDS.keys())),
|
|
epsilon=st.floats(min_value=1e-15, max_value=0.1),
|
|
)
|
|
def test_above_threshold_is_always_fast_path(
|
|
self, document_type: str, epsilon: float
|
|
):
|
|
"""Confidence above threshold always results in fast path."""
|
|
thresholds = FastPathThresholds()
|
|
threshold_value = thresholds.resolve_threshold(document_type, None)
|
|
above = threshold_value + epsilon
|
|
|
|
if above <= 1.0:
|
|
result = evaluate_thresholds(above, document_type, None, thresholds)
|
|
assert result == RouteDecision.FAST_PATH
|