perf: scale Observe inventory views

CHANGELOG.md

@@ -49,6 +49,13 @@ This format follows [Keep a Changelog](https://keepachangelog.com/) and adheres
   contextual information icons, and totals for additive metrics. Model deployment
   falls back to the request model when a deployment attribute is absent, while the
   response model remains independently visible.
+- **Observe now scales large agent inventories without re-querying Azure Monitor
+  for every table interaction.** Discovery and normalized aggregates use shared
+  single-flight caches with stale-while-revalidate refresh, Azure Monitor batches
+  run with bounded concurrency, and Agents, Models, Tools, Runs, and Coverage add
+  server-side search, sorting, and pagination. Responses expose stage-level timing
+  through diagnostics and the `Server-Timing` header. Status badges also use
+  stronger borders, filled surfaces, and larger type in both themes.
 - **Observability readiness only reports what it can actually verify.** Multi-turn
   coverage is treated as a dataset property and inferred solely from conversation
   rows; rubric evaluators count as ready only when they are both declared and

docs/observe.md

@@ -73,6 +73,28 @@ filters, **Tools** accepts `tool_name` and **Runs** accepts `run_key`. Both only
 narrow results; blank values are rejected, and values are escaped before they
 reach telemetry queries.
 
+### Large inventories
+
+Observe keeps large inventories responsive by separating telemetry collection
+from table navigation. One bounded Azure Monitor collection builds a normalized
+aggregate of at most 5,000 rows. Agents, Models, Tools, Runs, and Coverage then
+search, sort, and paginate that aggregate through the Observe API, returning at
+most 100 rows per page. Changing the page, search text, sort column, or sort
+direction does not issue another Azure Monitor query.
+
+Discovery is cached for 15 minutes. Normalized aggregates stay fresh for two
+minutes and can be served stale for up to five additional minutes while one
+background refresh runs. Single-flight coordination prevents concurrent requests
+for the same scope from duplicating discovery or telemetry work, and source
+batches use bounded concurrency. An explicit refresh bypasses reusable view data.
+Protected trace content and delegated user-level results remain excluded from
+shared caches.
+
+Every response reports discovery, Azure Monitor, normalization, and total
+durations. The HTTP endpoint exposes the same stages in `Server-Timing`, together
+with cache hit, miss, bypass, or stale state, so operators can distinguish slow
+discovery from a slow telemetry query.
+
 ## Allocate declared billed totals
 
 The Cost view is an operational allocation of totals supplied by an operator.
@@ -583,7 +605,10 @@ and a visually hidden data `<table>`).
 
 Filters stay compact and visually subordinate to the summary. The Agents,
 Models, Tools, Runs, Costs, Attribution, and Coverage views are clear
-drill-down tables.
+drill-down tables. Status and classification badges use filled semantic surfaces,
+high-contrast borders, 12-pixel bold text, and the same minimum height as nearby
+controls in both themes, avoiding the thin low-resolution treatment used
+previously.
 
 ### Intentional states
 

specs/011-deploy-hosted-cockpit/contracts/observe-api.openapi.yaml

@@ -244,12 +244,32 @@ components:
       required: [view, filters]
       properties:
         view:
-          enum: [overview, agents, models, coverage]
+          enum: [overview, agents, models, coverage, tools, runs, cost]
         filters:
           $ref: "#/components/schemas/ObserveFilters"
         refresh:
           type: boolean
           default: false
+        page:
+          type: integer
+          minimum: 1
+          maximum: 1000
+          default: 1
+        page_size:
+          type: integer
+          minimum: 1
+          maximum: 100
+          default: 50
+        search:
+          type: [string, "null"]
+          maxLength: 200
+        sort_by:
+          type: [string, "null"]
+          maxLength: 64
+          pattern: "^[a-z][a-z0-9_]*$"
+        sort_direction:
+          enum: [asc, desc]
+          default: desc
     AgentDetailRequest:
       type: object
       additionalProperties: false
@@ -276,6 +296,30 @@ components:
           type: [string, "null"]
         model:
           type: [string, "null"]
+        tool_name:
+          type: [string, "null"]
+          maxLength: 256
+        run_key:
+          type: [string, "null"]
+          maxLength: 256
+        cost_period_id:
+          type: [string, "null"]
+          maxLength: 64
+        cost_breakdown:
+          type: [string, "null"]
+          enum: [agents, tools, runs, null]
+        cost_component_id:
+          type: [string, "null"]
+          maxLength: 64
+        cost_agent_key:
+          type: [string, "null"]
+          maxLength: 512
+        user_filter_token:
+          type: [string, "null"]
+          maxLength: 1024
+        department_filter_token:
+          type: [string, "null"]
+          maxLength: 1024
         start:
           type: string
           format: date-time
