""" Dependency injection module for the ARF Agentic Reliability Framework API. Provides FastAPI dependencies for database sessions, rate limiting, and singleton instances of the core ARF engines (RiskEngine, DecisionEngine, LyapunovStabilityController, CausalEffectEstimator, RAGGraphMemory, and (v4.3.1) SkillRegistry). All engine dependencies are lazily initialised and cached for the lifetime of the application process. v4.3.2: Added verify_internal_key dependency to secure direct API access. """ import os import sys from app.database.session import SessionLocal from slowapi import Limiter from slowapi.util import get_remote_address from app.core.config import settings from fastapi import Header, HTTPException, Request # ARF core engine imports from agentic_reliability_framework.core.governance.risk_engine import RiskEngine from agentic_reliability_framework.core.decision.decision_engine import DecisionEngine from agentic_reliability_framework.core.governance.stability_controller import LyapunovStabilityController from agentic_reliability_framework.core.governance.causal_effect_estimator import CausalEffectEstimator from agentic_reliability_framework.runtime.memory.rag_graph import RAGGraphMemory from agentic_reliability_framework.core.models.event import ReliabilityEvent, HealingAction # ── v4.3.1: Skill Registry (optional) ────────────────────────── try: from agentic_reliability_framework.core.governance.skill_registry import SkillRegistry _SKILL_REGISTRY_AVAILABLE = True except ImportError: SkillRegistry = None _SKILL_REGISTRY_AVAILABLE = False # --------------------------------------------------------------------------- # Database dependency # --------------------------------------------------------------------------- def get_db(): """ Yield a SQLAlchemy database session and ensure it is closed after use. This dependency is intended to be used with FastAPI's `Depends` mechanism. """ db = SessionLocal() try: yield db finally: db.close() # --------------------------------------------------------------------------- # Rate limiter # --------------------------------------------------------------------------- limiter = Limiter( key_func=get_remote_address, default_limits=[settings.RATE_LIMIT], ) # --------------------------------------------------------------------------- # Internal API key verification (v4.3.2) # --------------------------------------------------------------------------- INTERNAL_API_KEY = os.getenv("ARF_INTERNAL_API_KEY", "") async def verify_internal_key(x_internal_key: str = Header(default=None, alias="X-Internal-Key")): """ FastAPI dependency that verifies the internal API key header. If the environment variable ARF_INTERNAL_API_KEY is set, the request must include a matching X‑Internal‑Key header. Without the key, a 401 Unauthorized response is returned. This guards against direct access to the API when deployed behind the Go gateway. The gateway is configured to inject this header for authenticated requests. """ if INTERNAL_API_KEY: if x_internal_key is None: raise HTTPException(status_code=401, detail="Missing internal API key") # Use a constant‑time comparison to avoid timing attacks. if not _constant_time_compare(x_internal_key, INTERNAL_API_KEY): raise HTTPException(status_code=401, detail="Invalid internal API key") # If no key is configured, the dependency is a no‑op (for local dev). def _constant_time_compare(a: str, b: str) -> bool: """Compare two strings in constant time to prevent timing attacks.""" if len(a) != len(b): return False result = 0 for x, y in zip(a, b): result |= ord(x) ^ ord(y) return result == 0 # --------------------------------------------------------------------------- # Singleton engine instances (lazy, cached) # --------------------------------------------------------------------------- _risk_engine = None _decision_engine = None _stability_controller = None _causal_explainer = None _rag_graph = None _skill_registry = None def _seed_rag_graph(rag: RAGGraphMemory) -> None: """ Populate the RAG graph with a small set of synthetic historical healing‑action outcomes to provide initial memory for the decision engine. Parameters ---------- rag : RAGGraphMemory An already‑instantiated RAG graph memory instance. """ seed_data = [ ("seed_restart_1", "test", HealingAction.RESTART_CONTAINER.value, True, 2), ("seed_restart_2", "test", HealingAction.RESTART_CONTAINER.value, True, 3), ("seed_restart_3", "test", HealingAction.RESTART_CONTAINER.value, False, 10), ("seed_rollback_1", "test", HealingAction.ROLLBACK.value, True, 1), ("seed_rollback_2", "test", HealingAction.ROLLBACK.value, True, 2), ("seed_rollback_3", "test", HealingAction.ROLLBACK.value, False, 5), ("seed_scale_1", "test", HealingAction.SCALE_OUT.value, True, 5), ("seed_scale_2", "test", HealingAction.SCALE_OUT.value, False, 15), ("seed_cb_1", "test", HealingAction.CIRCUIT_BREAKER.value, True, 1), ("seed_cb_2", "test", HealingAction.CIRCUIT_BREAKER.value, True, 2), ("seed_ts_1", "test", HealingAction.TRAFFIC_SHIFT.value, True, 4), ("seed_ts_2", "test", HealingAction.TRAFFIC_SHIFT.value, False, 8), ] for inc_id, comp, action, success, res_time in seed_data: event = ReliabilityEvent( component=comp, latency_p99=500, error_rate=0.1, service_mesh="default", ) rag.record_outcome( incident_id=inc_id, event=event, action_taken=action, success=success, resolution_time_minutes=res_time, ) print("Seeded RAG graph with historical data", file=sys.stderr) def get_rag_graph() -> RAGGraphMemory: """ Return a singleton instance of the RAG graph memory, seeded with synthetic historical data on first access. """ global _rag_graph if _rag_graph is None: _rag_graph = RAGGraphMemory() _seed_rag_graph(_rag_graph) return _rag_graph def get_decision_engine() -> DecisionEngine: """ Return a singleton DecisionEngine, wiring it to the shared RAG graph memory. """ global _decision_engine if _decision_engine is None: rag = get_rag_graph() _decision_engine = DecisionEngine(rag_graph=rag) return _decision_engine def get_risk_engine() -> RiskEngine: """ Return a singleton RiskEngine instance. """ global _risk_engine if _risk_engine is None: _risk_engine = RiskEngine() return _risk_engine def get_stability_controller() -> LyapunovStabilityController: """ Return a singleton LyapunovStabilityController instance. """ global _stability_controller if _stability_controller is None: _stability_controller = LyapunovStabilityController() return _stability_controller def get_causal_explainer() -> CausalEffectEstimator: """ Return a singleton CausalEffectEstimator instance. The estimator uses Inverse Probability Weighting (IPW) and causal forests to provide counterfactual explanations for governance decisions. """ global _causal_explainer if _causal_explainer is None: _causal_explainer = CausalEffectEstimator() return _causal_explainer def get_skill_registry() -> "Optional[SkillRegistry]": """ Return a singleton SkillRegistry instance (v4.3.1). The registry manages procedural skill artefacts, versioning, per‑skill reliability models (Beta‑Binomial), and the COLLECT‑DIAGNOSE‑REVISE‑PROMOTE evolution loop. If the SkillRegistry module is not installed, returns None. """ global _skill_registry if not _SKILL_REGISTRY_AVAILABLE: return None if _skill_registry is None: from agentic_reliability_framework.core.governance.skill_registry import SkillRegistry _skill_registry = SkillRegistry() return _skill_registry