Spaces:
Build error
Build error
File size: 10,504 Bytes
5bbe84d 2f536a5 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 5a3a712 5bbe84d 5a3a712 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 2f536a5 5a3a712 2f536a5 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 5a3a712 5bbe84d 5a3a712 5bbe84d 5a3a712 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 50bdbd0 5bbe84d 5a3a712 5bbe84d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 | """
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 logging
import os
import sys
from typing import Optional
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
# 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
# ---------------------------------------------------------------------------
logger = logging.getLogger(__name__)
_RATE_LIMIT_FALLBACK = "100/minute"
def _resolve_rate_limit(configured: str | None) -> str:
"""Return a parseable rate-limit string, falling back loudly.
An unparseable RATE_LIMIT takes the whole service down, and does it in
the most confusing way available: `Limiter` accepts any string, so the
app boots clean and reports healthy, and the parse only happens inside
SlowAPIMiddleware on each request -- which raises, hits main.py's
catch-all handler, and returns 500 for *every* route including
/health, which otherwise touches nothing. Startup logs look perfect
while the service serves nothing.
An empty value is the easy way to land there: setting RATE_LIMIT to a
blank string in a dashboard doesn't restore the default, it overrides
it with "" (pydantic treats blank as provided), and "" doesn't parse.
Rate limiting is a protective control, not a correctness one, so a bad
value degrades to the documented default rather than refusing to boot
-- turning a typo into an outage is the failure mode being fixed here,
not one to reintroduce. It is logged at CRITICAL because running with
a limit nobody chose is not something to discover later.
"""
candidate = (configured or "").strip()
if not candidate:
logger.warning(
"RATE_LIMIT is unset or blank; using the default %r. A blank value "
"does not restore the default on its own -- it overrides it.",
_RATE_LIMIT_FALLBACK,
)
return _RATE_LIMIT_FALLBACK
try:
for group in Limiter(key_func=get_remote_address, default_limits=[candidate])._default_limits:
for _ in group:
pass
except Exception:
logger.critical(
"RATE_LIMIT=%r is not a parseable rate limit (expected e.g. '100/minute'); "
"falling back to %r. Left as-is this returns 500 on every request, "
"health checks included.",
candidate,
_RATE_LIMIT_FALLBACK,
)
return _RATE_LIMIT_FALLBACK
return candidate
limiter = Limiter(
key_func=get_remote_address,
default_limits=[_resolve_rate_limit(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.
The request must include an X‑Internal‑Key header matching
ARF_INTERNAL_API_KEY. This fails closed: if ARF_INTERNAL_API_KEY is not
configured, every request is rejected with 401 rather than being let
through unauthenticated.
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 not INTERNAL_API_KEY:
raise HTTPException(status_code=401, detail="Internal API key is not configured")
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")
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
|