@@ -286,11 +330,11 @@ components:
       type: object
       additionalProperties: false
       required:
-        [view, data, trends, portal_links, coverage, partial_failures, diagnostics,
-         refreshed_at, cache_status]
+        [view, data, coverage, partial_failures, diagnostics, bounds, refreshed_at,
+         cache_status]
       properties:
         view:
-          enum: [overview, agents, models, coverage]
+          enum: [overview, agents, models, coverage, tools, runs, cost]
         data:
           oneOf:
             - type: object
@@ -306,11 +350,15 @@ components:
             $ref: "#/components/schemas/QuerySourceFailure"
         diagnostics:
           $ref: "#/components/schemas/QueryDiagnostics"
+        bounds:
+          oneOf:
+            - $ref: "#/components/schemas/ResultBounds"
+            - type: "null"
         refreshed_at:
           type: string
           format: date-time
         cache_status:
-          enum: [hit, miss, bypass]
+          enum: [hit, miss, bypass, stale]
     AgentDetailResponse:
       type: object
       additionalProperties: false
@@ -377,6 +425,9 @@ components:
         - started_at
         - completed_at
         - duration_ms
+        - discovery_duration_ms
+        - query_duration_ms
+        - normalization_duration_ms
         - source_count
         - successful_sources
         - partial_sources
@@ -392,10 +443,18 @@ components:
         duration_ms:
           type: integer
           minimum: 0
+        discovery_duration_ms:
+          type: integer
+          minimum: 0
+        query_duration_ms:
+          type: integer
+          minimum: 0
+        normalization_duration_ms:
+          type: integer
+          minimum: 0
         source_count:
           type: integer
           minimum: 0
-          maximum: 10
         successful_sources:
           type: integer
           minimum: 0
@@ -407,6 +466,41 @@ components:
           minimum: 0
         cache_status:
           enum: [hit, miss, bypass]
+    ResultBounds:
+      type: object
+      additionalProperties: false
+      required:
+        - rows_shown
+        - rows_total_in_scope
+        - truncated
+        - page
+        - page_size
+        - has_previous_page
+        - has_next_page
+      properties:
+        rows_shown:
+          type: integer
+          minimum: 0
+          maximum: 5000
+        rows_total_in_scope:
+          type: [integer, "null"]
+          minimum: 0
+        truncated:
+          type: boolean
+          default: false
+        page:
+          type: [integer, "null"]
+          minimum: 1
+        page_size:
+          type: [integer, "null"]
+          minimum: 1
+          maximum: 100
+        has_previous_page:
+          type: boolean
+          default: false
+        has_next_page:
+          type: boolean
+          default: false
     SourceFailure:
       type: object
       additionalProperties: false

src/agentops/agent/cockpit.py

@@ -20,6 +20,7 @@
 import re
 import shutil
 import subprocess
+import time
 from dataclasses import dataclass
 from importlib.resources import files as _pkg_files
 from pathlib import Path
