Spaces:
Runtime error
Runtime error
File size: 3,661 Bytes
bc2df66 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """
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 unittest.mock import Mock, patch
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
from agentic_reliability_framework.core.governance.healing_intent import HealingIntent
# 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"
|