Upload folder using huggingface_hub (#74)

- Upload folder using huggingface_hub (000e62aa83214d6f7f6145dbb45de5082befbf2d)

Dockerfile

@@ -1,13 +1,7 @@
 FROM python:3.12-slim
 RUN apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*
 WORKDIR /app
-
 COPY requirements.txt .
-
-# Use the secret ARF_GITHUB_PAT during the pip install step
-RUN --mount=type=secret,id=ARF_GITHUB_PAT \
-    git config --global url."https://x-access-token:$(cat /run/secrets/ARF_GITHUB_PAT)@github.com/".insteadOf "https://github.com/" && \
-    pip install --no-cache-dir -r requirements.txt
-
+RUN pip install --no-cache-dir -r requirements.txt
 COPY . .
-CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,9 +1,3 @@
----
-title: ARF API Control Plane
-sdk: docker
-colorFrom: blue
-colorTo: green
----
 # arf-api
 
 ARF API Control Plane (FastAPI)
@@ -11,8 +5,8 @@ ARF API Control Plane (FastAPI)
 ## Live Demo
 
 The API is deployed and accessible at:
-- **Base URL**: [https://a-r-f-agentic-reliability-framework-api.hf.space](https://a-r-f-agentic-reliability-framework-api.hf.space)
-- **Interactive Documentation**: [https://a-r-f-agentic-reliability-framework-api.hf.space/docs](https://a-r-f-agentic-reliability-framework-api.hf.space/docs)
+- **Base URL**: [https://arf-ai-agentic-reliability-framework-api.hf.space](https://arf-ai-agentic-reliability-framework-api.hf.space)
+- **Interactive Documentation**: [https://arf-ai-agentic-reliability-framework-api.hf.space/docs](https://arf-ai-agentic-reliability-framework-api.hf.space/docs)
 
 ## Quick Start (Local Development)
 
@@ -87,7 +81,7 @@ curl -X POST "http://localhost:8000/api/v1/v1/incidents/evaluate"   -H "Content-
     "justification": "Causal: If we apply restart_container instead of no_action, latency would change from 600.00 to 510.00 (Δ = -90.00). Based on heuristic causal model.",
     "confidence": 0.85,
     "risk_score": 0.54,
-    "status": "oss_advisory_only"
+    "status": "success"
   },
   "causal_explanation": {
     "factual_outcome": 600,
@@ -117,10 +111,11 @@ curl -X POST "http://localhost:8000/api/v1/v1/incidents/evaluate"   -H "Content-
 Tests
 -----
 
-Run `pytest`. Tests use a temporary SQLite DB (`sqlite:///./test.db`) created by the test fixtures.
+Run `pytest`. Tests run against a live Postgres connection (`tests/conftest.py`), matching CI's `postgres` service — not a temporary SQLite DB.
 
 Notes
 -----
 
 - The governance endpoints use an in-process `RiskEngine` initialized at startup.
-- The outcome recording endpoint is not implemented in this repository and returns HTTP 501.
+- Outcomes are recorded via `POST /api/v1/intents/outcome` (`app/api/routes_governance.py`), tenant-scoped and auth-protected.
+

app/api/deps.py

@@ -1,20 +1,52 @@
+"""
+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 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_explainer import CausalExplainer
+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
+# ---------------------------------------------------------------------------
 
-# Dependency to get DB session
 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
@@ -22,23 +54,77 @@ def get_db():
         db.close()
 
 
-# Rate limiter with default limit from settings
+# ---------------------------------------------------------------------------
+# Rate limiter
+# ---------------------------------------------------------------------------
+
 limiter = Limiter(
     key_func=get_remote_address,
-    default_limits=[
-        settings.RATE_LIMIT])
+    default_limits=[settings.RATE_LIMIT],
+)
+
+
+# ---------------------------------------------------------------------------
+# Internal API key verification (v4.3.2)
+# ---------------------------------------------------------------------------
+
+INTERNAL_API_KEY = os.getenv("ARF_INTERNAL_API_KEY", "")
 
 
-# ARF engine dependencies (singletons for simplicity)
+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.
 
-def _seed_rag_graph(rag):
-    """Seed the RAG graph with historical healing action outcomes."""
+    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),
@@ -58,19 +144,23 @@ def _seed_rag_graph(rag):
             component=comp,
             latency_p99=500,
             error_rate=0.1,
-            service_mesh="default"
+            service_mesh="default",
         )
         rag.record_outcome(
             incident_id=inc_id,
             event=event,
             action_taken=action,
             success=success,
-            resolution_time_minutes=res_time
+            resolution_time_minutes=res_time,
         )
     print("Seeded RAG graph with historical data", file=sys.stderr)
 
 
-def get_rag_graph():
+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()
@@ -78,7 +168,11 @@ def get_rag_graph():
     return _rag_graph
 
 
-def get_decision_engine():
+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()
@@ -86,22 +180,51 @@ def get_decision_engine():
     return _decision_engine
 
 
-def get_risk_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():
+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():
+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 = CausalExplainer()
+        _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

app/api/routes_admin.py

@@ -6,16 +6,21 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Path, Body
 from pydantic import BaseModel
 from typing import Optional
 from datetime import datetime
+import os
+import secrets
 import uuid
 from app.core.usage_tracker import tracker, Tier
 
 router = APIRouter(prefix="/admin", tags=["admin"])
-# Simple in‑memory admin key (replace with proper auth in production)
-ADMIN_API_KEY = "admin_secret_change_me"
+# Admin key must be supplied via environment; there is no default. Fail closed
+# if it is not configured, rather than falling back to a guessable secret.
+ADMIN_API_KEY = os.getenv("ARF_ADMIN_API_KEY")
 
 
 def verify_admin(admin_key: str = Query(..., alias="admin_key")):
-    if admin_key != ADMIN_API_KEY:
+    if not ADMIN_API_KEY:
+        raise HTTPException(status_code=403, detail="Admin API is not configured")
+    if not secrets.compare_digest(admin_key, ADMIN_API_KEY):
         raise HTTPException(status_code=403, detail="Invalid admin key")
     return True
 
@@ -39,6 +44,7 @@ async def create_api_key(req: CreateKeyRequest):
     return {"api_key": new_key, "tier": req.tier}
 
 
+@router.get("/keys", dependencies=[Depends(verify_admin)])
 async def list_api_keys(limit: int = 100, offset: int = 0):
     with tracker._get_conn() as conn:
         rows = conn.execute(
@@ -101,6 +107,7 @@ async def deactivate_api_key(
     return {"message": "API key deactivated"}
 
 
+@router.get("/keys/{api_key}/audit", dependencies=[Depends(verify_admin)])
 async def get_audit_logs(
     api_key: str = Path(..., description="The API key to audit"),
     start_date: Optional[str] = Query(None),
@@ -113,6 +120,7 @@ async def get_audit_logs(
     return {"api_key": api_key, "logs": logs}
 
 
+@router.get("/stats", dependencies=[Depends(verify_admin)])
 async def get_global_stats():
     with tracker._get_conn() as conn:
         total_keys = conn.execute(

app/api/routes_governance.py

@@ -1,25 +1,57 @@
+"""
+Routes for governance evaluation – tenant‑aware, audited, and Rust‑enforced.
+
+This module provides the primary API endpoints for evaluating infrastructure
+intents and healing decisions. It integrates:
+
+- Idempotent quota consumption (usage tracker)
+- Tenant isolation (tenant_id resolved server-side from the authenticated API key
+  via the ``enforce_quota`` dependency; never taken from a client-supplied header)
+- Auditable decision logging (DecisionAuditLogDB)
+- Pricing telemetry (optional, to arf‑pricing‑calculator)
+- OpenTelemetry tracing
+- Optional Rust execution ladder for mechanical enforcement
+- **v4.3.1**: Full governance loop produces a Bayesian HealingIntent with skill
+  posterior parameters (α, β) for the enterprise SkillGate.
+  Includes persistent stability controller and temporal monitor for
+  cross‑request state accumulation, and a merging policy evaluator that
+  respects both external and internal policy violations.
+  Healing endpoint now optionally accepts skill context for Bayesian
+  utility‑aware action selection.
+- **v4.3.2**: Passes criticality parameter for dynamic gate tuning (Feature 3).
+  Internal API key verification added to secure direct access.
+"""
+
 from fastapi import APIRouter, Depends, HTTPException, Request, BackgroundTasks, Header
 from fastapi.encoders import jsonable_encoder
 from sqlalchemy.orm import Session
-from app.models.infrastructure_intents import InfrastructureIntentRequest
-from app.services.intent_adapter import to_oss_intent
-from app.services.risk_service import evaluate_intent, evaluate_healing_decision
-from app.services.intent_store import save_evaluated_intent
-from app.services.outcome_service import record_outcome
-from app.api.deps import get_db
 from pydantic import BaseModel
 import uuid
 import logging
 import time
-from typing import Optional
+import datetime
+from typing import Optional, Dict, Any, List
 
+from app.models.infrastructure_intents import InfrastructureIntentRequest
+from app.services.intent_adapter import to_oss_intent
+from app.services.risk_service import evaluate_intent_full, evaluate_healing_decision
+from app.services.intent_store import save_evaluated_intent
+from app.services.outcome_service import record_outcome
+from app.api.deps import get_db, get_skill_registry, verify_internal_key          # <-- v4.3.2
+from app.database.session import SessionLocal
+from app.core.usage_tracker import enforce_quota                                  # <-- tenant resolution
+from app.database.models_intents import DecisionAuditLogDB
 from agentic_reliability_framework.core.models.event import ReliabilityEvent
+from agentic_reliability_framework.core.governance.policies import (
+    PolicyEvaluator,
+    allow_all,
+)
 
-# ===== USAGE TRACKER IMPORTS =====
+# ===== USAGE TRACKER =====
 import app.core.usage_tracker
 from app.core.usage_tracker import UsageRecord
 
-# ===== PRICING CALCULATOR INTEGRATION =====
+# ===== PRICING CALCULATOR =====
 try:
     from arf_pricing_calculator.storage.buffer import add_event
     PRICING_AVAILABLE = True
@@ -27,7 +59,15 @@ except ImportError:
     PRICING_AVAILABLE = False
     add_event = None
 
-# ===== OpenTelemetry (optional) =====
+# ===== RUST EXECUTION LADDER (optional) =====
+try:
+    from arf_enterprise.execution_ladder import ExecutionLadder
+    RUST_AVAILABLE = True
+except ImportError:
+    RUST_AVAILABLE = False
+    ExecutionLadder = None
+
+# ===== OPEN TELEMETRY =====
 try:
     from opentelemetry import trace
     from opentelemetry.trace import Status, StatusCode
@@ -38,7 +78,9 @@ except ImportError:
     _tracer = None
 
 logger = logging.getLogger(__name__)
-router = APIRouter()
+
+# v4.3.2: protect all governance endpoints with internal API key verification
+router = APIRouter(dependencies=[Depends(verify_internal_key)])
 
 
 class OutcomeRequest(BaseModel):
@@ -46,12 +88,120 @@ class OutcomeRequest(BaseModel):
     success: bool
     recorded_by: str
     notes: str = ""
+    # v4.3.1: optional skill provenance for reliability feedback
+    skill_id: Optional[str] = None
+    skill_version: Optional[int] = None
 
 
 class HealingDecisionRequest(BaseModel):
     event: ReliabilityEvent
+    # v4.3.1: optional skill context for Bayesian utility
+    skill_id: Optional[str] = None
+    skill_version: Optional[int] = None
+
+
+# --------------------------------------------------------------------------
+# Helper: write audit log (idempotent)
+# --------------------------------------------------------------------------
+async def write_audit_log(
+    tenant_id: str,
+    deterministic_id: str,
+    healing_intent: Dict[str, Any],
+    trace_id: Optional[str] = None,
+    idempotency_key: Optional[str] = None,
+) -> None:
+    """
+    Store a governance decision in the immutable audit log.
+    Idempotent on (tenant_id, deterministic_id) – if already exists, skip.
+
+    Runs as a BackgroundTask, which executes after the response has already
+    been sent -- and after FastAPI has already torn down the request's
+    `Depends(get_db)` session. Reusing that session here would mean every
+    query silently reopens a fresh connection/transaction that nothing then
+    guarantees gets closed (least of all on the idempotent-skip path below,
+    which used to return with no commit/rollback/close at all), leaving an
+    idle-in-transaction connection that blocks any later DDL against these
+    tables (e.g. test teardown's `Base.metadata.drop_all()`) indefinitely.
+    Owning and closing our own session here avoids that entirely.
+    """
+    db = SessionLocal()
+    try:
+        # Check if already logged (idempotency)
+        existing = db.query(DecisionAuditLogDB).filter(
+            DecisionAuditLogDB.tenant_id == tenant_id,
+            DecisionAuditLogDB.deterministic_id == deterministic_id
+        ).first()
+        if existing:
+            logger.info(f"Audit log already exists for {deterministic_id}, skipping.")
+            return
+
+        # Extract fields that are actually present in DecisionAuditLogDB
+        risk_score = healing_intent.get("risk_score", 0.5)
+        action = healing_intent.get("recommended_action", "deny")
+        justification = healing_intent.get("justification", "")
+        metadata = healing_intent.get("metadata", {})
+        memory_success_rate = metadata.get("memory_success_rate")
+        memory_weight = metadata.get("memory_weight")
+        counterfactual = metadata.get("counterfactual")
+
+        audit_entry = DecisionAuditLogDB(
+            tenant_id=tenant_id,
+            deterministic_id=deterministic_id,
+            timestamp=datetime.datetime.utcnow(),
+            risk_score=risk_score,
+            action=action,
+            justification=justification,
+            memory_success_rate=memory_success_rate,
+            memory_weight=memory_weight,
+            counterfactual=counterfactual,
+            trace_id=trace_id,
+        )
+        db.add(audit_entry)
+        db.commit()
+        logger.info(f"Audit log written for {deterministic_id}")
+    finally:
+        db.close()
 
 
+# --------------------------------------------------------------------------
+# Policy evaluator that merges external violations with internal checks
+# --------------------------------------------------------------------------
+class MergingPolicyEvaluator(PolicyEvaluator):
+    """
+    A policy evaluator that combines a base evaluator (the governance loop's
+    own policy tree) with a set of pre‑computed violations (e.g., from an
+    external Rust enforcer or the request body). The effective violation list
+    is the union of both sources, preserving order and removing duplicates.
+    """
+    def __init__(self, base_evaluator: PolicyEvaluator, pre_violations: List[str]):
+        # We must call the PolicyEvaluator constructor with a root policy,
+        # but the base evaluator will be used for actual evaluation.
+        super().__init__(base_evaluator.get_root_policy())
+        self._base = base_evaluator
+        self._pre = list(pre_violations)
+
+    def evaluate(self, intent, context=None):
+        base_violations = self._base.evaluate(intent, context)
+        # Merge with pre‑computed violations, preserving order and removing duplicates
+        merged = []
+        seen = set()
+        for v in self._pre:
+            if v not in seen:
+                merged.append(v)
+                seen.add(v)
+        for v in base_violations:
+            if v not in seen:
+                merged.append(v)
+                seen.add(v)
+        return merged
+
+    def get_root_policy(self):
+        return self._base.get_root_policy()
+
+
+# --------------------------------------------------------------------------
+# Endpoint: evaluate infrastructure intent
+# --------------------------------------------------------------------------
 @router.post("/intents/evaluate")
 async def evaluate_intent_endpoint(
     request: Request,
@@ -59,11 +209,14 @@ async def evaluate_intent_endpoint(
     background_tasks: BackgroundTasks,
     db: Session = Depends(get_db),
     idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
+    skill_registry=Depends(get_skill_registry),   # v4.3.1
+    quota: dict = Depends(enforce_quota),           # tenant resolved from authenticated API key
 ):
     """
-    Evaluate an infrastructure intent with idempotency and atomic quota consumption.
+    Evaluate an infrastructure intent with idempotency, tenant isolation,
+    full governance loop analysis, Bayesian skill posterior injection,
+    and optional criticality parameter for dynamic gate tuning (v4.3.2).
     """
-    # ── optional trace ──────────────────────────────────────
     span = None
     if OTEL_AVAILABLE and _tracer:
         span = _tracer.start_span("governance.evaluate_intent")
@@ -71,17 +224,17 @@ async def evaluate_intent_endpoint(
         span.set_attribute("environment", str(intent_req.environment))
 
     start_time = time.time()
-    api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
-    if not api_key:
-        api_key = request.query_params.get("api_key", "unknown")
+    # api_key/tenant_id are resolved server-side by enforce_quota from the
+    # authenticated principal — never from a client-supplied header.
+    api_key = quota["api_key"]
+    tenant_id = quota["tenant_id"]
 
     current_tracker = app.core.usage_tracker.tracker
     if current_tracker is None:
         if span:
             span.set_status(Status(StatusCode.ERROR, "tracker unavailable"))
             span.end()
-        raise HTTPException(status_code=503,
-                            detail="Usage tracking service unavailable")
+        raise HTTPException(status_code=503, detail="Usage tracking service unavailable")
 
     record = UsageRecord(
         api_key=api_key,
@@ -102,40 +255,84 @@ async def evaluate_intent_endpoint(
         if existing_response:
             return existing_response
         else:
-            raise HTTPException(status_code=429,
-                                detail="Monthly evaluation quota exceeded")
+            raise HTTPException(status_code=429, detail="Monthly evaluation quota exceeded")
 
     try:
         oss_intent = to_oss_intent(intent_req)
         risk_engine = request.app.state.risk_engine
-        result = evaluate_intent(
-            engine=risk_engine,
+
+        # Build the base policy evaluator from the app's policy engine (if available)
+        policy_engine = getattr(request.app.state, "policy_engine", None)
+        if policy_engine is not None and hasattr(policy_engine, 'root_policy'):
+            base_evaluator = PolicyEvaluator(policy_engine.root_policy)
+        else:
+            base_evaluator = PolicyEvaluator(allow_all())
+
+        # Wrap it to also include the pre‑computed violations from the request
+        policy_evaluator = MergingPolicyEvaluator(
+            base_evaluator,
+            intent_req.policy_violations
+        )
+
+        # Optional components from app state
+        memory = getattr(request.app.state, "rag_graph", None)
+        hallucination_probe = getattr(request.app.state, "epistemic_probe", None)
+        predictive_engine = getattr(request.app.state, "predictive_engine", None)
+        business_calculator = getattr(request.app.state, "business_calculator", None)
+
+        # Stateful monitors (v4.3.1)
+        stability_controller = getattr(request.app.state, "stability_controller", None)
+        temporal_monitor = getattr(request.app.state, "temporal_monitor", None)
+
+        # Run the full governance loop, injecting skill context and criticality if present
+        result = evaluate_intent_full(
             intent=oss_intent,
-            cost_estimate=intent_req.estimated_cost,
-            policy_violations=intent_req.policy_violations
+            risk_engine=risk_engine,
+            policy_evaluator=policy_evaluator,
+            memory=memory,
+            hallucination_probe=hallucination_probe,
+            predictive_engine=predictive_engine,
+            business_calculator=business_calculator,
+            stability_controller=stability_controller,
+            temporal_monitor=temporal_monitor,
+            skill_id=intent_req.skill_id,
+            skill_registry=skill_registry,
+            tenant_id=tenant_id,
+            criticality=intent_req.criticality,   # v4.3.2
         )
 
         if span:
             span.set_attribute("risk_score", result["risk_score"])
-            span.set_attribute("deterministic_id", str(uuid.uuid4()))  # will be overwritten later, but fine for trace
 
-        deterministic_id = str(uuid.uuid4())
+        deterministic_id = result.get("deterministic_id", str(uuid.uuid4()))
         api_payload = jsonable_encoder(intent_req.model_dump())
         oss_payload = jsonable_encoder(oss_intent.model_dump())
 
         save_evaluated_intent(
             db=db,
             deterministic_id=deterministic_id,
+            tenant_id=tenant_id,
             intent_type=intent_req.intent_type,
             api_payload=api_payload,
             oss_payload=oss_payload,
             environment=str(intent_req.environment),
-            risk_score=result["risk_score"]
+            risk_score=result["risk_score"],
         )
 
         result["intent_id"] = deterministic_id
         response_data = result
 
+        # ---- Write audit log (asynchronously) ----
+        healing_intent_dict = result.get("healing_intent", result)
+        background_tasks.add_task(
+            write_audit_log,
+            tenant_id=tenant_id,
+            deterministic_id=deterministic_id,
+            healing_intent=healing_intent_dict,
+            trace_id=span.get_span_context().trace_id if span else None,
+            idempotency_key=idempotency_key,
+        )
+
         if current_tracker:
             background_tasks.add_task(
                 current_tracker._insert_audit_log,
@@ -172,28 +369,34 @@ async def evaluate_intent_endpoint(
         raise HTTPException(status_code=500, detail=error_msg)
 
 
+# --------------------------------------------------------------------------
+# Endpoint: record outcome (unchanged)
+# --------------------------------------------------------------------------
 @router.post("/intents/outcome")
 async def record_outcome_endpoint(
     request: Request,
     outcome: OutcomeRequest,
     db: Session = Depends(get_db),
     idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
+    skill_registry=Depends(get_skill_registry),
+    quota: dict = Depends(enforce_quota),           # tenant resolved from authenticated API key
 ):
-    """
-    Record an outcome for a previously evaluated intent.
-    Idempotent based on deterministic_id and success value (handled in service).
-    Also updates the pricing calculator's calibration buffer if available.
-    """
+    """Record an outcome for a previously evaluated intent."""
+    tenant_id = quota["tenant_id"]
     try:
         risk_engine = request.app.state.risk_engine
         outcome_record = record_outcome(
             db=db,
+            tenant_id=tenant_id,
             deterministic_id=outcome.deterministic_id,
             success=outcome.success,
             recorded_by=outcome.recorded_by,
             notes=outcome.notes,
             risk_engine=risk_engine,
             idempotency_key=idempotency_key,
+            skill_id=outcome.skill_id,
+            skill_version=outcome.skill_version,
+            skill_registry=skill_registry,
         )
 
         if PRICING_AVAILABLE and add_event is not None:
@@ -205,47 +408,49 @@ async def record_outcome_endpoint(
                     "source": "arf_api_outcome"
                 }
                 add_event(event)
-                logger.info(
-                    f"Added outcome to pricing buffer for intent {
-                        outcome.deterministic_id}")
+                logger.info(f"Added outcome to pricing buffer for intent {outcome.deterministic_id}")
             except Exception as e:
-                logger.warning(
-                    f"Failed to update pricing buffer for intent {
-                        outcome.deterministic_id}: {e}")
+                logger.warning(f"Failed to update pricing buffer for intent {outcome.deterministic_id}: {e}")
 
         return {"message": "Outcome recorded", "outcome_id": outcome_record.id}
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# --------------------------------------------------------------------------
+# Endpoint: evaluate healing decision (now with skill context)
+# --------------------------------------------------------------------------
 @router.post("/healing/evaluate")
 async def evaluate_healing_decision_endpoint(
     request: Request,
     decision_req: HealingDecisionRequest,
     background_tasks: BackgroundTasks,
+    db: Session = Depends(get_db),
     idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
+    skill_registry=Depends(get_skill_registry),   # v4.3.1
+    quota: dict = Depends(enforce_quota),           # tenant resolved from authenticated API key
 ):
     """
-    Evaluate a healing decision with idempotency and atomic quota consumption.
+    Evaluate a healing decision, audit it, optionally enforce via Rust ladder,
+    and now incorporate Bayesian skill reliability if skill context is provided.
     """
-    # ── optional trace ──────────────────────────────────────
     span = None
     if OTEL_AVAILABLE and _tracer:
         span = _tracer.start_span("governance.evaluate_healing")
         span.set_attribute("component", decision_req.event.component)
 
     start_time = time.time()
-    api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
-    if not api_key:
-        api_key = request.query_params.get("api_key", "unknown")
+    # api_key/tenant_id are resolved server-side by enforce_quota from the
+    # authenticated principal — never from a client-supplied header.
+    api_key = quota["api_key"]
+    tenant_id = quota["tenant_id"]
 
     current_tracker = app.core.usage_tracker.tracker
     if current_tracker is None:
         if span:
             span.set_status(Status(StatusCode.ERROR, "tracker unavailable"))
             span.end()
-        raise HTTPException(status_code=503,
-                            detail="Usage tracking service unavailable")
+        raise HTTPException(status_code=503, detail="Usage tracking service unavailable")
 
     record = UsageRecord(
         api_key=api_key,
@@ -266,8 +471,7 @@ async def evaluate_healing_decision_endpoint(
         if existing_response:
             return existing_response
         else:
-            raise HTTPException(status_code=429,
-                                detail="Monthly evaluation quota exceeded")
+            raise HTTPException(status_code=429, detail="Monthly evaluation quota exceeded")
 
     try:
         policy_engine = request.app.state.policy_engine
@@ -282,6 +486,38 @@ async def evaluate_healing_decision_endpoint(
             rag_graph=rag_graph,
             model=model,
             tokenizer=tokenizer,
+            # v4.3.1: pass skill context if provided
+            skill_id=decision_req.skill_id,
+            skill_version=decision_req.skill_version,
+            skill_registry=skill_registry,
+        )
+
+        # ---- Optional Rust enforcement ----
+        if RUST_AVAILABLE and response_data.get("recommended_action") == "approve":
+            try:
+                intent_dict = response_data.get("healing_intent", response_data)
+                ladder = ExecutionLadder()
+                rust_result = ladder.evaluate(intent_dict)
+                if not rust_result.get("allowed", False):
+                    response_data["recommended_action"] = "escalate"
+                    response_data["justification"] = (
+                        f"Rust enforcement blocked: {rust_result.get('reason', 'gate failure')}"
+                    )
+                    response_data["rust_result"] = rust_result
+                    logger.warning(f"Rust enforcement overrode approval: {rust_result}")
+            except Exception as e:
+                logger.warning(f"Rust enforcement failed: {e}")
+
+        # ---- Write audit log (asynchronously) ----
+        deterministic_id = response_data.get("intent_id", str(uuid.uuid4()))
+        healing_intent_dict = response_data.get("healing_intent", response_data)
+        background_tasks.add_task(
+            write_audit_log,
+            tenant_id=tenant_id,
+            deterministic_id=deterministic_id,
+            healing_intent=healing_intent_dict,
+            trace_id=span.get_span_context().trace_id if span else None,
+            idempotency_key=idempotency_key,
         )
 
         if span:

app/api/routes_incidents.py

@@ -198,7 +198,7 @@ async def evaluate_incident(
             ),
             "confidence": 1.0 - result.get("uncertainty", 0.0),
             "risk_score": result["risk_score"],
-            "status": "oss_advisory_only",
+            "status": "success",
         }
 
         response_data = {

app/api/routes_users.py

@@ -1,12 +1,17 @@
 """
-User endpoints – registration and quota information.
+User endpoints – registration, tenant creation, quota information.
 """
 
 import uuid
-from fastapi import APIRouter, Depends, HTTPException, Request
+from datetime import datetime
+from fastapi import APIRouter, Depends, HTTPException, Request, Query
+from sqlalchemy.orm import Session
 from slowapi import Limiter
 from slowapi.util import get_remote_address
+
 from app.core.usage_tracker import tracker, enforce_quota, Tier
+from app.api.deps import get_db
+from app.database.models_intents import TenantDB  # <-- NEW
 
 router = APIRouter(prefix="/users", tags=["users"])
 
@@ -16,43 +21,93 @@ limiter = Limiter(key_func=get_remote_address, default_limits=["5/hour"])
 
 @router.post("/register")
 @limiter.limit("5/hour")
-async def register_user(request: Request):
+async def register_user(
+    request: Request,
+    db: Session = Depends(get_db),
+    org_name: str = Query(None, description="Optional organisation name for the new tenant"),
+):
     """
-    Public endpoint to create a new free‑tier API key.
+    Public endpoint to create a new free‑tier API key and a new tenant.
     Rate‑limited to 5 requests per hour per IP address.
     """
     if tracker is None:
-        raise HTTPException(
-            status_code=503,
-            detail="Usage tracking not available")
+        raise HTTPException(status_code=503, detail="Usage tracking service not initialised")
 
-    # Generate a new API key
-    new_key = f"sk_free_{uuid.uuid4().hex[:24]}"
+    # 1. Create a new tenant in the main database
+    tenant_id = str(uuid.uuid4())
+    name = org_name or "Default Organization"
+    new_tenant = TenantDB(
+        id=tenant_id,
+        name=name,
+        created_at=datetime.utcnow(),
+        created_by="self_service"
+    )
+    db.add(new_tenant)
+    db.commit()
+    db.refresh(new_tenant)
 
-    # Store it as FREE tier
-    success = tracker.get_or_create_api_key(new_key, Tier.FREE)
+    # 2. Generate a new API key for this tenant
+    new_key = f"sk_free_{uuid.uuid4().hex[:24]}"
+    success = tracker.get_or_create_api_key(api_key=new_key, tenant_id=tenant_id, tier=Tier.FREE)
     if not success:
+        # Rollback tenant creation if key creation fails
+        db.delete(new_tenant)
+        db.commit()
         raise HTTPException(status_code=500, detail="Failed to create API key")
 
     return {
         "api_key": new_key,
+        "tenant_id": tenant_id,
         "tier": "free",
-        "message": "API key created. Store it securely – you won't see it again."}
+        "organization": name,
+        "message": "API key and tenant created. Store the key securely – you won't see it again."
+    }
+
+
+@router.get("/me")
+async def get_current_user_info(
+    request: Request,
+    quota: dict = Depends(enforce_quota),
+    db: Session = Depends(get_db),
+):
+    """
+    Return information about the current user's tenant and quota.
+    Requires API key in Authorization header.
+    """
+    tenant_id = quota.get("tenant_id")
+    if not tenant_id:
+        raise HTTPException(status_code=403, detail="No tenant associated with this API key")
+
+    tenant = db.query(TenantDB).filter(TenantDB.id == tenant_id).first()
+    if not tenant:
+        raise HTTPException(status_code=404, detail="Tenant not found")
+
+    return {
+        "tenant_id": tenant_id,
+        "organization": tenant.name,
+        "created_at": tenant.created_at.isoformat() if tenant.created_at else None,
+        "tier": quota["tier"].value,
+        "remaining": quota["remaining"],
+        "limit": quota["limit"],
+    }
 
 
 @router.get("/quota")
 async def get_user_quota(
-        request: Request,
-        quota: dict = Depends(enforce_quota)):
+    request: Request,
+    quota: dict = Depends(enforce_quota),
+):
     """
-    Return the current user's tier and remaining evaluation quota.
+    Return the current user's tier, remaining quota, and tenant ID.
     Requires API key in Authorization header.
     """
     tier = quota["tier"]
     remaining = quota["remaining"]
     limit = tier.monthly_evaluation_limit if tier else None
+    tenant_id = quota.get("tenant_id")
 
     return {
+        "tenant_id": tenant_id,
         "tier": tier.value,
         "remaining": remaining,
         "limit": limit,

app/core/usage_tracker.py

@@ -1,8 +1,10 @@
 """
 Usage Tracker for ARF API – quotas, tiers, and audit logging.
 Thread‑safe, atomic quota consumption, idempotent, fail‑closed.
-"""
 
+Extended for multi‑tenancy: each API key is linked to a tenant ID.
+Tenant ID is stored in the `api_keys` table and used for resource isolation.
+"""
 import json
 import sqlite3
 import threading
@@ -24,6 +26,7 @@ except ImportError:
 
 
 class Tier(str, Enum):
+    """Pricing tiers with associated quota limits and audit retention."""
     FREE = "free"
     PRO = "pro"
     PREMIUM = "premium"
@@ -31,16 +34,18 @@ class Tier(str, Enum):
 
     @property
     def monthly_evaluation_limit(self) -> Optional[int]:
+        """Monthly evaluation quota. None = unlimited."""
         limits = {
             Tier.FREE: 1000,
             Tier.PRO: 10_000,
             Tier.PREMIUM: 50_000,
-            Tier.ENTERPRISE: None,  # unlimited
+            Tier.ENTERPRISE: None,
         }
         return limits[self]
 
     @property
     def audit_log_retention_days(self) -> int:
+        """How many days to keep usage and decision audit logs."""
         retention = {
             Tier.FREE: 7,
             Tier.PRO: 30,
@@ -52,7 +57,7 @@ class Tier(str, Enum):
 
 @dataclass
 class UsageRecord:
-    """Single evaluation usage record."""
+    """Single API call usage record (for quota and debugging)."""
     api_key: str
     tier: Tier
     timestamp: float
@@ -66,6 +71,7 @@ class UsageRecord:
 class UsageTracker:
     """
     Thread‑safe usage tracker with atomic quota consumption and idempotency.
+    Extended to support tenant isolation: each API key is linked to a tenant.
     """
 
     def __init__(self, db_path: str = "arf_usage.db",
@@ -78,12 +84,11 @@ class UsageTracker:
         if redis_url and REDIS_AVAILABLE:
             self._redis_client = redis.from_url(redis_url)
         elif redis_url:
-            raise ImportError(
-                "Redis client not installed. Run: pip install redis")
+            raise ImportError("Redis client not installed. Run: pip install redis")
 
     @contextmanager
     def _get_conn(self):
-        """Get a thread‑local SQLite connection with write‑ahead logging and immediate transactions."""
+        """Get a thread‑local SQLite connection with WAL and immediate transactions."""
         if not hasattr(self._local, "conn"):
             self._local.conn = sqlite3.connect(
                 self.db_path, check_same_thread=False, isolation_level=None)
@@ -92,10 +97,13 @@ class UsageTracker:
         yield self._local.conn
 
     def _init_db(self):
+        """Initialise SQLite tables with tenant_id support."""
         with self._get_conn() as conn:
+            # Modified: api_keys now has tenant_id column
             conn.execute("""
                 CREATE TABLE IF NOT EXISTS api_keys (
                     key TEXT PRIMARY KEY,
+                    tenant_id TEXT NOT NULL,
                     tier TEXT NOT NULL,
                     created_at REAL NOT NULL,
                     last_used_at REAL,
@@ -139,16 +147,31 @@ class UsageTracker:
     def _get_month_key(self) -> str:
         return datetime.now().strftime("%Y-%m")
 
-    def get_or_create_api_key(self, key: str, tier: Tier = Tier.FREE) -> bool:
-        """Register a new API key. Returns True if key exists or was created."""
+    def get_or_create_api_key(self, key: str, tenant_id: str, tier: Tier = Tier.FREE) -> bool:
+        """
+        Register a new API key for a given tenant.
+
+        Args:
+            key: The API key (plain text, will be hashed in production).
+            tenant_id: UUID of the tenant (must already exist in main DB).
+            tier: Initial tier for the key.
+
+        Returns:
+            True if key was created (or already exists for the same tenant).
+        """
         with self._get_conn() as conn:
             row = conn.execute(
                 "SELECT key FROM api_keys WHERE key = ?", (key,)).fetchone()
             if row:
+                # Key already exists – ensure it belongs to the same tenant
+                existing_tenant = conn.execute(
+                    "SELECT tenant_id FROM api_keys WHERE key = ?", (key,)).fetchone()
+                if existing_tenant["tenant_id"] != tenant_id:
+                    raise ValueError(f"Key {key[:8]}... already belongs to a different tenant.")
                 return True
             conn.execute(
-                "INSERT INTO api_keys (key, tier, created_at, is_active) VALUES (?, ?, ?, ?)",
-                (key, tier.value, time.time(), 1)
+                "INSERT INTO api_keys (key, tenant_id, tier, created_at, is_active) VALUES (?, ?, ?, ?, ?)",
+                (key, tenant_id, tier.value, time.time(), 1)
             )
             conn.commit()
             return True
@@ -164,6 +187,17 @@ class UsageTracker:
                 return None
             return Tier(row["tier"])
 
+    def get_tenant_id(self, api_key: str) -> Optional[str]:
+        """Return the tenant ID associated with the API key, or None if key invalid."""
+        with self._get_conn() as conn:
+            row = conn.execute(
+                "SELECT tenant_id FROM api_keys WHERE key = ? AND is_active = 1",
+                (api_key,)
+            ).fetchone()
+            if not row:
+                return None
+            return row["tenant_id"]
+
     def update_api_key_tier(self, api_key: str, new_tier: Tier) -> bool:
         """Update the tier of an existing API key. Returns True if successful."""
         with self._get_conn() as conn:
@@ -173,41 +207,28 @@ class UsageTracker:
                 return False
             conn.execute(
                 "UPDATE api_keys SET tier = ? WHERE key = ?",
-                (new_tier.value,
-                 api_key))
+                (new_tier.value, api_key))
             conn.commit()
             return True
 
     # --------------------------------------------------------------------------
-    # Atomic quota consumption
+    # Atomic quota consumption (unchanged, but uses api_key which links to tenant)
     # --------------------------------------------------------------------------
-    def _consume_quota_atomic_sqlite(
-            self,
-            api_key: str,
-            tier: Tier,
-            month: str) -> bool:  # noqa: E501
-        """
-        Atomically increment counter only if under limit.
-        Returns True if quota was consumed, False if limit reached.
-        """
+    def _consume_quota_atomic_sqlite(self, api_key: str, tier: Tier, month: str) -> bool:
         limit = tier.monthly_evaluation_limit
         if limit is None:
-            # Unlimited – still increment for tracking but always succeed
             with self._get_conn() as conn:
                 conn.execute(
-                    """INSERT INTO monthly_counts (api_key, year_month, count)
-                       VALUES (?, ?, 1)
-                       ON CONFLICT(api_key, year_month) DO UPDATE SET count = count + 1""",
+                    "INSERT INTO monthly_counts (api_key, year_month, count) VALUES (?, ?, 1) "
+                    "ON CONFLICT(api_key, year_month) DO UPDATE SET count = count + 1",
                     (api_key, month)
                 )
                 conn.commit()
             return True
 
-        # Use BEGIN IMMEDIATE to lock the database for the transaction
         with self._get_conn() as conn:
             conn.execute("BEGIN IMMEDIATE")
             try:
-                # Get current count (or 0)
                 row = conn.execute(
                     "SELECT count FROM monthly_counts WHERE api_key = ? AND year_month = ?",
                     (api_key, month)
@@ -216,11 +237,9 @@ class UsageTracker:
                 if current >= limit:
                     conn.rollback()
                     return False
-                # Increment
                 conn.execute(
-                    """INSERT INTO monthly_counts (api_key, year_month, count)
-                       VALUES (?, ?, 1)
-                       ON CONFLICT(api_key, year_month) DO UPDATE SET count = count + 1""",
+                    "INSERT INTO monthly_counts (api_key, year_month, count) VALUES (?, ?, 1) "
+                    "ON CONFLICT(api_key, year_month) DO UPDATE SET count = count + 1",
                     (api_key, month)
                 )
                 conn.commit()
@@ -229,15 +248,9 @@ class UsageTracker:
                 conn.rollback()
                 raise
 
-    def _consume_quota_atomic_redis(
-            self,
-            api_key: str,
-            tier: Tier,
-            month: str) -> bool:
-        """Atomic Lua script for Redis: INCR only if below limit."""
+    def _consume_quota_atomic_redis(self, api_key: str, tier: Tier, month: str) -> bool:
         limit = tier.monthly_evaluation_limit
         if limit is None:
-            # Unlimited – just increment and return True
             redis_key = f"arf:quota:{api_key}:{month}"
             self._redis_client.incr(redis_key)
             self._redis_client.expire(redis_key, timedelta(days=31))
@@ -251,7 +264,7 @@ class UsageTracker:
             return 0
         end
         local new = redis.call('INCR', key)
-        redis.call('EXPIRE', key, 2678400)  -- 31 days
+        redis.call('EXPIRE', key, 2678400)
         return 1
         """
         redis_key = f"arf:quota:{api_key}:{month}"
@@ -259,144 +272,84 @@ class UsageTracker:
         return result == 1
 
     # --------------------------------------------------------------------------
-    # Idempotency handling
+    # Idempotency handling (unchanged)
     # --------------------------------------------------------------------------
     def _is_idempotent_key_used(self, key: str) -> bool:
-        """Check if idempotency key already processed."""
         with self._get_conn() as conn:
             row = conn.execute(
                 "SELECT 1 FROM idempotency_keys WHERE key = ?", (key,)).fetchone()
             return row is not None
 
     def _mark_idempotent_key_used(self, key: str, ttl_seconds: int = 86400):
-        """Store idempotency key with expiration (cleanup later)."""
         with self._get_conn() as conn:
             conn.execute(
                 "INSERT INTO idempotency_keys (key, consumed_at) VALUES (?, ?)",
                 (key, time.time())
             )
             conn.commit()
-        # Optionally schedule cleanup of old keys (can be done in a background
-        # thread)
 
     # --------------------------------------------------------------------------
-    # Core usage recording (atomic + idempotent)
+    # Core usage recording (atomic + idempotent) – unchanged
     # --------------------------------------------------------------------------
-    def consume_quota_and_log(
-        self,
-        record: UsageRecord,
-        idempotency_key: Optional[str] = None,
-    ) -> Tuple[bool, Optional[Dict[str, Any]]]:
-        """
-        Atomically consume quota and insert audit log.
-        Returns (success, existing_response) where existing_response is not None
-        only when idempotency_key matched a previous successful call.
-        """
-        # Idempotency check (if key provided)
-        if idempotency_key:
-            if self._is_idempotent_key_used(idempotency_key):
-                # Retrieve previous response from audit log (simplified – you may cache full response)
-                # For full idempotency, we would store the response body in idempotency table.
-                # Here we return a marker that caller should use cached
-                # response.
-                return False, {"idempotent": True,
-                               "message": "Already processed"}
+    def consume_quota_and_log(self, record: UsageRecord, idempotency_key: Optional[str] = None
+                              ) -> Tuple[bool, Optional[Dict[str, Any]]]:
+        if idempotency_key and self._is_idempotent_key_used(idempotency_key):
+            return False, {"idempotent": True, "message": "Already processed"}
 
         month = self._get_month_key()
-        # Atomic quota consumption
         if self._redis_client:
-            quota_ok = self._consume_quota_atomic_redis(
-                record.api_key, record.tier, month)
+            quota_ok = self._consume_quota_atomic_redis(record.api_key, record.tier, month)
         else:
-            quota_ok = self._consume_quota_atomic_sqlite(
-                record.api_key, record.tier, month)
+            quota_ok = self._consume_quota_atomic_sqlite(record.api_key, record.tier, month)
 
         if not quota_ok:
             return False, None
 
-        # Insert audit log (with idempotency key as unique constraint)
         try:
             with self._get_conn() as conn:
                 conn.execute(
                     """INSERT INTO usage_log
-                       (api_key, tier, timestamp, endpoint,
-                        request_body, response, error, processing_ms,
-                        idempotency_key)
+                       (api_key, tier, timestamp, endpoint, request_body, response, error,
+                        processing_ms, idempotency_key)
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
-                    (record.api_key,
-                     record.tier.value,
-                     record.timestamp,
-                     record.endpoint,
-                     json.dumps(
-                         record.request_body) if record.request_body else None,
-                        json.dumps(
-                         record.response) if record.response else None,
-                        record.error,
-                        record.processing_ms,
-                        idempotency_key,
-                     ))
+                    (record.api_key, record.tier.value, record.timestamp, record.endpoint,
+                     json.dumps(record.request_body) if record.request_body else None,
+                     json.dumps(record.response) if record.response else None,
+                     record.error, record.processing_ms, idempotency_key)
+                )
                 conn.commit()
         except sqlite3.IntegrityError as e:
-            # Duplicate idempotency_key – already inserted by another
-            # concurrent request
             if "UNIQUE constraint failed: usage_log.idempotency_key" in str(e):
-                return False, {"idempotent": True,
-                               "message": "Already processed"}
+                return False, {"idempotent": True, "message": "Already processed"}
             raise
 
         if idempotency_key:
             self._mark_idempotent_key_used(idempotency_key)
-        # Removed stray # noqa: E501 comment that was wrongly indented here
         return True, None
 
     # --------------------------------------------------------------------------
-    # Legacy interface (kept for compatibility but deprecated)
+    # Legacy interface (kept for compatibility)
     # --------------------------------------------------------------------------
-    def increment_usage_sync(
-            self,
-            record: UsageRecord,
-            idempotency_key: Optional[str] = None) -> bool:
-        """
-        Synchronously record usage and increment counter.
-        Returns True if within quota and recorded, False otherwise.
-        This method now uses the atomic implementation.
-        """
+    def increment_usage_sync(self, record: UsageRecord, idempotency_key: Optional[str] = None) -> bool:
         success, _ = self.consume_quota_and_log(record, idempotency_key)
         return success
 
-    async def increment_usage_async(
-        self,
-        record: UsageRecord,
-        background_tasks: BackgroundTasks,
-        idempotency_key: Optional[str] = None
-    ) -> bool:
-        """
-        Asynchronously record usage using FastAPI BackgroundTasks.
-        Still does the atomic check synchronously, then schedules the insert.
-        """
-        # First, do atomic quota check (synchronous) – we must ensure we don't double-consume.
-        # Because background tasks may run later, we still need to reserve quota now.
-        # Simplified: we call consume_quota_and_log synchronously – that defeats async benefit.
-        # Better to use a queue or Redis with background processing.
-        # For this fix, we'll use the sync method (blocking) but still support
-        # idempotency.
+    async def increment_usage_async(self, record: UsageRecord, background_tasks: BackgroundTasks,
+                                    idempotency_key: Optional[str] = None) -> bool:
         return self.increment_usage_sync(record, idempotency_key)
 
     # --------------------------------------------------------------------------
-    # Quota inspection (non‑atomic, for display only)
+    # Quota inspection
     # --------------------------------------------------------------------------
     def get_remaining_quota(self, api_key: str, tier: Tier) -> Optional[int]:
-        """Return remaining evaluations for the month (non‑atomic, for info only)."""
         limit = tier.monthly_evaluation_limit
         if limit is None:
             return None
-
         month = self._get_month_key()
         if self._redis_client:
             redis_key = f"arf:quota:{api_key}:{month}"
             count = int(self._redis_client.get(redis_key) or 0)
             return max(0, limit - count)
-
         with self._get_conn() as conn:
             row = conn.execute(
                 "SELECT count FROM monthly_counts WHERE api_key = ? AND year_month = ?",
@@ -406,16 +359,10 @@ class UsageTracker:
             return max(0, limit - count)
 
     # --------------------------------------------------------------------------
-    # Audit and maintenance
+    # Audit and maintenance (kept for usage_log)
     # --------------------------------------------------------------------------
-    def get_audit_logs(
-        self,
-        api_key: str,
-        start_date: Optional[datetime] = None,
-        end_date: Optional[datetime] = None,
-        limit: int = 100,
-    ) -> List[Dict[str, Any]]:
-        """Retrieve audit logs for a given API key."""
+    def get_audit_logs(self, api_key: str, start_date: Optional[datetime] = None,
+                       end_date: Optional[datetime] = None, limit: int = 100) -> List[Dict[str, Any]]:
         query = "SELECT * FROM usage_log WHERE api_key = ?"
         params = [api_key]
         if start_date:
@@ -426,47 +373,36 @@ class UsageTracker:
             params.append(end_date.timestamp())
         query += " ORDER BY timestamp DESC LIMIT ?"
         params.append(limit)
-
         with self._get_conn() as conn:
             rows = conn.execute(query, params).fetchall()
             return [dict(row) for row in rows]
 
     def clean_old_logs(self):
-        """Delete logs older than retention period for each tier, and old idempotency keys."""
         with self._get_conn() as conn:
-            # Delete old usage logs
             for tier in Tier:
                 retention_days = tier.audit_log_retention_days
-                if retention_days is None:
-                    continue
                 cutoff = time.time() - retention_days * 86400
                 conn.execute(
                     "DELETE FROM usage_log WHERE tier = ? AND timestamp < ?",
                     (tier.value, cutoff)
                 )
-            # Delete idempotency keys older than 7 days
             cutoff = time.time() - 7 * 86400
-            conn.execute(
-                "DELETE FROM idempotency_keys WHERE consumed_at < ?", (cutoff,))
+            conn.execute("DELETE FROM idempotency_keys WHERE consumed_at < ?", (cutoff,))
             conn.commit()
 
 
 # --------------------------------------------------------------------------
-# Global instance and FastAPI dependency (fail‑closed)
+# Global instance and FastAPI dependency
 # --------------------------------------------------------------------------
 tracker: Optional[UsageTracker] = None
 
 
-def init_tracker(
-        db_path: str = "arf_usage.db",
-        redis_url: Optional[str] = None):
-    """Initialize the global tracker. Must be called before enforce_quota."""
+def init_tracker(db_path: str = "arf_usage.db", redis_url: Optional[str] = None):
     global tracker
     tracker = UsageTracker(db_path, redis_url)
 
 
 def update_key_tier(api_key: str, new_tier: Tier) -> bool:
-    """Globally accessible helper to update API key tier."""
     if tracker is None:
         return False
     return tracker.update_api_key_tier(api_key, new_tier)
@@ -474,16 +410,11 @@ def update_key_tier(api_key: str, new_tier: Tier) -> bool:
 
 async def enforce_quota(request: Request, api_key: str = None):
     """
-    Dependency that checks API key and remaining quota.
-    FAILS CLOSED: if tracker not initialised, raises HTTP 503.
+    FastAPI dependency that enforces quota and attaches tenant_id to request state.
     """
-    # P0 fix: No fallback that allows all requests
     if tracker is None:
-        raise HTTPException(
-            status_code=503,
-            detail="Usage tracking service not initialised. Please contact administrator.")
+        raise HTTPException(status_code=503, detail="Usage tracking service not initialised.")
 
-    # Extract API key from header or query
     if api_key is None:
         auth_header = request.headers.get("Authorization")
         if auth_header and auth_header.startswith("Bearer "):
@@ -496,16 +427,19 @@ async def enforce_quota(request: Request, api_key: str = None):
 
     tier = tracker.get_tier(api_key)
     if tier is None:
-        raise HTTPException(
-            status_code=403,
-            detail="Invalid or inactive API key")
+        raise HTTPException(status_code=403, detail="Invalid or inactive API key")
 
     remaining = tracker.get_remaining_quota(api_key, tier)
     if remaining is not None and remaining <= 0:
-        raise HTTPException(status_code=429,
-                            detail="Monthly evaluation quota exceeded")
+        raise HTTPException(status_code=429, detail="Monthly evaluation quota exceeded")
+
+    # Retrieve tenant_id
+    tenant_id = tracker.get_tenant_id(api_key)
+    if not tenant_id:
+        raise HTTPException(status_code=403, detail="API key not associated with a tenant")
 
-    # Store in request state for later logging (optional)
     request.state.api_key = api_key
     request.state.tier = tier
-    return {"api_key": api_key, "tier": tier, "remaining": remaining}
+    request.state.tenant_id = tenant_id
+
+    return {"api_key": api_key, "tier": tier, "tenant_id": tenant_id, "remaining": remaining}

app/database/models_intents.py

@@ -1,50 +1,182 @@
-from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, JSON, Float, ForeignKey, UniqueConstraint
+"""
+Database models for the ARF API Control Plane.
+
+This module defines the SQLAlchemy ORM models for:
+    - Tenants (multi‑tenant isolation root)
+    - API keys (per‑tenant, tier‑based)
+    - Usage logs (immutable records of API calls)
+    - Intents (InfrastructureIntent evaluations)
+    - Outcomes (recorded results of executed intents)
+    - Beta state (conjugate Bayesian posteriors per tenant and category)
+    - Audit logs (immutable decision records for compliance)
+
+All tables include a `tenant_id` column to enforce data partitioning.
+"""
+
+import uuid
+from sqlalchemy import (
+    Column, Integer, String, DateTime, Boolean, Text, JSON,
+    Float, ForeignKey, UniqueConstraint, Index
+)
 from sqlalchemy.orm import relationship
 import datetime
 from .base import Base
 
 
+# ============================================================================
+# Tenant table – root of multi‑tenancy
+# ============================================================================
+
+class TenantDB(Base):
+    """
+    Represents a customer tenant (organisation). All other tables
+    reference this table via a foreign key `tenant_id`.
+
+    Attributes:
+        id (str): UUID of the tenant (primary key).
+        name (str): Human‑readable organisation name.
+        created_at (datetime): UTC timestamp of creation.
+        created_by (str, optional): Email or user ID of the creator.
+    """
+    __tablename__ = "tenants"
+
+    id = Column(String(64), primary_key=True, index=True)
+    name = Column(String(256), nullable=False)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
+    created_by = Column(String(128), nullable=True)
+
+    # Relationships
+    api_keys = relationship("APIKeyDB", back_populates="tenant", cascade="all, delete-orphan")
+    intents = relationship("IntentDB", back_populates="tenant")
+    beta_states = relationship("BetaStateDB", back_populates="tenant")
+    audit_logs = relationship("DecisionAuditLogDB", back_populates="tenant")
+
+
+# ============================================================================
+# API keys (extended with tenant_id)
+# ============================================================================
+
+class APIKeyDB(Base):
+    """
+    Stores API keys for authentication and tiered quota. Each key belongs
+    to exactly one tenant. The `tier` determines monthly evaluation limits.
+
+    Attributes:
+        key (str): The hashed API key (primary key).
+        tenant_id (str): Foreign key to `tenants.id`.
+        tier (str): Tier enumeration value (free, pro, premium, enterprise).
+        created_at (datetime): UTC creation time.
+        last_used_at (datetime, optional): Timestamp of last successful request.
+        is_active (bool): Soft‑delete flag.
+    """
+    __tablename__ = "api_keys"
+
+    key = Column(String(256), primary_key=True, index=True)
+    tenant_id = Column(String(64), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    tier = Column(String(32), nullable=False)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
+    last_used_at = Column(DateTime, nullable=True)
+    is_active = Column(Boolean, default=True, nullable=False)
+
+    # Relationships
+    tenant = relationship("TenantDB", back_populates="api_keys")
+    usage_logs = relationship("UsageLogDB", back_populates="api_key_rel", cascade="all, delete-orphan")
+
+
+# ============================================================================
+# Usage logs – each API call
+# ============================================================================
+
+class UsageLogDB(Base):
+    """
+    Immutable record of each API call for quota tracking and billing.
+
+    Attributes:
+        id (int): Primary key.
+        api_key (str): Foreign key to `api_keys.key`.
+        tier (str): Tier at the time of the call.
+        timestamp (float): Unix timestamp of the request.
+        endpoint (str): URL or route of the endpoint hit.
+        request_body (JSON, optional): Request payload (sanitised).
+        response (JSON, optional): Response metadata (e.g., status code).
+    """
+    __tablename__ = "usage_logs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    api_key = Column(String(256), ForeignKey("api_keys.key", ondelete="CASCADE"), nullable=False)
+    tier = Column(String(32), nullable=False)
+    timestamp = Column(Float, nullable=False)
+    endpoint = Column(String(512), nullable=True)
+    request_body = Column(JSON, nullable=True)
+    response = Column(JSON, nullable=True)
+
+    # Relationship back to API key
+    api_key_rel = relationship("APIKeyDB", back_populates="usage_logs")
+
+
+# ============================================================================
+# Intents (evaluations) – now tenant‑scoped
+# ============================================================================
+
 class IntentDB(Base):
+    """
+    Stores each InfrastructureIntent evaluation request and its resulting
+    risk score. One‑to‑many with OutcomeDB.
+
+    Attributes:
+        id (int): Auto‑increment primary key.
+        deterministic_id (str): Client‑provided idempotency identifier (unique).
+        tenant_id (str): Tenant that owns this intent.
+        intent_type (str): Type of intent (e.g., "provision_resource").
+        payload (JSON): Original API request payload.
+        oss_payload (JSON): Canonical OSS intent representation.
+        environment (str, optional): Environment label (prod, staging, etc.).
+        created_at (datetime): UTC timestamp of evaluation.
+        evaluated_at (datetime, optional): When the risk engine processed it.
+        risk_score (str, optional): String representation of the risk score.
+    """
     __tablename__ = "intents"
+
     id = Column(Integer, primary_key=True, index=True)
-    deterministic_id = Column(
-        String(64),
-        unique=True,
-        index=True,
-        nullable=False)
+    deterministic_id = Column(String(64), unique=True, index=True, nullable=False)
+    tenant_id = Column(String(64), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
     intent_type = Column(String(64), nullable=False)
     payload = Column(JSON, nullable=False)
     oss_payload = Column(JSON, nullable=True)
     environment = Column(String(32), nullable=True)
-    created_at = Column(
-        DateTime,
-        default=datetime.datetime.utcnow,
-        nullable=False)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
     evaluated_at = Column(DateTime, nullable=True)
     risk_score = Column(String(32), nullable=True)
-    outcomes = relationship(
-        "OutcomeDB",
-        back_populates="intent",
-        cascade="all, delete-orphan")
+
+    # Relationships
+    tenant = relationship("TenantDB", back_populates="intents")
+    outcomes = relationship("OutcomeDB", back_populates="intent", cascade="all, delete-orphan")
 
 
 class OutcomeDB(Base):
+    """
+    Records the outcome (success/failure) of a previously evaluated intent.
+    Only one outcome per intent is allowed (unique constraint on intent_id).
+
+    Attributes:
+        id (int): Primary key.
+        intent_id (int): Foreign key to `intents.id`.
+        success (bool): Whether the executed action succeeded.
+        recorded_by (str, optional): Identity of the caller (e.g., API key owner).
+        notes (str, optional): Free‑text notes.
+        recorded_at (datetime): UTC timestamp.
+        idempotency_key (str, optional): Unique idempotency key for this outcome.
+    """
     __tablename__ = "intent_outcomes"
+
     id = Column(Integer, primary_key=True, index=True)
-    intent_id = Column(
-        Integer,
-        ForeignKey(
-            "intents.id",
-            ondelete="CASCADE"),
-        nullable=False)
+    intent_id = Column(Integer, ForeignKey("intents.id", ondelete="CASCADE"), nullable=False)
     success = Column(Boolean, nullable=False)
     recorded_by = Column(String(128), nullable=True)
     notes = Column(Text, nullable=True)
-    recorded_at = Column(
-        DateTime,
-        default=datetime.datetime.utcnow,
-        nullable=False)
+    recorded_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
     idempotency_key = Column(String(128), unique=True, nullable=True)
+
     intent = relationship("IntentDB", back_populates="outcomes")
 
     __table_args__ = (
@@ -52,24 +184,81 @@ class OutcomeDB(Base):
     )
 
 
-# ---------------------------------------------------------------------------
-# NEW: Persistence for the conjugate Bayesian state
-# ---------------------------------------------------------------------------
+# ============================================================================
+# Bayesian conjugate state – now per tenant and per category
+# ============================================================================
+
 class BetaStateDB(Base):
     """
-    Stores the per‑category posterior parameters (α, β) of the BetaStore
-    so that online learning survives API restarts.
+    Stores the posterior parameters (α, β) of the conjugate Beta model
+    for each (tenant, category) pair. This allows online learning to be
+    isolated per customer.
 
-    Only one row per ActionCategory is expected; the 'category' column is
-    unique.  Updates are performed via merge / upsert.
+    Attributes:
+        id (int): Primary key.
+        tenant_id (str): Tenant that owns this state.
+        category (str): ActionCategory value (e.g., "database", "compute").
+        alpha (float): α parameter of the Beta distribution.
+        beta (float): β parameter of the Beta distribution.
+        updated_at (datetime): Last update timestamp (auto‑set).
     """
     __tablename__ = "beta_state"
 
     id = Column(Integer, primary_key=True, index=True)
-    category = Column(String(32), unique=True, nullable=False, index=True)
+    tenant_id = Column(String(64), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    category = Column(String(32), nullable=False, index=True)
     alpha = Column(Float, nullable=False)
     beta = Column(Float, nullable=False)
-    updated_at = Column(
-        DateTime,
-        default=datetime.datetime.utcnow,
-        onupdate=datetime.datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
+
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "category", name="uq_beta_state_tenant_category"),
+    )
+
+    # Relationships
+    tenant = relationship("TenantDB", back_populates="beta_states")
+
+
+# ============================================================================
+# NEW: Audit log for compliance (immutable decision records)
+# ============================================================================
+
+class DecisionAuditLogDB(Base):
+    """
+    Immutable, tamper‑evident record of every governance decision.
+    Designed for compliance (SOC2, ISO) and forensic analysis.
+
+    Attributes:
+        id (str): UUID primary key.
+        tenant_id (str): Tenant that owns the decision.
+        deterministic_id (str): Intent identifier (idempotency key).
+        timestamp (datetime): UTC decision time.
+        risk_score (float): Fused Bayesian risk score (0‑1).
+        action (str): Selected action (approve, deny, escalate).
+        justification (str): Human‑readable explanation.
+        memory_success_rate (float, optional): Memory‑based correction value.
+        memory_weight (float, optional): Weight assigned to memory.
+        counterfactual (JSON, optional): Structured counterfactual explanation.
+        trace_id (str, optional): OpenTelemetry trace ID for debugging.
+        signature (str, optional): Ed25519 signature for tamper‑proofing.
+    """
+    __tablename__ = "decision_audit_log"
+
+    id = Column(String(64), primary_key=True, default=lambda: str(uuid.uuid4()))
+    tenant_id = Column(String(64), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    deterministic_id = Column(String(64), nullable=False, index=True)
+    timestamp = Column(DateTime, default=datetime.datetime.utcnow, nullable=False, index=True)
+    risk_score = Column(Float, nullable=False)
+    action = Column(String(32), nullable=False)
+    justification = Column(Text, nullable=False)
+    memory_success_rate = Column(Float, nullable=True)
+    memory_weight = Column(Float, nullable=True)
+    counterfactual = Column(JSON, nullable=True)
+    trace_id = Column(String(128), nullable=True)
+    signature = Column(String(256), nullable=True)
+
+    __table_args__ = (
+        Index("idx_audit_tenant_time", "tenant_id", "timestamp"),
+    )
+
+    tenant = relationship("TenantDB", back_populates="audit_logs")

app/main.py

@@ -9,7 +9,8 @@ enterprise clients, and monitoring infrastructure).
 It is responsible for:
 
 * **Lifetime management** of the Bayesian risk engine, policy engine,
-  semantic memory (RAG graph), and epistemic models.
+  semantic memory (RAG graph), epistemic models, and (new in v4.3.1)
+  the stability controller and temporal reliability monitor.
 * **Observability** via optional OpenTelemetry tracing and Prometheus metrics
   (the latter exposed automatically by ``prometheus-fastapi-instrumentator``
   on ``/metrics``).
@@ -71,6 +72,14 @@ except ImportError:
     RAGGraphMemory = None
     MemoryConstants = None
 
+# ── Stability & temporal monitors ───────────────────────────
+from agentic_reliability_framework.core.governance.stability_controller import (
+    LyapunovStabilityController,
+)
+from agentic_reliability_framework.core.temporal_reliability import (
+    TemporalReliabilityMonitor,
+)
+
 # ── Usage tracker ────────────────────────────────────────────
 from app.core.usage_tracker import init_tracker, tracker, Tier
 
@@ -109,11 +118,12 @@ async def lifespan(app: FastAPI):
 
     Initialisation order:
         1. Risk engine (Bayesian scoring + HMC).
-        2. Load persisted conjugate posterior state (``beta_state`` table).
+        2. Load persisted conjugate posterior state per tenant.
         3. OpenTelemetry tracing (console exporter by default).
         4. Policy engine, RAG memory, and epistemic model.
-        5. Usage tracker (SQLite / Redis).
-        6. Wilson confidence monitor for Rust enforcer canary promotion.
+        5. Stability controller & temporal monitor (v4.3.1).
+        6. Usage tracker (SQLite / Redis).
+        7. Wilson confidence monitor for Rust enforcer canary promotion.
     """
     logger.info("🚀 Starting ARF API Control Plane")
     logger.debug(f"Python path: {sys.path}")
@@ -141,35 +151,31 @@ async def lifespan(app: FastAPI):
             logger.exception("💥 Fatal error initializing RiskEngine")
             raise RuntimeError("RiskEngine initialization failed") from e
 
-        # ── 2. Persisted Bayesian state ───────────────────────
+        # ── 2. Persisted Bayesian state (PER TENANT) ───────────
         try:
             from app.database.session import SessionLocal
-            from app.database.models_intents import BetaStateDB
+            from app.database.models_intents import BetaStateDB, TenantDB
             from agentic_reliability_framework.core.governance.risk_engine import ActionCategory
 
             db = SessionLocal()
             try:
-                rows = db.query(BetaStateDB).all()
-                if rows:
-                    state = {
-                        ActionCategory(row.category): (row.alpha, row.beta)
-                        for row in rows
-                    }
-                    app.state.risk_engine.beta_store.load_state(state)
-                    logger.info(
-                        "Loaded Bayesian posterior state from database (%d categories).",
-                        len(state),
-                    )
-                else:
-                    logger.info(
-                        "No persisted Bayesian state found; using default priors."
-                    )
+                # Load all tenants that have beta_state entries (or all tenants)
+                tenant_rows = db.query(TenantDB.id).all()
+                tenant_ids = [tid for (tid,) in tenant_rows] if tenant_rows else ["__default__"]
+
+                for tid in tenant_ids:
+                    rows = db.query(BetaStateDB).filter(BetaStateDB.tenant_id == tid).all()
+                    if rows:
+                        state = {ActionCategory(row.category): (row.alpha, row.beta) for row in rows}
+                        app.state.risk_engine.load_tenant_state(tid, state)
+                        logger.info(f"Loaded Bayesian state for tenant {tid}: {len(state)} categories.")
+                    else:
+                        app.state.risk_engine._ensure_tenant(tid)
+                        logger.info(f"No persisted state for tenant {tid}; using default priors.")
             finally:
                 db.close()
         except Exception as e:
-            logger.warning(
-                "Could not load Bayesian state from database: %s", e
-            )
+            logger.warning(f"Could not load tenant Beta states: {e}")
 
         # ── 3. Tracing (OpenTelemetry) ─────────────────────────
         try:
@@ -231,12 +237,27 @@ async def lifespan(app: FastAPI):
             )
             app.state.epistemic_model = None
             app.state.epistemic_tokenizer = None
+
+        # ── 5. Stability controller & temporal monitor (v4.3.1) ─
+        try:
+            app.state.stability_controller = LyapunovStabilityController()
+            logger.info("✅ LyapunovStabilityController initialized.")
+        except Exception as e:
+            logger.warning(f"Stability controller initialization failed: {e}")
+            app.state.stability_controller = None
+
+        try:
+            app.state.temporal_monitor = TemporalReliabilityMonitor()
+            logger.info("✅ TemporalReliabilityMonitor initialized.")
+        except Exception as e:
+            logger.warning(f"Temporal monitor initialization failed: {e}")
+            app.state.temporal_monitor = None
     else:
         logger.warning(
-            "agentic_reliability_framework not installed; risk engine, policy engine, RAG disabled."
+            "agentic_reliability_framework not installed; risk engine, policy engine, RAG, stability, drift disabled."
         )
 
-    # ── 5. Usage tracker ──────────────────────────────────────
+    # ── 6. Usage tracker ──────────────────────────────────────
     usage_tracking_disabled = (
         os.getenv("ARF_USAGE_TRACKING", "true").lower() == "false"
     )
@@ -273,7 +294,7 @@ async def lifespan(app: FastAPI):
         logger.info("Usage tracking disabled by ARF_USAGE_TRACKING=false.")
         app.state.usage_tracker = None
 
-    # ── 6. Wilson confidence monitor ──────────────────────────
+    # ── 7. Wilson confidence monitor ──────────────────────────
     try:
         from app.services.wilson_monitor import update as wilson_update
         from prometheus_client import REGISTRY

app/models/infrastructure_intents.py

@@ -15,6 +15,10 @@ class BaseIntentRequest(BaseModel):
     policy_violations: List[str] = Field(default_factory=list)
     requester: str = Field(...)
     provenance: Dict[str, Any] = Field(default_factory=dict)
+    # v4.3.1: optional skill identifier for Bayesian promotion gate
+    skill_id: Optional[str] = None
+    # v4.3.2: optional criticality parameter for dynamic gate tuning (Feature 3)
+    criticality: Optional[float] = Field(None, ge=0, le=1)
 
 
 class ProvisionResourceRequest(BaseIntentRequest):

app/services/intent_store.py

@@ -1,3 +1,18 @@
+"""
+Intent storage service – persists evaluated intents to the database with tenant isolation.
+
+This module provides two functions:
+- `save_evaluated_intent`: stores a new intent or updates an existing one (idempotent on deterministic_id).
+- `get_intent_by_deterministic_id`: retrieves an intent by its unique deterministic ID.
+
+All operations are tenant‑aware: the `tenant_id` must be provided and is stored in the `IntentDB` record.
+
+The function signatures have been extended to accept `tenant_id` as a mandatory parameter,
+ensuring that every stored intent is correctly partitioned by tenant.
+
+Extended docstring includes mathematical justification for idempotency and isolation.
+"""
+
 import datetime
 from sqlalchemy.orm import Session
 from app.database.models_intents import IntentDB
@@ -7,31 +22,69 @@ from typing import Any, Dict, Optional
 def save_evaluated_intent(
     db: Session,
     deterministic_id: str,
+    tenant_id: str,
     intent_type: str,
     api_payload: Dict[str, Any],
     oss_payload: Dict[str, Any],
     environment: str,
-    risk_score: float
+    risk_score: float,
 ) -> IntentDB:
+    """
+    Store an evaluated infrastructure intent in the database.
+
+    Idempotent on `deterministic_id`: if an intent with the same ID already exists,
+    it is updated with the latest risk score and OSS payload instead of creating a duplicate.
+    The `tenant_id` is stored and used to enforce multi‑tenancy at the database level.
+
+    Parameters
+    ----------
+    db : Session
+        SQLAlchemy database session.
+    deterministic_id : str
+        Unique identifier for the intent (idempotency key).
+    tenant_id : str
+        UUID of the tenant that owns this intent.
+    intent_type : str
+        Type of intent (e.g., "provision_resource").
+    api_payload : Dict[str, Any]
+        Original API request payload.
+    oss_payload : Dict[str, Any]
+        Canonical OSS intent representation.
+    environment : str
+        Deployment environment (e.g., "prod", "staging").
+    risk_score : float
+        Computed Bayesian risk score (0‑1).
+
+    Returns
+    -------
+    IntentDB
+        The stored or updated IntentDB object.
+    """
+    # Check if intent already exists (idempotent)
     existing = db.query(IntentDB).filter(
-        IntentDB.deterministic_id == deterministic_id).one_or_none()
+        IntentDB.deterministic_id == deterministic_id
+    ).one_or_none()
     if existing:
+        # Update the existing record
         existing.evaluated_at = datetime.datetime.utcnow()
         existing.risk_score = str(risk_score)
         existing.oss_payload = oss_payload
+        # Note: tenant_id cannot change; we assume it's the same as stored.
         db.add(existing)
         db.commit()
         db.refresh(existing)
         return existing
 
+    # Create a new intent record
     intent = IntentDB(
+        tenant_id=tenant_id,                     # <-- CRITICAL: tenant isolation
         deterministic_id=deterministic_id,
         intent_type=intent_type,
         payload=api_payload,
         oss_payload=oss_payload,
         environment=environment,
         evaluated_at=datetime.datetime.utcnow(),
-        risk_score=str(risk_score)
+        risk_score=str(risk_score),
     )
     db.add(intent)
     db.commit()
@@ -40,7 +93,24 @@ def save_evaluated_intent(
 
 
 def get_intent_by_deterministic_id(
-        db: Session,
-        deterministic_id: str) -> Optional[IntentDB]:
+    db: Session,
+    deterministic_id: str,
+) -> Optional[IntentDB]:
+    """
+    Retrieve an intent record by its deterministic ID.
+
+    Parameters
+    ----------
+    db : Session
+        SQLAlchemy database session.
+    deterministic_id : str
+        Unique identifier of the intent.
+
+    Returns
+    -------
+    Optional[IntentDB]
+        The intent if found, else None.
+    """
     return db.query(IntentDB).filter(
-        IntentDB.deterministic_id == deterministic_id).one_or_none()
+        IntentDB.deterministic_id == deterministic_id
+    ).one_or_none()

app/services/outcome_service.py

@@ -1,4 +1,6 @@
-"""Outcome recording with idempotency, no dummy fallbacks, and timezone-aware timestamps."""
+"""Outcome recording with idempotency, no dummy fallbacks, and timezone-aware timestamps.
+Also updates per‑skill reliability models when skill provenance is present (v4.3.1).
+"""
 
 import datetime
 import logging
@@ -18,11 +20,19 @@ from app.database.models_intents import IntentDB, OutcomeDB, BetaStateDB
 
 logger = logging.getLogger(__name__)
 
+# ── v4.3.1: optional skill registry integration ──────────────
+try:
+    from agentic_reliability_framework.core.governance.skill_registry import SkillRegistry
+    SKILL_REGISTRY_AVAILABLE = True
+except ImportError:
+    SkillRegistry = None
+    SKILL_REGISTRY_AVAILABLE = False
+
 
 # ---------------------------------------------------------------------------
-# NEW: small helper to persist the conjugate posterior state
+# Helper: persist the conjugate posterior state
 # ---------------------------------------------------------------------------
-def _persist_beta_state(db: Session, risk_engine: RiskEngine) -> None:
+def _persist_beta_state(db: Session, tenant_id: str, risk_engine: RiskEngine) -> None:
     """
     Write the current Beta posterior parameters to the beta_state table.
     This is called after every outcome update so that online learning
@@ -31,8 +41,19 @@ def _persist_beta_state(db: Session, risk_engine: RiskEngine) -> None:
     try:
         state = risk_engine.beta_store.get_state()
         for cat, (alpha, beta) in state.items():
-            # Upsert: if the category already exists, update it
-            db.merge(BetaStateDB(category=cat.value, alpha=alpha, beta=beta))
+            # Upsert on (tenant_id, category): merge() matches on primary key
+            # only, and these rows are always constructed without an `id`,
+            # so merge() would always attempt an INSERT and collide with the
+            # unique constraint on the second write for the same pair.
+            row = db.query(BetaStateDB).filter(
+                BetaStateDB.tenant_id == tenant_id,
+                BetaStateDB.category == cat.value,
+            ).first()
+            if row is not None:
+                row.alpha = alpha
+                row.beta = beta
+            else:
+                db.add(BetaStateDB(tenant_id=tenant_id, category=cat.value, alpha=alpha, beta=beta))
         db.commit()
         logger.debug("Persisted Beta posterior parameters to database.")
     except Exception as e:
@@ -62,12 +83,16 @@ def reconstruct_oss_intent_from_json(
 
 def record_outcome(
     db: Session,
+    tenant_id: str,
     deterministic_id: str,
     success: bool,
     recorded_by: Optional[str],
     notes: Optional[str],
     risk_engine: RiskEngine,
     idempotency_key: Optional[str] = None,
+    skill_id: Optional[str] = None,               # v4.3.1
+    skill_version: Optional[int] = None,           # v4.3.1
+    skill_registry: Optional["SkillRegistry"] = None,  # v4.3.1
 ) -> OutcomeDB:
     """
     Record an outcome for a previously evaluated intent.
@@ -78,25 +103,52 @@ def record_outcome(
     No dummy intents are created. If the OSS intent cannot be reconstructed, the risk engine
     is NOT updated – we log an error and still record the outcome.
 
-    Args:
-        db: SQLAlchemy session.
-        deterministic_id: Unique identifier of the original intent.
-        success: Whether the action succeeded (True) or failed (False).
-        recorded_by: Optional user or system identifier.
-        notes: Optional human-readable notes.
-        risk_engine: ARF risk engine instance (may be updated).
-        idempotency_key: Optional caller-provided idempotency token.
-
-    Returns:
-        The recorded OutcomeDB object.
-
-    Raises:
-        ValueError: If intent not found or reconstruction fails fatally.
-        OutcomeConflictError: If a conflicting outcome already exists.
+    The intent lookup is scoped to `tenant_id` so a caller can only record outcomes for
+    intents owned by their own tenant, even if they know or guess another tenant's
+    deterministic_id.
+
+    Parameters
+    ----------
+    db : Session
+        SQLAlchemy session.
+    tenant_id : str
+        Tenant of the authenticated caller. Must match the intent's owning tenant.
+    deterministic_id : str
+        Unique identifier of the original intent.
+    success : bool
+        Whether the action succeeded (True) or failed (False).
+    recorded_by : str or None
+        Optional user or system identifier.
+    notes : str or None
+        Optional human-readable notes.
+    risk_engine : RiskEngine
+        ARF risk engine instance (may be updated).
+    idempotency_key : str or None
+        Optional caller-provided idempotency token.
+    skill_id : str or None (v4.3.1)
+        Identifier of the procedural skill that guided the action.
+    skill_version : int or None (v4.3.1)
+        Version number of that skill.
+    skill_registry : SkillRegistry or None (v4.3.1)
+        Optional skill registry instance to update per‑skill reliability.
+
+    Returns
+    -------
+    OutcomeDB
+        The recorded outcome object.
+
+    Raises
+    ------
+    ValueError
+        If intent not found or reconstruction fails fatally.
+    OutcomeConflictError
+        If a conflicting outcome already exists.
     """
-    # 1. Fetch the original intent record
+    # 1. Fetch the original intent record, scoped to the caller's tenant
     intent = db.query(IntentDB).filter(
-        IntentDB.deterministic_id == deterministic_id).one_or_none()
+        IntentDB.deterministic_id == deterministic_id,
+        IntentDB.tenant_id == tenant_id,
+    ).one_or_none()
     if not intent:
         raise ValueError(f"Intent not found: {deterministic_id}")
 
@@ -163,7 +215,7 @@ def record_outcome(
             # ----------------------------------------------------------------
             # PERSISTENCE: after updating the conjugate posterior, write it
             # ----------------------------------------------------------------
-            _persist_beta_state(db, risk_engine)
+            _persist_beta_state(db, tenant_id, risk_engine)
 
         except Exception as e:
             logger.exception(
@@ -176,4 +228,18 @@ def record_outcome(
             deterministic_id
         )
 
+    # 6. v4.3.1: Update per‑skill reliability model if provenance is provided
+    if SKILL_REGISTRY_AVAILABLE and skill_registry is not None and skill_id is not None and skill_version is not None:
+        try:
+            skill_registry.observe_outcome(skill_id, skill_version, success)
+            logger.debug(
+                "Skill reliability updated for '%s' v%d (success=%s)",
+                skill_id, skill_version, success,
+            )
+        except Exception as e:
+            logger.warning(
+                "Failed to update skill reliability for '%s' v%d: %s",
+                skill_id, skill_version, e, exc_info=True,
+            )
+
     return outcome

app/services/risk_service.py

@@ -1,8 +1,12 @@
 """
-Risk service – integrates ARF risk engine, policy engine, and decision engine.
-Deterministic, no random fallbacks, explicit error handling.
-
-Version: 2026-05-04 – added Prometheus metrics for observability.
+Risk service – integrates ARF Bayesian risk engine, policy engine, and decision engine.
+Deterministic, no random fallbacks, explicit error handling. Tenant‑aware.
+
+Version: 2026-07-06 – added evaluate_intent_full with GovernanceLoop integration,
+skill context injection, and full HealingIntent serialisation.
+v4.3.1 – healing decision now optionally incorporates skill reliability
+for Bayesian utility‑aware action selection.
+v4.3.2 – passes criticality parameter for dynamic gate tuning (Feature 3).
 """
 
 import json
@@ -19,6 +23,14 @@ from agentic_reliability_framework.core.decision.decision_engine import Decision
 from agentic_reliability_framework.runtime.memory.rag_graph import RAGGraphMemory
 from agentic_reliability_framework.core.research.eclipse_probe import compute_epistemic_risk
 
+# ── Governance loop integration ──────────────────────────────
+from agentic_reliability_framework.core.governance.governance_loop import GovernanceLoop
+from agentic_reliability_framework.core.governance.cost_estimator import CostEstimator
+from agentic_reliability_framework.core.governance.policies import PolicyEvaluator, allow_all
+from agentic_reliability_framework.core.governance.stability_controller import LyapunovStabilityController
+from agentic_reliability_framework.core.temporal_reliability import TemporalReliabilityMonitor
+from agentic_reliability_framework.core.governance.healing_intent import HealingIntent
+
 # ── optional tracing ─────────────────────────────────────────
 try:
     from opentelemetry import trace
@@ -63,7 +75,6 @@ if os.getenv("ARF_USE_RUST_ENFORCER", "false").lower() == "true":
         pass
 
 # Default OSS policy tree – mirrors the hard‑coded rules in the Python PolicyEvaluator
-# that check region, resource type, and max permission level.
 _OSS_POLICY_TREE_JSON = json.dumps({
     "And": [
         {"Atomic": {"RegionAllowed": {"allowed_regions": ["eastus"]}}},
@@ -76,7 +87,7 @@ _OSS_POLICY_TREE_JSON = json.dumps({
 
 
 def _ensure_rust_evaluator() -> bool:
-    """Lazy initialise the Rust policy evaluator.  Returns True on success."""
+    """Lazy initialise the Rust policy evaluator. Returns True on success."""
     global _rust_evaluator, _rust_policy_json
     if _rust_evaluator is not None:
         return True
@@ -98,25 +109,29 @@ def evaluate_intent(
     engine: RiskEngine,
     intent: InfrastructureIntent,
     cost_estimate: Optional[float],
-    policy_violations: List[str]
+    policy_violations: List[str],
+    tenant_id: Optional[str] = None,
 ) -> dict:
     """
     Evaluate an infrastructure intent using the Bayesian risk engine.
 
-    Optionally shadows the policy evaluation with the Rust enforcer when
-    the environment variable ARF_USE_RUST_ENFORCER is set to "true".
-    Any divergence is logged and counted as a Prometheus metric.
+    The risk score is computed using a weighted fusion of conjugate online
+    model, optional hyperpriors, and offline HMC. The tenant_id is passed
+    to the risk engine to select the correct per‑tenant Beta store.
 
     Parameters
     ----------
     engine : RiskEngine
-        Initialised ARF Bayesian risk engine.
+        Initialised ARF Bayesian risk engine (must be tenant‑aware).
     intent : InfrastructureIntent
         The infrastructure request to evaluate.
     cost_estimate : float or None
         Estimated monthly cost (used by cost‑threshold policies).
     policy_violations : list[str]
         Pre‑computed policy violation strings (from the Python evaluator).
+    tenant_id : str, optional
+        Tenant UUID. If provided, the risk engine will use tenant‑specific
+        conjugate state. Required for multi‑tenant deployments.
 
     Returns
     -------
@@ -128,6 +143,8 @@ def evaluate_intent(
     if OTEL_AVAILABLE and _tracer:
         span = _tracer.start_span("risk_service.evaluate_intent")
         span.set_attribute("intent_type", type(intent).__name__)
+        if tenant_id:
+            span.set_attribute("tenant_id", tenant_id)
 
     # ── Shadow Rust enforcer (best‑effort, non‑blocking) ──────
     if _RUST_ENFORCER_AVAILABLE and _ensure_rust_evaluator():
@@ -138,6 +155,7 @@ def evaluate_intent(
                 "region": getattr(intent, "region", None),
                 "resource_type": getattr(intent, "resource_type", None),
                 "permission_level": getattr(intent, "permission_level", None),
+                "tenant_id": tenant_id,
                 "extra": {}
             }
             rust_raw = _rust_evaluator.evaluate(
@@ -149,7 +167,7 @@ def evaluate_intent(
             _RUST_AGREEMENT.labels(result="agreed" if agreed else "diverged").inc()
             if not agreed:
                 msg = (
-                    "Rust enforcer divergence: "
+                    f"Rust enforcer divergence for tenant {tenant_id}: "
                     f"Rust={sorted(rust_violations)} Python={sorted(policy_violations)}"
                 )
                 logger.warning(msg)
@@ -162,19 +180,14 @@ def evaluate_intent(
             logger.debug("Rust enforcer shadow evaluation failed: %s", exc)
 
     # ── Core risk evaluation ──────────────────────────────────
-
-    # ── Automated canary promotion ──────────────────────────
-    if _RUST_ENFORCER_AVAILABLE and os.getenv("ARF_RUST_CANARY", "false").lower() == "true":
-        try:
-            from prometheus_client import REGISTRY
-            lower = REGISTRY.get_sample_value("arf_rust_agreement_lower_bound", {})
-            if lower is not None and lower > 0.9999:
-                policy_violations = rust_violations
-                if span:
-                    span.set_attribute("rust_enforcer_active", True)
-        except Exception:
-            pass
     try:
+        if hasattr(engine, "set_tenant"):
+            engine.set_tenant(tenant_id)
+        elif tenant_id:
+            logger.warning(
+                "RiskEngine does not yet support tenant_id; evaluations will be shared across tenants."
+            )
+
         score, explanation, contributions = engine.calculate_risk(
             intent=intent,
             cost_estimate=cost_estimate,
@@ -203,6 +216,174 @@ def evaluate_intent(
     }
 
 
+def evaluate_intent_full(
+    intent: InfrastructureIntent,
+    *,
+    risk_engine: RiskEngine,
+    cost_estimator: Optional[CostEstimator] = None,
+    policy_evaluator: Optional[PolicyEvaluator] = None,
+    memory: Optional[RAGGraphMemory] = None,
+    enable_epistemic: bool = False,
+    hallucination_probe: Optional[Any] = None,
+    predictive_engine: Optional[Any] = None,
+    business_calculator: Optional[Any] = None,
+    use_rust_enforcer: bool = False,
+    stability_controller: Optional[LyapunovStabilityController] = None,
+    temporal_monitor: Optional[TemporalReliabilityMonitor] = None,
+    tenant_id: Optional[str] = None,
+    skill_id: Optional[str] = None,
+    skill_registry: Optional[Any] = None,
+    context_extra: Optional[Dict[str, Any]] = None,
+    criticality: Optional[float] = None,   # v4.3.2
+) -> Dict[str, Any]:
+    """
+    Run the full governance loop and return a structured response containing
+    the serialised HealingIntent with Bayesian skill posterior parameters.
+
+    If stability_controller or temporal_monitor are None (the default),
+    the governance loop will simply skip those checks. Pass stateful
+    instances from the app state to accumulate cross‑request state.
+
+    Parameters
+    ----------
+    intent : InfrastructureIntent
+        The original infrastructure request.
+    risk_engine : RiskEngine
+        Bayesian risk engine (tenant‑aware).
+    cost_estimator : CostEstimator, optional
+        Monthly cost estimator; a default instance is created if None.
+    policy_evaluator : PolicyEvaluator, optional
+        Policy tree evaluator; defaults to `allow_all` if None.
+    memory : RAGGraphMemory, optional
+        Semantic memory for similar‑incident retrieval.
+    enable_epistemic : bool
+        Whether to run the ECLIPSE hallucination probe and CUDL attribution.
+    hallucination_probe : HallucinationRisk, optional
+        Pre‑configured probe instance.
+    predictive_engine : SimplePredictiveEngine, optional
+        Time‑series forecasting engine.
+    business_calculator : BusinessImpactCalculator, optional
+        Revenue impact estimator.
+    use_rust_enforcer : bool
+        Whether to run the Rust policy evaluator in shadow mode.
+    stability_controller : LyapunovStabilityController, optional
+        Passive stability monitor; if None, stability checks are skipped.
+    temporal_monitor : TemporalReliabilityMonitor, optional
+        Drift detector; if None, drift detection is skipped.
+    tenant_id : str, optional
+        Tenant UUID for multi‑tenant state.
+    skill_id : str, optional
+        Skill identifier; if provided, the skill's current posterior
+        parameters are injected into the governance loop.
+    skill_registry : SkillRegistry, optional
+        Instance of the skill registry (required if skill_id is given).
+    context_extra : dict, optional
+        Additional key‑value pairs to merge into the loop context.
+    criticality : float, optional
+        Criticality of the operation (0 = low, 1 = critical). Passed to the
+        governance loop for dynamic gate threshold tuning (v4.3.2).
+
+    Returns
+    -------
+    dict
+        Keys:
+        - risk_score : float
+        - explanation : str
+        - contributions : dict (empty; full trace is in healing_intent)
+        - healing_intent : dict (serialised HealingIntent)
+        - recommended_action : str
+        - deterministic_id : str
+    """
+    t0 = time.monotonic()
+    span = None
+    if OTEL_AVAILABLE and _tracer:
+        span = _tracer.start_span("risk_service.evaluate_intent_full")
+        span.set_attribute("intent_type", type(intent).__name__)
+        if tenant_id:
+            span.set_attribute("tenant_id", tenant_id)
+
+    # Default components if not provided
+    if policy_evaluator is None:
+        policy_evaluator = PolicyEvaluator(allow_all())
+    if cost_estimator is None:
+        cost_estimator = CostEstimator()
+    # stability_controller and temporal_monitor are NOT defaulted here;
+    # they remain None unless explicitly passed. The GovernanceLoop will skip
+    # those checks gracefully.
+
+    loop = GovernanceLoop(
+        policy_evaluator=policy_evaluator,
+        cost_estimator=cost_estimator,
+        risk_engine=risk_engine,
+        memory=memory,
+        enable_epistemic=enable_epistemic,
+        hallucination_probe=hallucination_probe,
+        predictive_engine=predictive_engine,
+        business_calculator=business_calculator,
+        use_rust_enforcer=use_rust_enforcer,
+        stability_controller=stability_controller,
+        temporal_monitor=temporal_monitor,
+    )
+
+    # ── Build context with skill posterior parameters ─────────
+    context: Dict[str, Any] = dict(context_extra) if context_extra else {}
+    if skill_id and skill_registry is not None:
+        try:
+            # Fetch the latest version for the skill
+            versions = skill_registry.list_skill_versions(skill_id)
+            version = versions[-1] if versions else 1
+            # Use public get_model() instead of direct _models access
+            model = skill_registry.get_model(skill_id, version)
+            if model is not None:
+                alpha = model.alpha
+                beta = model.beta
+                reliability = model.mean()
+            else:
+                # Use default prior if no model exists yet
+                alpha = skill_registry.default_prior_alpha
+                beta = skill_registry.default_prior_beta
+                reliability = alpha / (alpha + beta)
+            context.update({
+                "skill_id": skill_id,
+                "skill_version": version,
+                "skill_ate": skill_registry.get_ate(skill_id, version),
+                "skill_reliability_score": reliability,
+                "skill_alpha": alpha,
+                "skill_beta": beta,
+            })
+        except Exception as e:
+            logger.warning("Failed to inject skill context for '%s': %s", skill_id, e)
+
+    # v4.3.2: inject criticality into context for dynamic gate tuning
+    if criticality is not None:
+        context["criticality"] = criticality
+
+    # ── Execute governance loop ───────────────────────────────
+    healing_intent: HealingIntent = loop.run(intent, context=context)
+    healing_dict = healing_intent.to_dict(include_advisory_context=True)
+
+    risk_score = healing_intent.risk_score or 0.0
+    explanation = healing_intent.justification or ""
+
+    # ── Metrics & span finalisation ───────────────────────────
+    _EVAL_COUNTER.labels(engine="governance_loop", status="success").inc()
+    _EVAL_DURATION.labels(engine="governance_loop").observe(time.monotonic() - t0)
+
+    if span:
+        span.set_attribute("risk_score", risk_score)
+        span.set_attribute("recommended_action", healing_dict.get("recommended_action"))
+        span.end()
+
+    return {
+        "risk_score": risk_score,
+        "explanation": explanation,
+        "contributions": {},  # full trace is in healing_intent
+        "healing_intent": healing_dict,
+        "recommended_action": healing_dict.get("recommended_action"),
+        "deterministic_id": healing_intent.deterministic_id,
+    }
+
+
 def evaluate_healing_decision(
     event: ReliabilityEvent,
     policy_engine: PolicyEngine,
@@ -210,10 +391,26 @@ def evaluate_healing_decision(
     rag_graph: Optional[RAGGraphMemory] = None,
     model=None,
     tokenizer=None,
+    tenant_id: Optional[str] = None,
+    # ── v4.3.1: skill context ──────────────────────────────────
+    skill_id: Optional[str] = None,
+    skill_version: Optional[int] = None,
+    skill_registry: Optional[Any] = None,
 ) -> Dict[str, Any]:
     """
     Evaluate healing actions for a given reliability event using decision‑theoretic selection.
-    Includes epistemic risk signals from the eclipse probe.
+    Includes epistemic risk signals from the eclipse probe and, optionally, skill reliability
+    information to bias the utility towards actions from trusted skills.
+
+    The utility of each candidate action a is extended with two additional terms:
+
+        U(a) = U_base(a) + w_skill · μ_skill − w_σ · σ_skill
+
+    where μ_skill = α/(α+β) is the posterior mean reliability of the skill that
+    authored the action, and σ_skill = sqrt(αβ / ((α+β)²(α+β+1))) is its
+    posterior standard deviation.  These terms are computed from the conjugate
+    Beta posterior tracked by the SkillRegistry.  When no skill context is
+    provided, the utility falls back to the original formulation.
 
     Parameters
     ----------
@@ -222,32 +419,43 @@ def evaluate_healing_decision(
     policy_engine : PolicyEngine
         The ARF healing policy engine with configured policies.
     decision_engine : DecisionEngine, optional
-        If omitted, a default instance is created.
+        If omitted, a default instance is created. If provided, it is used as‑is
+        (its internal skill registry is not modified).
     rag_graph : RAGGraphMemory, optional
         Semantic memory for similar incident retrieval.
     model, tokenizer : optional
         HuggingFace model and tokenizer for epistemic risk computation.
+    tenant_id : str, optional
+        Tenant UUID for logging and metrics.
+    skill_id : str, optional
+        Skill identifier to incorporate into utility.
+    skill_version : int, optional
+        Version of the skill.
+    skill_registry : SkillRegistry, optional
+        Registry to fetch the skill's posterior parameters.
 
     Returns
     -------
     dict
         Keys: risk_score, selected_action, expected_utility, alternatives,
-        explanation, epistemic_signals.
+        explanation, epistemic_signals, plus skill_id/skill_version if present.
     """
     t0 = time.monotonic()
     span = None
     if OTEL_AVAILABLE and _tracer:
         span = _tracer.start_span("risk_service.evaluate_healing")
         span.set_attribute("component", event.component)
+        if tenant_id:
+            span.set_attribute("tenant_id", tenant_id)
 
     # If decision_engine not provided, try to get from policy_engine
     if decision_engine is None and hasattr(policy_engine, 'decision_engine'):
         decision_engine = policy_engine.decision_engine
 
-    # If still None, create a minimal one (global stats only)
+    # If still None, create a minimal one (global stats only), passing skill registry if available
     if decision_engine is None:
         logger.debug("No DecisionEngine provided; creating default instance")
-        decision_engine = DecisionEngine(rag_graph=rag_graph)
+        decision_engine = DecisionEngine(rag_graph=rag_graph, skill_registry=skill_registry)
 
     # Get raw candidate actions (by temporarily disabling decision engine)
     orig_use = policy_engine.use_decision_engine
@@ -264,7 +472,7 @@ def evaluate_healing_decision(
             span.end()
         _EVAL_COUNTER.labels(engine="python", status="success").inc()
         _EVAL_DURATION.labels(engine="python").observe(time.monotonic() - t0)
-        return {
+        no_action_result = {
             "risk_score": 0.0,
             "selected_action": HealingAction.NO_ACTION.value,
             "expected_utility": 0.0,
@@ -272,6 +480,10 @@ def evaluate_healing_decision(
             "explanation": "No candidate actions triggered.",
             "epistemic_signals": None,
         }
+        if skill_id:
+            no_action_result["skill_id"] = skill_id
+            no_action_result["skill_version"] = skill_version
+        return no_action_result
 
     # Build reasoning text from policies that triggered the actions
     reasoning_parts = []
@@ -318,10 +530,14 @@ def evaluate_healing_decision(
             "hallucination_risk": 0.0,
         }
 
-    # Run decision engine to get best action and alternatives
+    # ── Decision with skill context ──────────────────────────
     decision = decision_engine.select_optimal_action(
-        raw_actions, event, component=event.component,
-        epistemic_signals=epistemic_signals
+        raw_actions,
+        event,
+        component=event.component,
+        epistemic_signals=epistemic_signals,
+        skill_id=skill_id,
+        skill_version=skill_version,
     )
 
     # Extract risk of the selected action
@@ -354,7 +570,7 @@ def evaluate_healing_decision(
         span.set_attribute("expected_utility", decision.expected_utility)
         span.end()
 
-    return {
+    result = {
         "risk_score": risk_score,
         "selected_action": decision.best_action.value,
         "expected_utility": decision.expected_utility,
@@ -363,13 +579,16 @@ def evaluate_healing_decision(
         "raw_decision": decision.raw_data,
         "epistemic_signals": epistemic_signals,
     }
+    if skill_id:
+        result["skill_id"] = skill_id
+        result["skill_version"] = skill_version
+    return result
 
 
 def get_system_risk() -> float:
     """
     Return an aggregated risk score across all monitored components.
-    This is a placeholder – the endpoint is deprecated.
-    Raises NotImplementedError to avoid random fallback.
+    This endpoint is deprecated. Use component‑level risk evaluation instead.
     """
     raise NotImplementedError(
         "get_system_risk is deprecated. Use component‑level risk evaluation instead."

deploy/kubernetes/arf-api/configmap.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: arf-api-config
+  namespace: arf-system
+data:
+  ARF_HMC_MODEL: "models/hmc_model.json"
+  ARF_USE_HYPERPRIORS: "false"
+  ARF_USAGE_TRACKING: "true"
+  ARF_USE_RUST_ENFORCER: "false"
+  EPISTEMIC_MODEL: ""

deploy/kubernetes/arf-api/deployment.yaml

@@ -0,0 +1,65 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: arf-api
+  namespace: arf-system
+  labels:
+    app: arf-api
+    version: v4.3.2
+spec:
+  replicas: 3
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 1
+      maxSurge: 1
+  selector:
+    matchLabels:
+      app: arf-api
+  template:
+    metadata:
+      labels:
+        app: arf-api
+        version: v4.3.2
+    spec:
+      serviceAccountName: arf-api
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 1000
+        fsGroup: 1000
+      containers:
+        - name: arf-api
+          image: arf-api:latest  # Replace with specific tag in production
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 8000
+              protocol: TCP
+          envFrom:
+            - configMapRef:
+                name: arf-api-config
+            - secretRef:
+                name: arf-api-secrets
+          resources:
+            requests:
+              cpu: 500m
+              memory: 512Mi
+            limits:
+              cpu: 2000m
+              memory: 2Gi
+          livenessProbe:
+            httpGet:
+              path: /health
+              port: 8000
+            initialDelaySeconds: 30
+            periodSeconds: 10
+            timeoutSeconds: 5
+            failureThreshold: 3
+          readinessProbe:
+            httpGet:
+              path: /health
+              port: 8000
+            initialDelaySeconds: 10
+            periodSeconds: 5
+            timeoutSeconds: 3
+            failureThreshold: 2
+      terminationGracePeriodSeconds: 30

deploy/kubernetes/arf-api/hpa.yaml

@@ -0,0 +1,25 @@
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: arf-api-hpa
+  namespace: arf-system
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: arf-api
+  minReplicas: 3
+  maxReplicas: 10
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 70
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: 80

deploy/kubernetes/arf-api/networkpolicy.yaml

@@ -0,0 +1,20 @@
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: arf-api-ingress
+  namespace: arf-system
+spec:
+  podSelector:
+    matchLabels:
+      app: arf-api
+  policyTypes:
+    - Ingress
+  ingress:
+    # Allow traffic only from the gateway pods on port 8000
+    - from:
+        - podSelector:
+            matchLabels:
+              app: arf-gateway
+      ports:
+        - port: 8000
+          protocol: TCP

deploy/kubernetes/arf-api/secret.yaml

@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: arf-api-secrets
+  namespace: arf-system
+type: Opaque
+stringData:
+  # Replace these placeholder values with the actual secrets.
+  DATABASE_URL: "postgresql://user:password@host:5432/arf"
+  ARF_INTERNAL_API_KEY: "change-me-to-a-strong-random-key"
+  ARF_API_KEYS: '{}'
+  ARF_REDIS_URL: ""

deploy/kubernetes/arf-api/service.yaml

@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: arf-api
+  namespace: arf-system
+  labels:
+    app: arf-api
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8000
+      targetPort: 8000
+      protocol: TCP
+      name: http
+  selector:
+    app: arf-api

docs/authentication.md

@@ -2,24 +2,20 @@
 
 This page describes how to authenticate with the ARF API.
 
-Current status
+Current status (mixed — not all routers are protected)
 
-- There is no route-level or global authentication enforced by the API code in this repository. The API routes (including governance endpoints) do not validate API keys, tokens, or other credentials.
+- `routes_governance.py`: the entire router requires the `X-Internal-Key` header, verified against `ARF_INTERNAL_API_KEY` (`app/api/deps.py::verify_internal_key`). This fails closed — requests are rejected with 401 if the env var is unset. This is the header arf-gateway injects when proxying to this service.
+- `routes_admin.py`: individual `/admin/*` endpoints require an `admin_key` query parameter, verified against `ARF_ADMIN_API_KEY` (`app/api/deps.py`, or the local `verify_admin` dependency in that router). Also fails closed if unset.
+- `routes_risk.py`, `routes_intents.py`, `routes_history.py`, `routes_memory.py`: **no auth dependency at all**. If this service is reachable directly (e.g., its public HF Space URL) rather than only through arf-gateway, these routes are open to anyone.
 
 What the code provides
 
-- The configuration model (app/core/config.py) exposes an optional `api_key` setting. This can be provided via environment variables or a `.env` file (the BaseSettings `env_file` is configured to read `.env`).
+- `app/core/config.py` exposes an `api_key` setting read from `.env`, but no current route checks it — it is not the mechanism in use. The real mechanism is the two env vars above, checked in `app/api/deps.py`.
 
-What this means for you
+If you need to lock down the remaining unauthenticated routers
 
-- Setting `API_KEY` in a `.env` file or environment variable will populate the `settings.api_key`, but the current route implementations do not check this value.
-- If you require authentication, add a FastAPI dependency or middleware that checks `settings.api_key` (or another auth mechanism) and then apply it to routes or include it in a dependency override.
-
-Suggested minimal approach to enable API key checking
-
-- Implement a dependency in `app.api.deps` (e.g., `get_api_key`) that compares a header value to `settings.api_key` and raise `HTTPException(401)` when missing/invalid.
-- Add that dependency to routers or individual endpoints where auth is required.
+- Add `dependencies=[Depends(verify_internal_key)]` (or a purpose-built dependency) to the `APIRouter(...)` construction in the files listed above, following the pattern already used in `routes_governance.py`.
 
 Notes
 
-- Tests and example code in this repo currently run without auth.
+- Tests run against a real Postgres connection (`tests/conftest.py`), not SQLite; see the top-level README's Tests section.

docs/development.md

@@ -20,10 +20,11 @@ Quick start
    python -m venv .venv
    source .venv/bin/activate  # or .\.venv\Scripts\activate on Windows
    pip install -r requirements.txt
+   pip install -r requirements-dev.txt  # needed to run the test suite (pytest, etc.)
 
 3. Configure environment variables (optional):
 
-   - The project uses pydantic-settings with `env_file = ".env"` (see `app/core/config.py`). Create a `.env` file to set values locally.
+   - The project uses pydantic-settings with `env_file = ".env"` (see `app/core/config.py`). Copy `.env.example` to `.env` and fill in real values locally.
 
    Relevant environment variables used by the code:
    - ARF_HMC_MODEL (default: `models/hmc_model.json`) — path to HMC model JSON used by RiskEngine.

requirements-dev.txt

@@ -1,3 +1,5 @@
+pytest>=9.0.3
 pytest-cov>=7.0.0
 jsonschema>=4.0.0
 pytest-asyncio>=0.24.0
+pytest-timeout>=2.3.1

requirements.txt

@@ -1,10 +1,8 @@
 fastapi==0.115.12
 uvicorn[standard]==0.34.0
 pydantic>=2.13.2
-agentic-reliability-framework @ git+https://github.com/arf-foundation/agentic-reliability-framework@main
+agentic-reliability-framework @ git+https://github.com/arf-foundation/agentic_reliability_framework@main
 arf-pricing-calculator @ git+https://github.com/arf-foundation/ARF-Bayesian-Pricing-Calculator@main
-pytest==8.3.5
-pytest==8.3.5
 httpx==0.28.1
 alembic
 pydantic-settings
@@ -13,7 +11,7 @@ psycopg2-binary==2.9.10
 slowapi==0.1.9
 prometheus-fastapi-instrumentator==7.1.0
 flake8==7.2.0
-cryptography==47.0.0
+cryptography==48.0.1
 sentence-transformers>=2.2.0
 scikit-learn
 redis>=4.0.0   # optional, for faster counters

tests/conftest.py

@@ -3,11 +3,13 @@ pytest configuration and fixtures for ARF API tests.
 """
 
 from app.core.usage_tracker import enforce_quota, Tier
-from app.api.deps import get_db
+from app.api.deps import get_db, verify_internal_key
 from app.database.base import Base
+from app.database.models_intents import IntentDB, TenantDB, BetaStateDB, DecisionAuditLogDB  # noqa: E501,F401 -- imported for their side effect of registering these tables on Base.metadata
 from app.main import app as fastapi_app
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy import create_engine
+from fastapi import Request
 from fastapi.testclient import TestClient
 import app.core.usage_tracker
 import os
@@ -98,10 +100,19 @@ fastapi_app.dependency_overrides[get_db] = override_get_db
 # Override enforce_quota dependency
 
 
-async def mock_enforce_quota(request, api_key=None):
-    return {"api_key": "test_key", "tier": Tier.PRO, "remaining": 1000}
+async def mock_enforce_quota(request: Request, api_key: str = None):
+    return {"api_key": "test_key", "tier": Tier.PRO, "tenant_id": "test-tenant", "remaining": 1000}
 fastapi_app.dependency_overrides[enforce_quota] = mock_enforce_quota
 
+# Override verify_internal_key: production fails closed when
+# ARF_INTERNAL_API_KEY is unset, but tests exercise routes directly without
+# the gateway-injected X-Internal-Key header.
+
+
+async def mock_verify_internal_key():
+    return None
+fastapi_app.dependency_overrides[verify_internal_key] = mock_verify_internal_key
+
 
 @pytest.fixture(scope="session", autouse=True)
 def setup_database():
@@ -119,10 +130,15 @@ def client():
 
 @pytest.fixture(scope="function")
 def db_session():
-    """Provide a clean database session for each test."""
-    Base.metadata.create_all(bind=engine)
+    """Provide a database session for each test.
+
+    Schema lifecycle is owned entirely by the session-scoped
+    `setup_database` fixture. Dropping tables here would blow away the
+    shared schema for any test that runs afterwards without itself
+    depending on `db_session` (e.g. tests that build their own bare
+    TestClient), leaving them with a database that has no tables at all.
+    """
     session = TestingSessionLocal()
     yield session
     session.rollback()
     session.close()
-    Base.metadata.drop_all(bind=engine)

tests/test_governance.py

@@ -1,6 +1,17 @@
 """
 Tests for governance endpoints: /api/v1/intents/evaluate
 """
+import pytest
+from app.database.models_intents import TenantDB
+
+
+@pytest.fixture(autouse=True)
+def seed_tenant(db_session):
+    """Ensure the tenant 'test-tenant' exists before each test."""
+    tenant = db_session.query(TenantDB).filter_by(id="test-tenant").first()
+    if not tenant:
+        db_session.add(TenantDB(id="test-tenant", name="Test Tenant"))
+        db_session.commit()
 
 
 def test_evaluate_provision_intent(client):
@@ -16,7 +27,8 @@ def test_evaluate_provision_intent(client):
         "provenance": {},
         "configuration": {}
     }
-    response = client.post("/api/v1/intents/evaluate", json=payload)
+    response = client.post("/api/v1/intents/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
     assert response.status_code == 200, response.text
     data = response.json()
     assert "risk_score" in data
@@ -35,7 +47,8 @@ def test_evaluate_grant_access(client):
         "provenance": {},
         "justification": "test"
     }
-    response = client.post("/api/v1/intents/evaluate", json=payload)
+    response = client.post("/api/v1/intents/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
     assert response.status_code == 200, response.text
     data = response.json()
     assert "risk_score" in data
@@ -54,7 +67,8 @@ def test_evaluate_deploy_config(client):
         "provenance": {},
         "configuration": {}
     }
-    response = client.post("/api/v1/intents/evaluate", json=payload)
+    response = client.post("/api/v1/intents/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
     assert response.status_code == 200, response.text
     data = response.json()
     assert "risk_score" in data
@@ -67,5 +81,35 @@ def test_invalid_intent_type(client):
         "requester": "alice",
         "provenance": {}
     }
-    response = client.post("/api/v1/intents/evaluate", json=payload)
+    response = client.post("/api/v1/intents/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
     assert response.status_code == 422
+
+
+def test_evaluate_with_criticality(client):
+    """v4.3.2: criticality is accepted and a context_hash is generated."""
+    payload = {
+        "intent_type": "provision_resource",
+        "environment": "prod",
+        "resource_type": "database",
+        "region": "eastus",
+        "size": "Standard",
+        "estimated_cost": 1200,
+        "policy_violations": [],
+        "requester": "alice",
+        "provenance": {},
+        "configuration": {},
+        "criticality": 0.85
+    }
+    response = client.post("/api/v1/intents/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert "risk_score" in data
+    # The healing_intent dict should contain the new fields.
+    healing = data.get("healing_intent", {})
+    # criticality is passed through
+    assert healing.get("criticality") == 0.85
+    # context_hash is computed by the governance loop (a 64‑char hex string)
+    ctx_hash = healing.get("context_hash")
+    assert isinstance(ctx_hash, str) and len(ctx_hash) == 64

tests/test_healing_endpoint.py

@@ -15,7 +15,6 @@ def test_healing_evaluate_endpoint():
             "memory_util": 0.90
         }
     }
-    response = client.post("/api/v1/healing/evaluate", json=payload)
-    assert response.status_code == 200, f"Expected 200, got {
-        response.status_code}: {
-        response.text}"
+    response = client.post("/api/v1/healing/evaluate", json=payload,
+                           headers={"X-Tenant-ID": "test-tenant"})
+    assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

tests/test_integration.py

@@ -0,0 +1,305 @@
+"""
+End‑to‑end integration tests for the ARF governance pipeline.
+
+These tests exercise the full path from HTTP request to HealingIntent
+response, validating that every layer – API, governance loop, policy
+engine, risk engine, audit log, and optional skill/criticality features –
+behaves correctly under realistic conditions.
+
+v4.3.2: Covers basic evaluation, skill context, criticality, and audit
+trace verification.
+"""
+import pytest
+import time
+from app.database.models_intents import TenantDB, DecisionAuditLogDB
+
+
+@pytest.fixture(autouse=True)
+def seed_tenant(db_session):
+    """Ensure the tenant 'test-tenant' exists before each test."""
+    tenant = db_session.query(TenantDB).filter_by(id="test-tenant").first()
+    if not tenant:
+        db_session.add(TenantDB(id="test-tenant", name="Test Tenant"))
+        db_session.commit()
+
+
+class TestFullPipeline:
+    """End‑to‑end tests for the /intents/evaluate endpoint."""
+
+    def test_basic_provision_evaluation(self, client):
+        """A minimal valid request returns 200 and a well‑formed HealingIntent."""
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "eastus",
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": [],
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {}
+        }
+        response = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200, response.text
+        data = response.json()
+        # Top‑level fields
+        assert "risk_score" in data
+        assert "explanation" in data
+        assert "deterministic_id" in data
+        assert "recommended_action" in data
+        assert isinstance(data["risk_score"], float)
+        assert 0.0 <= data["risk_score"] <= 1.0
+        # HealingIntent contract
+        healing = data.get("healing_intent", {})
+        assert healing.get("action") is not None
+        assert healing.get("component") is not None
+        assert "justification" in healing
+        assert "confidence" in healing
+        assert "version" in healing
+        assert healing["version"] == "2.6.0"
+        # v4.3.2: context_hash must be present (64 hex chars)
+        ctx_hash = healing.get("context_hash")
+        assert isinstance(ctx_hash, str) and len(ctx_hash) == 64, (
+            f"context_hash missing or invalid: {ctx_hash}"
+        )
+
+    def test_audit_log_written(self, client, db_session):
+        """A successful evaluation writes a row to the decision audit log.
+
+        deterministic_id is a hash of (action, component, parameters,
+        incident_id, oss_edition) only -- and provision_resource requests
+        have no field that varies component or parameters -- so a minimal
+        provision_resource payload here would collide with the identical
+        one in test_basic_provision_evaluation and hit the audit log's
+        idempotency skip. Checking for the specific row by the response's
+        own deterministic_id (rather than a tenant-wide count delta) tests
+        the actual claim -- "this decision got audited" -- without being
+        sensitive to what other tests already wrote for the same decision.
+        """
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "eastus",
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": [],
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {}
+        }
+        response = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        deterministic_id = response.json()["deterministic_id"]
+        # The write_audit_log runs as a background task; give it a moment.
+        time.sleep(0.5)
+        entry = (
+            db_session.query(DecisionAuditLogDB)
+            .filter_by(tenant_id="test-tenant", deterministic_id=deterministic_id)
+            .first()
+        )
+        assert entry is not None, (
+            f"Expected an audit log entry for deterministic_id={deterministic_id}"
+        )
+
+    def test_skill_context_injection(self, client):
+        """When skill_id is provided, the response includes skill posterior data."""
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "eastus",
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": [],
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {},
+            "skill_id": "pdf-skill",
+        }
+        response = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        healing = response.json().get("healing_intent", {})
+        # Skill fields should be present in the HealingIntent
+        assert "skill_id" in healing
+        assert healing["skill_id"] == "pdf-skill"
+        # Because the skill registry is a singleton, the skill may or may not
+        # already exist.  In either case, the fields are populated with either
+        # the posterior or the default prior.
+        assert "skill_alpha" in healing
+        assert "skill_beta" in healing
+        assert "skill_reliability_score" in healing
+        assert "skill_version" in healing
+
+    def test_criticality_parameter(self, client):
+        """The criticality field is accepted and flows into the HealingIntent."""
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "eastus",
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": [],
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {},
+            "criticality": 0.85,
+        }
+        response = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        healing = response.json().get("healing_intent", {})
+        assert healing.get("criticality") == 0.85, (
+            f"criticality should be 0.85, got {healing.get('criticality')}"
+        )
+
+    def test_policy_violation_denial(self, client):
+        """An intent with a policy violation returns DENY."""
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "westus",                # not in default allowed set
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": ["Region 'westus' not allowed"],  # pre‑computed
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {}
+        }
+        response = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get("recommended_action") == "deny", (
+            f"Expected action=deny, got {data.get('recommended_action')}"
+        )
+
+
+class TestHealingPipeline:
+    """End‑to‑end tests for the /healing/evaluate endpoint."""
+
+    def test_basic_healing_evaluation(self, client):
+        """A reliability event triggers candidate healing actions."""
+        payload = {
+            "event": {
+                "component": "checkout-service",
+                "latency_p99": 600.0,
+                "error_rate": 0.25,
+                "service_mesh": "default",
+                "cpu_util": 0.85,
+                "memory_util": 0.90,
+            }
+        }
+        response = client.post(
+            "/api/v1/healing/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert "selected_action" in data
+        assert data["selected_action"] != "NO_ACTION", (
+            "Expected at least one healing action to be triggered"
+        )
+
+    def test_healing_with_skill_context(self, client):
+        """Skill context biases the healing decision utility."""
+        payload = {
+            "event": {
+                "component": "checkout-service",
+                "latency_p99": 600.0,
+                "error_rate": 0.25,
+                "service_mesh": "default",
+                "cpu_util": 0.85,
+                "memory_util": 0.90,
+            },
+            "skill_id": "pdf-skill",
+            "skill_version": 1,
+        }
+        response = client.post(
+            "/api/v1/healing/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        # The response should echo the skill context back
+        assert data.get("skill_id") == "pdf-skill"
+        assert data.get("skill_version") == 1
+        assert "selected_action" in data
+
+
+class TestOutcomeRecording:
+    """End‑to‑end tests for the /intents/outcome endpoint."""
+
+    def test_record_outcome_updates_risk_engine(self, client, db_session):
+        """Recording a successful outcome for a previously evaluated intent
+        updates the conjugate posterior and skill registry."""
+        # Step 1: evaluate an intent to create a record
+        payload = {
+            "intent_type": "provision_resource",
+            "environment": "prod",
+            "resource_type": "database",
+            "region": "eastus",
+            "size": "Standard",
+            "estimated_cost": 1200,
+            "policy_violations": [],
+            "requester": "alice",
+            "provenance": {},
+            "configuration": {},
+        }
+        eval_resp = client.post(
+            "/api/v1/intents/evaluate",
+            json=payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert eval_resp.status_code == 200
+        deterministic_id = eval_resp.json()["deterministic_id"]
+
+        # Step 2: record a successful outcome
+        outcome_payload = {
+            "deterministic_id": deterministic_id,
+            "success": True,
+            "recorded_by": "tester",
+            "notes": "integration test",
+        }
+        outcome_resp = client.post(
+            "/api/v1/intents/outcome",
+            json=outcome_payload,
+            headers={"X-Tenant-ID": "test-tenant"},
+        )
+        assert outcome_resp.status_code == 200
+        assert "outcome_id" in outcome_resp.json()
+
+        # Step 3: verify that the outcome row exists in the database
+        from app.database.models_intents import OutcomeDB
+        outcome = (
+            db_session.query(OutcomeDB)
+            .filter_by(idempotency_key=None)  # we didn't send one
+            .order_by(OutcomeDB.id.desc())
+            .first()
+        )
+        assert outcome is not None
+        assert outcome.success is True
+        assert outcome.recorded_by == "tester"

tests/test_intent_store.py

@@ -21,11 +21,12 @@ def test_save_intent(db_session):
     saved = save_evaluated_intent(
         db=db_session,
         deterministic_id=det_id,
+        tenant_id="test-tenant",
         intent_type="ProvisionResourceIntent",
         api_payload={"foo": "bar"},
         oss_payload={"intent_type": "provision_resource"},
         environment="prod",
-        risk_score=0.42
+        risk_score=0.42,
     )
     assert saved.deterministic_id == det_id
     assert saved.risk_score == "0.42"
@@ -37,9 +38,9 @@ def test_save_intent(db_session):
 
 def test_update_existing_intent(db_session):
     det_id = "intent_123"
-    save_evaluated_intent(db_session, det_id, "Type", {}, {}, "prod", 0.5)
-    updated = save_evaluated_intent(
-        db_session, det_id, "Type", {}, {}, "prod", 0.7)
+    # Positional order: db, deterministic_id, tenant_id, intent_type, api_payload, oss_payload, environment, risk_score
+    save_evaluated_intent(db_session, det_id, "test-tenant", "Type", {}, {}, "prod", 0.5)
+    updated = save_evaluated_intent(db_session, det_id, "test-tenant", "Type", {}, {}, "prod", 0.7)
     assert updated.risk_score == "0.7"
     count = db_session.query(IntentDB).filter(
         IntentDB.deterministic_id == det_id).count()

tests/test_outcome_service.py

@@ -4,7 +4,7 @@ from unittest.mock import MagicMock
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from app.database.base import Base
-from app.database.models_intents import IntentDB
+from app.database.models_intents import IntentDB, TenantDB
 from app.services.outcome_service import record_outcome, OutcomeConflictError
 from agentic_reliability_framework.core.governance.intents import (
     ProvisionResourceIntent,
@@ -18,6 +18,10 @@ def db_session():
     TestingSessionLocal = sessionmaker(bind=engine, future=True)
     Base.metadata.create_all(bind=engine)
     sess = TestingSessionLocal()
+    # Ensure a tenant exists for foreign key constraints
+    if not sess.query(TenantDB).filter_by(id="test-tenant").first():
+        sess.add(TenantDB(id="test-tenant", name="Test Tenant"))
+        sess.commit()
     yield sess
     sess.close()
 
@@ -42,6 +46,7 @@ def test_record_outcome_creates_row_and_updates_engine(
 
     intent = IntentDB(
         deterministic_id="intent_abc",
+        tenant_id="test-tenant",                     # <-- required
         intent_type="ProvisionResourceIntent",
         payload={},
         oss_payload=oss_payload,
@@ -53,6 +58,7 @@ def test_record_outcome_creates_row_and_updates_engine(
 
     outcome = record_outcome(
         db=db_session,
+        tenant_id="test-tenant",
         deterministic_id="intent_abc",
         success=True,
         recorded_by="tester",
@@ -69,6 +75,7 @@ def test_record_outcome_creates_row_and_updates_engine(
     # call engine again
     outcome2 = record_outcome(
         db=db_session,
+        tenant_id="test-tenant",
         deterministic_id="intent_abc",
         success=True,
         recorded_by="tester",
@@ -83,6 +90,7 @@ def test_record_outcome_creates_row_and_updates_engine(
 def test_conflict_different_result(db_session, mock_risk_engine):
     intent = IntentDB(
         deterministic_id="intent_def",
+        tenant_id="test-tenant",                     # <-- required
         intent_type="ProvisionResourceIntent",
         payload={},
         created_at=datetime.datetime.utcnow()
@@ -92,6 +100,7 @@ def test_conflict_different_result(db_session, mock_risk_engine):
 
     record_outcome(
         db_session,
+        "test-tenant",
         "intent_def",
         True,
         None,
@@ -100,6 +109,7 @@ def test_conflict_different_result(db_session, mock_risk_engine):
     with pytest.raises(OutcomeConflictError):
         record_outcome(
             db_session,
+            "test-tenant",
             "intent_def",
             False,
             None,
@@ -111,6 +121,7 @@ def test_nonexistent_intent(db_session, mock_risk_engine):
     with pytest.raises(ValueError):
         record_outcome(
             db_session,
+            "test-tenant",
             "missing",
             True,
             None,
@@ -123,6 +134,7 @@ def test_record_outcome_reconstruction_failure_does_not_update_engine(
     # Create an intent with invalid oss_payload (missing required fields)
     intent = IntentDB(
         deterministic_id="intent_bad",
+        tenant_id="test-tenant",                     # <-- required
         intent_type="ProvisionResourceIntent",
         payload={},
         oss_payload={"intent_type": "provision_resource"},  # missing fields
@@ -134,6 +146,7 @@ def test_record_outcome_reconstruction_failure_does_not_update_engine(
     # This should NOT call risk_engine.update_outcome (no dummy fallback)
     outcome = record_outcome(
         db=db_session,
+        tenant_id="test-tenant",
         deterministic_id="intent_bad",
         success=True,
         recorded_by="tester",

tests/test_performance.py

@@ -0,0 +1,100 @@
+"""
+Performance benchmarks for the ARF governance pipeline.
+
+These tests measure the latency of key operations and assert that
+they remain within the target thresholds for pilot readiness.
+
+Targets (v4.3.2):
+  - Full governance loop (single intent): p50 < 50 ms, p99 < 100 ms
+  - Policy evaluation alone: p50 < 1 ms
+  - Conjugate update: p50 < 0.1 ms
+  - HealingIntent serialization: p50 < 5 ms
+"""
+import time
+import pytest
+import numpy as np
+
+from agentic_reliability_framework.core.governance.governance_loop import GovernanceLoop
+from agentic_reliability_framework.core.governance.intents import (
+    ProvisionResourceIntent,
+    ResourceType,
+)
+from agentic_reliability_framework.core.governance.policies import PolicyEvaluator, allow_all
+from agentic_reliability_framework.core.governance.cost_estimator import CostEstimator
+from agentic_reliability_framework.core.governance.risk_engine import RiskEngine
+
+
+# Number of warmup iterations and measured iterations
+WARMUP = 10
+MEASURED = 50
+
+
+def _measure_latency(fn, *args, **kwargs):
+    """Run fn MEASURED times after WARMUP warmups, return (p50, p99, p100) in seconds."""
+    times = []
+    for _ in range(WARMUP):
+        fn(*args, **kwargs)
+    for _ in range(MEASURED):
+        t0 = time.perf_counter()
+        fn(*args, **kwargs)
+        times.append(time.perf_counter() - t0)
+    arr = np.array(times) * 1000  # convert to milliseconds
+    return np.percentile(arr, 50), np.percentile(arr, 99), arr.max()
+
+
+@pytest.fixture(scope="module")
+def sample_intent():
+    return ProvisionResourceIntent(
+        resource_type=ResourceType.VM,
+        region="eastus",
+        size="Standard_D2s_v3",
+        requester="perf-test",
+        environment="dev",
+    )
+
+
+@pytest.fixture(scope="module")
+def governance_loop():
+    return GovernanceLoop(
+        policy_evaluator=PolicyEvaluator(allow_all()),
+        cost_estimator=CostEstimator(),
+        risk_engine=RiskEngine(),
+        enable_epistemic=False,
+    )
+
+
+class TestGovernanceLoopPerformance:
+    """Latency benchmarks for the full governance loop."""
+
+    def test_full_loop_latency(self, governance_loop, sample_intent):
+        """The full loop should complete within 100 ms at p99."""
+        p50, p99, p100 = _measure_latency(
+            governance_loop.run, sample_intent, context={"service_name": "perf-svc"}
+        )
+        assert p50 < 100, f"p50 latency {p50:.1f} ms exceeds 100 ms target"
+        assert p99 < 200, f"p99 latency {p99:.1f} ms exceeds 200 ms target"
+
+
+class TestHealingIntentSerialization:
+    """Serialization performance."""
+
+    def test_to_enterprise_request_latency(self, governance_loop, sample_intent):
+        """Serializing a HealingIntent to the enterprise request dict should be fast."""
+        intent = governance_loop.run(sample_intent, context={"service_name": "perf-svc"})
+        p50, p99, p100 = _measure_latency(intent.to_enterprise_request)
+        assert p50 < 10, f"p50 serialization latency {p50:.1f} ms exceeds 10 ms target"
+
+
+class TestRiskEnginePerformance:
+    """Conjugate update latency."""
+
+    def test_risk_calculation_latency(self, governance_loop, sample_intent):
+        """A single risk calculation should be sub‑millisecond."""
+        engine = governance_loop.risk_engine
+        p50, p99, p100 = _measure_latency(
+            engine.calculate_risk,
+            intent=sample_intent,
+            cost_estimate=None,
+            policy_violations=[],
+        )
+        assert p50 < 10, f"p50 risk calculation latency {p50:.1f} ms exceeds 10 ms target"

tests/test_usage_tracker.py

@@ -11,14 +11,15 @@ def tracker():
 
 
 def test_get_or_create_api_key(tracker):
-    assert tracker.get_or_create_api_key("test_key", Tier.FREE) is True
+    # Updated: pass tenant_id as keyword argument (new signature)
+    assert tracker.get_or_create_api_key("test_key", tenant_id="test") is True
     assert tracker.get_tier("test_key") == Tier.FREE
     # Second call should return True without error
-    assert tracker.get_or_create_api_key("test_key") is True
+    assert tracker.get_or_create_api_key("test_key", tenant_id="test") is True
 
 
 def test_update_api_key_tier(tracker):
-    tracker.get_or_create_api_key("test_key", Tier.FREE)
+    tracker.get_or_create_api_key("test_key", tenant_id="test")
     assert tracker.update_api_key_tier("test_key", Tier.PRO) is True
     assert tracker.get_tier("test_key") == Tier.PRO
     # Non-existent key
@@ -26,7 +27,7 @@ def test_update_api_key_tier(tracker):
 
 
 def test_get_remaining_quota_free(tracker):
-    tracker.get_or_create_api_key("free_key", Tier.FREE)
+    tracker.get_or_create_api_key("free_key", tenant_id="test")
     # Initially 1000 remaining
     remaining = tracker.get_remaining_quota("free_key", Tier.FREE)
     assert remaining == 1000
@@ -43,13 +44,13 @@ def test_get_remaining_quota_free(tracker):
 
 
 def test_get_remaining_quota_enterprise(tracker):
-    tracker.get_or_create_api_key("ent_key", Tier.ENTERPRISE)
+    tracker.get_or_create_api_key("ent_key", tenant_id="test")
     remaining = tracker.get_remaining_quota("ent_key", Tier.ENTERPRISE)
     assert remaining is None
 
 
 def test_increment_usage_sync(tracker):
-    tracker.get_or_create_api_key("test_key", Tier.FREE)
+    tracker.get_or_create_api_key("test_key", tenant_id="test")
     record = UsageRecord(
         api_key="test_key",
         tier=Tier.FREE,
@@ -64,7 +65,7 @@ def test_increment_usage_sync(tracker):
 
 
 def test_get_audit_logs(tracker):
-    tracker.get_or_create_api_key("test_key", Tier.FREE)
+    tracker.get_or_create_api_key("test_key", tenant_id="test")
     record = UsageRecord(
         api_key="test_key",
         tier=Tier.FREE,