@@ -5861,15 +5862,33 @@ async def _api_observe_query(
         filters = payload.filters
         if effective_scope is not None:
             filters.validate_scope(ObserveScope.model_validate(effective_scope))
-        return JSONResponse(
-            await _service_call(
-                "query",
-                view=payload.view,
-                filters=filters.model_dump(mode="json"),
-                refresh=payload.refresh,
-                user_context=user_context,
-            )
+        request_started = time.perf_counter()
+        body = await _service_call(
+            "query",
+            view=payload.view,
+            filters=filters.model_dump(mode="json"),
+            refresh=payload.refresh,
+            page=payload.page,
+            page_size=payload.page_size,
+            search=payload.search,
+            sort_by=payload.sort_by,
+            sort_direction=payload.sort_direction,
+            user_context=user_context,
         )
+        total_ms = (time.perf_counter() - request_started) * 1000
+        diagnostics = body.get("diagnostics", {}) if isinstance(body, dict) else {}
+        cache_status = body.get("cache_status", "miss") if isinstance(body, dict) else "miss"
+        timings = [f'total;dur={total_ms:.1f}', f'cache;desc="{cache_status}"']
+        if cache_status not in {"hit", "stale"} and isinstance(diagnostics, dict):
+            for name, field in (
+                ("discovery", "discovery_duration_ms"),
+                ("monitor", "query_duration_ms"),
+                ("normalize", "normalization_duration_ms"),
+            ):
+                duration = diagnostics.get(field)
+                if isinstance(duration, (int, float)):
+                    timings.append(f"{name};dur={max(float(duration), 0):.1f}")
+        return JSONResponse(body, headers={"Server-Timing": ", ".join(timings)})
 
     @app.post("/api/observe/attribution")
     async def _api_observe_attribution(

src/agentops/agent/observe/adapters.py

@@ -960,10 +960,14 @@ def __init__(
         credential: Any,
         source_timeout_seconds: int = SOURCE_TIMEOUT_SECONDS,
         request_deadline_seconds: int = DEFAULT_REQUEST_DEADLINE_SECONDS,
+        max_concurrent_batches: int = 4,
         clock: Callable[[], float] = time.monotonic,
     ) -> None:
+        if max_concurrent_batches < 1:
+            raise ValueError("max_concurrent_batches must be positive")
         self._source_timeout_seconds = source_timeout_seconds
         self._request_deadline_seconds = request_deadline_seconds
+        self._max_concurrent_batches = max_concurrent_batches
         self._clock = clock
         self._logs_client = _LogsQueryAdapter(credential=credential)
 
@@ -973,8 +977,10 @@ async def _run(
         build_query: Callable[[TelemetrySource], str],
     ) -> list[SourceResult]:
         queryable = [source for source in sources if source.workspace_id]
-        results: list[SourceResult] = []
-        for chunk in _chunked(queryable, MAX_SOURCES_PER_BATCH):
+        chunks = list(_chunked(queryable, MAX_SOURCES_PER_BATCH))
+        semaphore = asyncio.Semaphore(self._max_concurrent_batches)
+
+        async def run_chunk(chunk: Sequence[TelemetrySource]) -> list[SourceResult]:
             queries = [
                 SourceQuery(
                     source_id=source.source_id,
@@ -985,17 +991,18 @@ async def _run(
                 if source.workspace_id
             ]
             if not queries:
-                continue
-            results.extend(
-                await execute_source_batch(
+                return []
+            async with semaphore:
+                return await execute_source_batch(
                     queries,
                     client=self._logs_client,
                     source_timeout_seconds=self._source_timeout_seconds,
                     request_deadline_seconds=self._request_deadline_seconds,
                     clock=self._clock,
                 )
-            )
-        return results
+
+        chunk_results = await asyncio.gather(*(run_chunk(chunk) for chunk in chunks))
+        return [result for batch in chunk_results for result in batch]
 
     async def query(
         self,

src/agentops/agent/observe/cache.py

@@ -6,7 +6,7 @@
 from collections import OrderedDict
 from dataclasses import dataclass
 from threading import RLock
-from typing import Any, Callable, Generic, Hashable, TypeVar
+from typing import Any, Callable, Generic, Hashable, Literal, TypeVar
 
 
 class SensitiveValueError(ValueError):
@@ -49,6 +49,14 @@ class _Entry(Generic[V]):
     value: V
 
 
+@dataclass(frozen=True)
+class CacheLookup(Generic[V]):
+    """One cache lookup, including whether an expired value is still usable."""
+
+    state: Literal["fresh", "stale", "miss"]
+    value: V | None = None
+
+
 class ObserveCache(Generic[K, V]):
     """Thread-safe TTL/LRU cache that refuses sensitive Observe values."""
 
@@ -70,17 +78,32 @@ def __init__(
         self._lock = RLock()
 
     def get(self, key: K, *, bypass: bool = False) -> V | None:
+        return self.lookup(key, bypass=bypass).value
+
+    def lookup(
+        self,
+        key: K,
+        *,
+        bypass: bool = False,
+        max_stale_seconds: float = 0,
+    ) -> CacheLookup[V]:
+        """Return a fresh or explicitly allowed stale value for *key*."""
         if bypass:
-            return None
+            return CacheLookup(state="miss")
+        if max_stale_seconds < 0:
+            raise ValueError("max_stale_seconds cannot be negative")
         with self._lock:
             entry = self._entries.get(key)
             if entry is None:
-                return None
-            if self._clock() - entry.created_at >= self._ttl_seconds:
+                return CacheLookup(state="miss")
+            age = self._clock() - entry.created_at
+            if age >= self._ttl_seconds + max_stale_seconds:
                 del self._entries[key]
-                return None
+                return CacheLookup(state="miss")
             self._entries.move_to_end(key)
-            return entry.value
+            if age >= self._ttl_seconds:
+                return CacheLookup(state="stale", value=entry.value)
+            return CacheLookup(state="fresh", value=entry.value)
 
     def set(self, key: K, value: V) -> None:
         if _contains_sensitive_value(value):