feat: add temporary Redis lookup diagnostic for controlled load tests

evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/factory.py

@@ -10,6 +10,7 @@
 from galileo_core.schemas.logging.span import (
     AgentSpan,
     ControlSpan,
+    HttpSpan,
     LlmSpan,
     RetrieverSpan,
     ToolSpan,
@@ -21,7 +22,9 @@
 
 type GalileoRecord = LlmSpan | ToolSpan | RetrieverSpan | Trace | Session
 type GalileoSpan = LlmSpan | ToolSpan | RetrieverSpan
-type GalileoCoreSpan = AgentSpan | WorkflowSpan | LlmSpan | RetrieverSpan | ToolSpan | ControlSpan
+type GalileoCoreSpan = (
+    AgentSpan | WorkflowSpan | LlmSpan | RetrieverSpan | ToolSpan | ControlSpan | HttpSpan
+)
 _MISSING = object()
 _SPAN_TYPES = {"llm", "tool", "retriever"}
 _RECORD_TYPES = _SPAN_TYPES | {"trace", "session"}
@@ -129,8 +132,10 @@ def record_from_step(
         selected_value = selected_data
         if selected_data is _MISSING:
             selected_value = selected_input if selected_input is not _MISSING else _MISSING
-        if selected_value is not _MISSING and selected_value is not None and not isinstance(
-            selected_value, Mapping
+        if (
+            selected_value is not _MISSING
+            and selected_value is not None
+            and not isinstance(selected_value, Mapping)
         ):
             raise RecordFactoryError(
                 f"Galileo {record_type} records require structured selector values; "

server/DIAGNOSTIC_LOOKUP.md

@@ -0,0 +1,21 @@
+# Controlled control-lookup diagnostic
+
+This default-disabled feature is for a dedicated owned load-test app. It is not production caching or production write-event invalidation. Do not enable it on a shared or customer app.
+
+The Orbit launcher injects its existing RedisConfig async client. The standalone server does not import Orbit or cache authorization. Without GALILEO_REDIS_HOST, Orbit does not construct a Redis client and database lookup remains available. Authentication and runtime-use authorization run on every evaluation, including Redis hits. The Redis connection is closed by the launcher lifespan.
+
+Enable AGENT_CONTROL_LOOKUP_DIAGNOSTIC_ENABLED only with an explicit AGENT_CONTROL_LOOKUP_DIAGNOSTIC_FIXTURES list and timezone-aware AGENT_CONTROL_LOOKUP_DIAGNOSTIC_EXPIRES_AT. Each fixture contains namespace, agent, target_type, target_id, generation, and controls_sha256. Compute the digest over ordered parsed runtime controls with definition_digest and verify it against fresh read-only fixture evidence before deployment. Duplicate target identities can declare distinct reviewed pair generations. Never put identifiers or contents in metric labels.
+
+Both modes use the same app image. Send X-LT-Control-Lookup-Cache as off or on and X-LT-Control-Lookup-Generation as the exact allowlisted pair generation. Without both headers the request is rejected; without either the original lookup path remains. Unknown/foreign/expired fixtures are rejected. Use a new reviewed generation for each comparison pair to prevent prior tests warming later pairs. Do not add excluded warmup traffic.
+
+OFF queries the database for the agent and effective controls, then parses the definitions. ON reads serialized runtime definitions from a versioned, hashed Redis key. It validates their digest and returns independently parsed models. A miss uses the same database loader and writes a bounded JSON value with a 300-second TTL. No ORM objects, results, authorization decisions or model outputs are cached. No pickle, global scan or global Redis cleanup is used.
+
+Every Redis command is bounded by the default 50ms deadline. A failed read, invalid cached value or timeout uses the DB path. A failed SET keeps the fresh DB result. The metric fallback counters must expose these cases; a passing semantic request does not prove a cache hit. Cache hits do not queue behind local locks. Local locks only coalesce misses within one process. They do not promise a single DB load across all replicas.
+
+While this diagnostic is enabled, its middleware rejects POST/PUT/PATCH/DELETE configuration operations under agents, controls, control-bindings and policies. This freezes the dedicated application, including after cache expiry, until the diagnostic is disabled or the original app restored. Reads, evaluation, token exchange and observation remain available. No external/direct database writer may change the fixtures. A rejected write or changed fixture blocks further load. Capture the original app digest before deployment and restore it after the test.
+
+Counter labels publish explicit startup zeros so missing series cannot be mistaken for no fallbacks. Numeric metrics are agent_control_lookup_diagnostic_seconds (mode and stage), agent_control_lookup_diagnostic_events_total (mode and outcome), and agent_control_lookup_diagnostic_sql_total (mode and query). Stages include authorization, pool_acquire, agent_query, effective_query, parse, control_load, redis_get, miss_wait, redis_set, cache_encode, cache_decode, engine and handler. Handler timing excludes authorization and response transport; use load-generator and existing HTTP middleware metrics for complete request latency. Query counters count attempted control-loader execute calls; they do not count authorization, pre-ping or every database operation.
+
+Accept the live treatment only with verified Redis hits, fewer loader SQL calls, unchanged control matches and zero failures, original SLOs, source/image/resource proof and required connected traces. Compare cold and steady windows and retain misses. Histograms are independent distributions and cannot be subtracted to reconstruct request timing. A single OFF/ON pair is exploratory and does not prove a causal bottleneck or capacity boundary.
+
+The snapshot-age histogram agent_control_lookup_diagnostic_snapshot_age_seconds measures seconds since successful database loading into the Redis envelope, only on valid hits. It is not the age of the underlying configuration. Cross-pod wall clocks are used; timestamps more than one second in the future or older than the maximum 600-second TTL fall back to the database. The miss_wait timer measures only waiting for the diagnostic process-local miss lock. It is not evaluator concurrency, model scheduling or database pool waiting. Cancellation always releases an acquired lock.

server/src/agent_control_server/diagnostic_freeze.py

@@ -0,0 +1,42 @@
+"""Reject configuration writes during the dedicated-stack diagnostic window."""
+
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from . import diagnostic_lookup
+from .diagnostic_metrics import _EVENTS
+
+
+class DiagnosticFixtureFreeze:
+    """Keep AC configuration fixed while the dedicated LT image is enabled.
+
+    Reads, token exchange, evaluation and observation remain available. Freeze
+    lasts until disabling the diagnostic or restoring the original image, even
+    after cache expiry. Never enable this on a shared application.
+    """
+
+    def __init__(self, app: ASGIApp, *, api_base: str) -> None:
+        self.app = app
+        self._prefixes = tuple(
+            api_base + "/" + name
+            for name in ("agents", "controls", "control-bindings", "policies")
+        )
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if (
+            diagnostic_lookup.settings.enabled
+            and scope["type"] == "http"
+            and scope["method"] in {"POST", "PUT", "PATCH", "DELETE"}
+            and any(
+                scope["path"] == p or scope["path"].startswith(p + "/")
+                for p in self._prefixes
+            )
+        ):
+            _EVENTS.labels("on", "fixture_write_rejected").inc()
+            response = JSONResponse(
+                status_code=409,
+                content={"detail": "Configuration is frozen for the controlled lookup diagnostic"},
+            )
+            await response(scope, receive, send)
+            return
+        await self.app(scope, receive, send)

server/src/agent_control_server/diagnostic_lookup.py

@@ -0,0 +1,227 @@
+"""Opt-in lookup-cache experiment for explicitly frozen, expiring LT fixtures.
+
+Not production invalidation: the controller must prohibit all fixture writes until
+restoration. Authorization always precedes this module. No ORM object is retained.
+"""
+
+import asyncio
+import hashlib
+import json
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Literal, Protocol
+
+from agent_control_models import ControlDefinitionRuntime
+from pydantic import BaseModel, Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from .diagnostic_metrics import _EVENTS, _SNAPSHOT_AGE, measured
+from .services.controls import RuntimeControl
+
+
+class FrozenFixture(BaseModel):
+    """Exact frozen fixture and expected digest of its ordered runtime definitions."""
+
+    namespace: str = Field(min_length=1)
+    agent: str = Field(min_length=1)
+    target_type: str = Field(min_length=1)
+    target_id: str = Field(min_length=1)
+    generation: str = Field(pattern=r"^[a-z0-9][a-z0-9-]{0,79}$")
+    controls_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+
+
+class DiagnosticSettings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="AGENT_CONTROL_LOOKUP_DIAGNOSTIC_", extra="ignore")
+    enabled: bool = False
+    fixtures: list[FrozenFixture] = Field(default_factory=list, max_length=32)
+    expires_at: datetime | None = None
+    ttl_seconds: float = Field(default=300, gt=0, le=600)
+    redis_timeout_seconds: float = Field(default=0.05, gt=0, le=0.25)
+    max_value_bytes: int = Field(default=1048576, ge=1024, le=1048576)
+
+    def fixture(
+        self,
+        namespace: str,
+        agent: str,
+        target_type: str | None,
+        target_id: str | None,
+        generation: str | None = None,
+    ) -> FrozenFixture | None:
+        expiry = self.expires_at
+        if (
+            not self.enabled
+            or expiry is None
+            or expiry.tzinfo is None
+            or expiry <= datetime.now(UTC)
+        ):
+            return None
+        return next(
+            (
+                f
+                for f in self.fixtures
+                if (f.namespace, f.agent, f.target_type, f.target_id)
+                == (namespace, agent, target_type, target_id)
+                and (generation is None or f.generation == generation)
+            ),
+            None,
+        )
+
+
+settings = DiagnosticSettings()
+
+
+def definition_digest(controls: list[RuntimeControl]) -> str:
+    """Stable ordered digest; output is a hash, never persisted control contents."""
+    data = [(c.id, c.name, c.control.model_dump(mode="json")) for c in controls]
+    return hashlib.sha256(
+        json.dumps(data, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+
+
+@dataclass(frozen=True)
+class CacheKey:
+    namespace: str
+    agent: str
+    target_type: str
+    target_id: str
+    generation: str
+    digest: str
+
+
+class RedisClient(Protocol):
+    """Small async transport seam; Orbit supplies RedisConfig's async client."""
+
+    async def get(self, name: str) -> bytes | str | None: ...
+    async def set(self, name: str, value: str, *, ex: int) -> object: ...
+    async def aclose(self) -> None: ...
+
+
+class CachedControl(BaseModel):
+    id: int
+    name: str
+    control: ControlDefinitionRuntime
+
+
+class CachedDefinitions(BaseModel):
+    format_version: Literal[1] = 1
+    controls: list[CachedControl]
+    loaded_at_unix: float = Field(gt=0, allow_inf_nan=False)
+
+
+def redis_key(key: CacheKey) -> str:
+    """Hash identifiers so keys expose no fixture identity and remain bounded."""
+    identity = json.dumps(
+        [key.namespace, key.agent, key.target_type, key.target_id, key.generation, key.digest],
+        separators=(",", ":"),
+    )
+    return "galileo:ac:lt-lookup:v1:" + hashlib.sha256(identity.encode()).hexdigest()
+
+
+class LookupCache:
+    """Redis shared cache restricted to a bounded, explicitly frozen fixture set.
+
+    No local results cache. Every ON request measures a Redis GET. Parse cached JSON
+    into independent runtime models. Failed, timed-out or corrupt Redis data falls
+    back to the authorized DB loader, never to stale local definitions. Generation
+    changes address a new key and TTL bounds old keys; fixture writes are forbidden
+    throughout this diagnostic. This is not production write-event invalidation.
+    """
+
+    def __init__(self, factory: Callable[[], RedisClient] | None = None) -> None:
+        self._factory = factory
+        self._client: RedisClient | None = None
+        self._locks = [asyncio.Lock() for _ in range(64)]
+
+    def _redis(self) -> RedisClient:
+        if self._client is None:
+            if self._factory is None:
+                raise RuntimeError("Redis diagnostic client not configured")
+            self._client = self._factory()
+        return self._client
+
+    async def aclose(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+
+    async def _read(self, key: CacheKey) -> list[RuntimeControl] | None:
+        with measured("redis_get"):
+            async with asyncio.timeout(settings.redis_timeout_seconds):
+                value = await self._redis().get(redis_key(key))
+        if value is None:
+            return None
+        if len(value) > settings.max_value_bytes:
+            raise ValueError("Redis diagnostic value exceeds bound")
+        with measured("cache_decode"):
+            parsed = CachedDefinitions.model_validate_json(value)
+            controls = [RuntimeControl(c.id, c.name, c.control) for c in parsed.controls]
+            if definition_digest(controls) != key.digest:
+                raise ValueError("Redis diagnostic digest mismatch")
+        age = time.time() - parsed.loaded_at_unix
+        if age < -1 or age > 600:
+            raise ValueError("Redis diagnostic snapshot timestamp outside bound")
+        _SNAPSHOT_AGE.labels("on").observe(max(0, age))
+        return controls
+
+    async def load(
+        self,
+        key: CacheKey,
+        loader: Callable[[], Awaitable[list[RuntimeControl]]],
+        *,
+        ttl: float,
+    ) -> list[RuntimeControl]:
+        # Hits must not queue behind a process-local lock. Only coalesce misses.
+        try:
+            controls = await self._read(key)
+        except Exception:
+            _EVENTS.labels("on", "redis_read_fallback").inc()
+            return await loader()
+        if controls is not None:
+            _EVENTS.labels("on", "hit").inc()
+            return controls
+        lock = self._locks[hash(key) % len(self._locks)]
+        with measured("miss_wait"):
+            await lock.acquire()
+        try:
+            try:
+                controls = await self._read(key)
+            except Exception:
+                _EVENTS.labels("on", "redis_read_fallback").inc()
+                return await loader()
+            if controls is not None:
+                _EVENTS.labels("on", "hit").inc()
+                return controls
+            _EVENTS.labels("on", "miss").inc()
+            controls = await loader()
+            if definition_digest(controls) != key.digest:
+                _EVENTS.labels("on", "fixture_mismatch").inc()
+                raise ValueError("Frozen diagnostic fixture digest mismatch")
+            with measured("cache_encode"):
+                value = CachedDefinitions(
+                    loaded_at_unix=time.time(),
+                    controls=[
+                        CachedControl(id=c.id, name=c.name, control=c.control) for c in controls
+                    ],
+                ).model_dump_json()
+            if len(value.encode()) > settings.max_value_bytes:
+                _EVENTS.labels("on", "oversize_fallback").inc()
+                return controls
+            try:
+                with measured("redis_set"):
+                    async with asyncio.timeout(settings.redis_timeout_seconds):
+                        await self._redis().set(redis_key(key), value, ex=max(1, int(ttl)))
+            except Exception:
+                _EVENTS.labels("on", "redis_write_fallback").inc()
+            return controls
+        finally:
+            lock.release()
+
+
+cache = LookupCache()
+
+
+def configure_redis_factory(factory: Callable[[], RedisClient]) -> None:
+    """Wire the deployed service's existing RedisConfig without importing Orbit."""
+    global cache
+    cache = LookupCache(factory)