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