from __future__ import annotations from dataclasses import asdict, dataclass from typing import Any from router.query_classifier import RouteDecisionFlags @dataclass class StoreSufficiencyResult: canonical_sufficient: bool feature_sufficient: bool risk_sufficient: bool freshness_ok: bool aggregate_level_ok: bool ranking_ready: bool explanation_ready: bool reason_codes: list[str] def to_dict(self) -> dict[str, Any]: return asdict(self) def _to_float(value: Any, default: float = 0.0) -> float: try: return float(value) except (TypeError, ValueError): return default def check_store_sufficiency( question_shape: RouteDecisionFlags, store_metadata: dict[str, Any], ) -> StoreSufficiencyResult: reason_codes: list[str] = [] freshness_threshold_hours = _to_float(store_metadata.get("freshness_threshold_hours", 6.0), default=6.0) refresh_age = _to_float(store_metadata.get("refresh_age_hours", 0.0), default=0.0) feature_age = _to_float(store_metadata.get("feature_age_hours", refresh_age), default=refresh_age) risk_age = _to_float(store_metadata.get("risk_age_hours", refresh_age), default=refresh_age) canonical_semantic_coverage = _to_float(store_metadata.get("canonical_semantic_coverage", 0.0), default=0.0) canonical_relation_types = int(store_metadata.get("canonical_relation_types", 0)) canonical_links_total = int(store_metadata.get("canonical_links_total", 0)) canonical_entities_total = int(store_metadata.get("canonical_entities_total", 0)) feature_ready = bool(store_metadata.get("feature_ready", False)) risk_ready = bool(store_metadata.get("risk_ready", False)) ranking_ready = bool(store_metadata.get("ranking_ready", False)) aggregate_ready = bool(store_metadata.get("aggregate_ready", False)) freshness_ok = refresh_age <= freshness_threshold_hours if question_shape.freshness_sensitive and not freshness_ok: reason_codes.append("refresh_stale") canonical_sufficient = ( canonical_entities_total > 0 and canonical_links_total > 0 and canonical_semantic_coverage >= 0.85 and canonical_relation_types >= 10 and (freshness_ok or not question_shape.freshness_sensitive) ) if not canonical_sufficient: reason_codes.append("canonical_not_sufficient") feature_sufficient = feature_ready and ( feature_age <= freshness_threshold_hours or not question_shape.freshness_sensitive ) if not feature_sufficient and question_shape.needs_anomaly_summary: reason_codes.append("feature_not_sufficient") risk_sufficient = risk_ready and ( risk_age <= freshness_threshold_hours or not question_shape.freshness_sensitive ) if not risk_sufficient and (question_shape.needs_anomaly_summary or question_shape.needs_ranking): reason_codes.append("risk_not_sufficient") if question_shape.needs_ranking: if not ranking_ready: reason_codes.append("ranking_not_ready") aggregate_level_ok = aggregate_ready and ( (not question_shape.needs_ranking or ranking_ready) ) and question_shape.precomputed_aggregate_available if not aggregate_level_ok and ( question_shape.needs_full_period_aggregation or question_shape.needs_ranking or question_shape.needs_anomaly_summary ): reason_codes.append("aggregate_not_sufficient") explanation_ready = ( canonical_sufficient and canonical_semantic_coverage >= 0.90 and canonical_relation_types >= 20 and not question_shape.needs_runtime_truth and not (question_shape.needs_cross_entity_join and question_shape.needs_causal_chain) ) if not explanation_ready and (question_shape.needs_causal_chain or question_shape.needs_cross_entity_join): reason_codes.append("explanation_not_ready") return StoreSufficiencyResult( canonical_sufficient=canonical_sufficient, feature_sufficient=feature_sufficient, risk_sufficient=risk_sufficient, freshness_ok=freshness_ok, aggregate_level_ok=aggregate_level_ok, ranking_ready=ranking_ready, explanation_ready=explanation_ready, reason_codes=reason_codes, )