diff --git a/CHANGELOG.md b/CHANGELOG.md
index 70667575..eef04a4f 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/docs/observe.md b/docs/observe.md
index 94bb2bb9..12b07619 100644
--- a/docs/observe.md
+++ b/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
 
diff --git a/specs/011-deploy-hosted-cockpit/contracts/observe-api.openapi.yaml b/specs/011-deploy-hosted-cockpit/contracts/observe-api.openapi.yaml
index 4f283460..1f69e147 100644
--- a/specs/011-deploy-hosted-cockpit/contracts/observe-api.openapi.yaml
+++ b/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
diff --git a/src/agentops/agent/cockpit.py b/src/agentops/agent/cockpit.py
index 04119739..d5e9a84f 100644
--- a/src/agentops/agent/cockpit.py
+++ b/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(
diff --git a/src/agentops/agent/observe/adapters.py b/src/agentops/agent/observe/adapters.py
index 71c01f67..1707e347 100644
--- a/src/agentops/agent/observe/adapters.py
+++ b/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,
diff --git a/src/agentops/agent/observe/cache.py b/src/agentops/agent/observe/cache.py
index 53b65117..b1484574 100644
--- a/src/agentops/agent/observe/cache.py
+++ b/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):
diff --git a/src/agentops/agent/observe/facade.py b/src/agentops/agent/observe/facade.py
index 1656ec9e..f7645e76 100644
--- a/src/agentops/agent/observe/facade.py
+++ b/src/agentops/agent/observe/facade.py
@@ -60,7 +60,13 @@
     classify_appgenai_content_result,
     execute_source_batch,
 )
-from agentops.agent.observe.service import CACHE_TTL_SECONDS, ObserveResult, ObserveService, View
+from agentops.agent.observe.service import (
+    CACHE_TTL_SECONDS,
+    INVENTORY_CACHE_TTL_SECONDS,
+    ObserveResult,
+    ObserveService,
+    View,
+)
 from agentops.agent.observe.ui import build_azure_resource_portal_url
 from agentops.core.attribution import (
     AttributionConfigurationLoadResult,
@@ -411,6 +417,9 @@ def __init__(
             credential=credential, clock=monotonic_clock
         )
         self._cache = cache or ObserveCache(ttl_seconds=CACHE_TTL_SECONDS)
+        self._inventory_cache: ObserveCache = ObserveCache(
+            ttl_seconds=INVENTORY_CACHE_TTL_SECONDS
+        )
         runtime_identity = (
             cast(str, uami_client_id)
             if auth_mode == "hosted"
@@ -423,6 +432,8 @@ def __init__(
             runtime=self._runtime,
             clock=clock,
             cache=self._cache,
+            inventory_cache=self._inventory_cache,
+            monotonic_clock=monotonic_clock,
         )
 
     # -- discover -----------------------------------------------------
@@ -442,6 +453,11 @@ async def query(
         view: str,
         filters: Mapping[str, Any],
         refresh: bool = False,
+        page: int = 1,
+        page_size: int = 50,
+        search: str | None = None,
+        sort_by: str | None = None,
+        sort_direction: Literal["asc", "desc"] = "desc",
         user_context: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Return one normalized, coverage-annotated Observe view.
@@ -484,7 +500,15 @@ async def query(
 
         native_view = cast(View, view)
         result = await self._service.query_view(
-            self._scope, filter_state, view=native_view, refresh=refresh
+            self._scope,
+            filter_state,
+            view=native_view,
+            refresh=refresh,
+            page=page,
+            page_size=page_size,
+            search=search,
+            sort_by=sort_by,
+            sort_direction=sort_direction,
         )
         return _serialize_observe_result(result)
 
diff --git a/src/agentops/agent/observe/service.py b/src/agentops/agent/observe/service.py
index e3813f3b..16166f3d 100644
--- a/src/agentops/agent/observe/service.py
+++ b/src/agentops/agent/observe/service.py
@@ -19,7 +19,10 @@
 
 import asyncio
 import json
-from dataclasses import dataclass
+import logging
+import time
+from collections.abc import Callable, Coroutine, Hashable
+from dataclasses import dataclass, replace
 from datetime import datetime, timedelta
 from decimal import Decimal
 from typing import Any, Literal, Mapping, Protocol, Sequence
@@ -93,6 +96,10 @@
 
 #: Identity/scope/filter cache entries stay fresh for two minutes (T046).
 CACHE_TTL_SECONDS = 120.0
+INVENTORY_CACHE_TTL_SECONDS = 15 * 60.0
+VIEW_STALE_TTL_SECONDS = 5 * 60.0
+
+logger = logging.getLogger(__name__)
 
 View = Literal["overview", "agents", "models", "tools", "runs", "cost"]
 
@@ -184,7 +191,7 @@ class ObserveResult:
     partial_failures: list[PartialFailure]
     bounds: ResultBounds | None
     refreshed_at: datetime
-    cache_status: Literal["hit", "miss", "bypass"]
+    cache_status: Literal["hit", "miss", "bypass", "stale"]
 
 
 @dataclass(frozen=True)
@@ -1171,6 +1178,187 @@ def _bound_view_data(view: View, data: Sequence[Any]) -> list[Any]:
     )[:MAX_ROWS_PER_QUERY]
 
 
+_VIEW_SORT_FIELDS: dict[str, frozenset[str]] = {
+    "agents": frozenset(
+        {
+            "agent_name",
+            "agent_id",
+            "source_kind",
+            "model",
+            "last_seen",
+            "invocations",
+            "failures",
+            "failure_rate",
+            "p95_latency_ms",
+            "input_tokens",
+            "output_tokens",
+            "total_tokens",
+            "cache_read_tokens",
+            "cache_write_tokens",
+            "reasoning_tokens",
+        }
+    ),
+    "models": frozenset(
+        {
+            "model",
+            "deployment",
+            "last_seen",
+            "requests",
+            "failures",
+            "failure_rate",
+            "p95_latency_ms",
+            "input_tokens",
+            "output_tokens",
+            "total_tokens",
+            "cache_read_tokens",
+            "cache_write_tokens",
+            "reasoning_tokens",
+        }
+    ),
+    "tools": frozenset(
+        {
+            "tool_name",
+            "agent_name",
+            "source_id",
+            "source_kind",
+            "last_seen",
+            "invocations",
+            "failures",
+            "p95_latency_ms",
+        }
+    ),
+    "runs": frozenset(
+        {
+            "run_key",
+            "run_key_kind",
+            "agent_name",
+            "source_id",
+            "source_kind",
+            "started_at",
+            "last_activity_at",
+            "duration_ms",
+            "status",
+            "turns",
+            "tool_invocations",
+            "input_tokens",
+            "output_tokens",
+            "total_tokens",
+            "cache_read_tokens",
+            "cache_write_tokens",
+            "reasoning_tokens",
+        }
+    ),
+}
+_DEFAULT_VIEW_SORT = {
+    "agents": "invocations",
+    "models": "requests",
+    "tools": "invocations",
+    "runs": "last_activity_at",
+}
+
+
+def _row_value(row: Any, field: str) -> Any:
+    if field == "total_tokens":
+        input_tokens = _row_value(row, "input_tokens")
+        output_tokens = _row_value(row, "output_tokens")
+        if input_tokens is None and output_tokens is None:
+            return None
+        return (input_tokens or 0) + (output_tokens or 0)
+    if field == "failure_rate":
+        failures = _row_value(row, "failures")
+        total = _row_value(row, "invocations")
+        if total is None:
+            total = _row_value(row, "requests")
+        if failures is None or not total:
+            return None
+        return failures / total
+    if isinstance(row, Mapping):
+        return row.get(field)
+    return getattr(row, field, None)
+
+
+def _sortable_value(value: Any) -> Any:
+    if isinstance(value, datetime):
+        return value.timestamp()
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, str):
+        return value.casefold()
+    return value
+
+
+def _searchable_row(row: Any) -> str:
+    if hasattr(row, "model_dump"):
+        value = row.model_dump(mode="json")
+    elif isinstance(row, Mapping):
+        value = dict(row)
+    else:
+        value = vars(row)
+    return json.dumps(value, sort_keys=True, default=str).casefold()
+
+
+def _page_observe_result(
+    result: ObserveResult,
+    *,
+    page: int,
+    page_size: int,
+    search: str | None,
+    sort_by: str | None,
+    sort_direction: Literal["asc", "desc"],
+) -> ObserveResult:
+    """Sort, filter, and page one cached aggregate without re-querying Azure."""
+    if result.view not in _VIEW_SORT_FIELDS or not isinstance(result.data, list):
+        return result
+    allowed_fields = _VIEW_SORT_FIELDS[result.view]
+    field = sort_by or _DEFAULT_VIEW_SORT[result.view]
+    if field not in allowed_fields:
+        raise ValueError(f"unsupported sort field {field!r} for {result.view}")
+
+    rows = list(result.data)
+    normalized_search = (search or "").strip().casefold()
+    if normalized_search:
+        rows = [row for row in rows if normalized_search in _searchable_row(row)]
+
+    present: list[tuple[int, Any, Any]] = []
+    missing: list[tuple[int, Any]] = []
+    for index, row in enumerate(rows):
+        value = _row_value(row, field)
+        if value is None:
+            missing.append((index, row))
+        else:
+            present.append((index, row, _sortable_value(value)))
+    present.sort(
+        key=lambda item: (item[2], item[0]),
+        reverse=sort_direction == "desc",
+    )
+    ordered = [item[1] for item in present] + [item[1] for item in missing]
+
+    start = (page - 1) * page_size
+    end = start + page_size
+    visible = ordered[start:end]
+    source_bounds = result.bounds
+    hard_truncated = bool(source_bounds and source_bounds.truncated)
+    if normalized_search and hard_truncated:
+        total: int | None = None
+    elif normalized_search:
+        total = len(ordered)
+    elif source_bounds is not None:
+        total = source_bounds.rows_total_in_scope
+    else:
+        total = len(ordered)
+    has_next = end < len(ordered)
+    bounds = ResultBounds(
+        rows_shown=len(visible),
+        rows_total_in_scope=total,
+        truncated=hard_truncated,
+        page=page,
+        page_size=page_size,
+        has_previous_page=page > 1,
+        has_next_page=has_next,
+    )
+    return replace(result, data=visible, bounds=bounds)
+
+
 def _build_partial_failures(results: Sequence[SourceResult]) -> list[PartialFailure]:
     """Summarize every non-``success`` source outcome as a safe partial failure (T061).
 
@@ -1779,12 +1967,79 @@ def __init__(
         runtime: RuntimeContext,
         clock: Clock,
         cache: ObserveCache,
+        inventory_cache: ObserveCache | None = None,
+        monotonic_clock: Callable[[], float] = time.monotonic,
     ) -> None:
         self._discovery_client = discovery_client
         self._query_client = query_client
         self._runtime = runtime
         self._clock = clock
         self._cache = cache
+        self._inventory_cache = inventory_cache or cache
+        self._monotonic_clock = monotonic_clock
+        self._inflight: dict[Hashable, asyncio.Task[Any]] = {}
+        self._inflight_lock = asyncio.Lock()
+        self._background_tasks: set[asyncio.Task[Any]] = set()
+
+    async def _run_coalesced(
+        self,
+        key: Hashable,
+        operation: Callable[[], Coroutine[Any, Any, Any]],
+    ) -> tuple[Any, bool]:
+        """Run one operation per cache key and let concurrent callers share it."""
+        async with self._inflight_lock:
+            task = self._inflight.get(key)
+            owner = task is None
+            if task is None:
+                task = asyncio.create_task(operation())
+                self._inflight[key] = task
+
+                def remove_inflight(completed: asyncio.Task[Any]) -> None:
+                    if self._inflight.get(key) is completed:
+                        self._inflight.pop(key, None)
+
+                task.add_done_callback(remove_inflight)
+        return await asyncio.shield(task), owner
+
+    def _schedule_refresh(
+        self,
+        key: Hashable,
+        operation: Callable[[], Coroutine[Any, Any, Any]],
+    ) -> None:
+        """Refresh a stale aggregate without delaying the current response."""
+        task = asyncio.create_task(self._run_coalesced(key, operation))
+        self._background_tasks.add(task)
+
+        def finish(completed: asyncio.Task[Any]) -> None:
+            self._background_tasks.discard(completed)
+            if completed.cancelled():
+                return
+            error = completed.exception()
+            if error is not None:
+                logger.warning(
+                    "Observe background refresh failed for key %r",
+                    key,
+                    exc_info=(type(error), error, error.__traceback__),
+                )
+
+        task.add_done_callback(finish)
+
+    def _cached_result(
+        self,
+        cached: _CachedView,
+        *,
+        cache_status: Literal["hit", "stale"],
+    ) -> ObserveResult:
+        return ObserveResult(
+            view=cached.view,
+            data=cached.data,
+            coverage=cached.coverage,
+            diagnostics=cached.diagnostics,
+            partial_failures=cached.partial_failures,
+            bounds=cached.bounds,
+            refreshed_at=cached.refreshed_at,
+            cache_status=cache_status,
+        )
 
     async def get_inventory(
         self, scope: ObserveScope, *, refresh: bool = False
@@ -1792,11 +2047,16 @@ async def get_inventory(
         """Discover (or return the cached) resource inventory for *scope*."""
         identity = _identity_key(self._runtime)
         key = _cache_key(identity, scope, "discovery", None)
-        cached = self._cache.get(key, bypass=refresh)
+        cached = self._inventory_cache.get(key, bypass=refresh)
         if cached is not None:
             return cached
-        inventory = await self._discovery_client.discover(scope)
-        self._cache.set(key, inventory)
+
+        async def discover() -> ResourceInventory:
+            inventory = await self._discovery_client.discover(scope)
+            self._inventory_cache.set(key, inventory)
+            return inventory
+
+        inventory, _owner = await self._run_coalesced(key, discover)
         return inventory
 
     async def query_view(
@@ -1806,41 +2066,109 @@ async def query_view(
         *,
         view: View,
         refresh: bool = False,
+        page: int = 1,
+        page_size: int = 50,
+        search: str | None = None,
+        sort_by: str | None = None,
+        sort_direction: Literal["asc", "desc"] = "desc",
     ) -> ObserveResult:
         """Return the normalized, coverage-annotated response for *view*."""
         filters.validate_scope(scope)
         identity = _identity_key(self._runtime)
         key = _cache_key(identity, scope, view, filters)
-        cached = self._cache.get(key, bypass=refresh)
-        if cached is not None:
-            return ObserveResult(
-                view=cached.view,
-                data=cached.data,
-                coverage=cached.coverage,
-                diagnostics=cached.diagnostics,
-                partial_failures=cached.partial_failures,
-                bounds=cached.bounds,
-                refreshed_at=cached.refreshed_at,
-                cache_status="hit",
+        cached = self._cache.lookup(
+            key,
+            bypass=refresh,
+            max_stale_seconds=VIEW_STALE_TTL_SECONDS,
+        )
+        if cached.state == "fresh" and cached.value is not None:
+            result = self._cached_result(cached.value, cache_status="hit")
+            return _page_observe_result(
+                result,
+                page=page,
+                page_size=page_size,
+                search=search,
+                sort_by=sort_by,
+                sort_direction=sort_direction,
+            )
+        if cached.state == "stale" and cached.value is not None:
+            self._schedule_refresh(
+                key,
+                lambda: self._query_view_uncached(
+                    scope,
+                    filters,
+                    view=view,
+                    key=key,
+                    cache_status="miss",
+                ),
+            )
+            result = self._cached_result(cached.value, cache_status="stale")
+            return _page_observe_result(
+                result,
+                page=page,
+                page_size=page_size,
+                search=search,
+                sort_by=sort_by,
+                sort_direction=sort_direction,
             )
 
+        result, owner = await self._run_coalesced(
+            key,
+            lambda: self._query_view_uncached(
+                scope,
+                filters,
+                view=view,
+                key=key,
+                cache_status="bypass" if refresh else "miss",
+            ),
+        )
+        if owner:
+            response = result
+        else:
+            response = replace(result, cache_status="hit")
+        return _page_observe_result(
+            response,
+            page=page,
+            page_size=page_size,
+            search=search,
+            sort_by=sort_by,
+            sort_direction=sort_direction,
+        )
+
+    async def _query_view_uncached(
+        self,
+        scope: ObserveScope,
+        filters: ObserveFilterState,
+        *,
+        view: View,
+        key: tuple[Any, ...],
+        cache_status: Literal["miss", "bypass"],
+    ) -> ObserveResult:
+        request_started_at = self._clock()
+        request_started = self._monotonic_clock()
+        discovery_started = self._monotonic_clock()
         # Refresh telemetry without repeating the slower control-plane discovery.
         # Inventory has its own cache and explicit discover endpoint for forced refreshes.
         inventory = await self.get_inventory(scope)
+        discovery_duration_ms = int(
+            (self._monotonic_clock() - discovery_started) * 1000
+        )
         available_sources = [
             source
             for source in inventory.telemetry_sources
             if source.state == "available"
         ]
 
-        started_at = self._clock()
+        query_started = self._monotonic_clock()
         source_results = list(
             await self._query_client.query(available_sources, filters, view=view)
         )
-        completed_at = self._clock()
+        query_duration_ms = int((self._monotonic_clock() - query_started) * 1000)
 
+        normalization_started = self._monotonic_clock()
+        normalization_refreshed_at = self._clock()
         coverage = self._discovery_coverage(
-            inventory.telemetry_sources, refreshed_at=completed_at
+            inventory.telemetry_sources, refreshed_at=normalization_refreshed_at
         )
         data, query_coverage = self._normalize_view(
             view,
@@ -1848,7 +2176,7 @@ async def query_view(
             inventory.telemetry_sources,
             inventory=inventory,
             window_end=filters.end,
-            refreshed_at=completed_at,
+            refreshed_at=normalization_refreshed_at,
         )
         if view in ("agents", "models", "tools", "runs"):
             data = _bound_view_data(view, data)
@@ -1858,12 +2186,21 @@ async def query_view(
             if view in ("agents", "models", "tools", "runs")
             else None
         )
+        completed_at = self._clock()
+        normalization_duration_ms = int(
+            (self._monotonic_clock() - normalization_started) * 1000
+        )
+        duration_ms = int((self._monotonic_clock() - request_started) * 1000)
 
         diagnostics = self._build_diagnostics(
             source_results,
-            started_at=started_at,
+            started_at=request_started_at,
             completed_at=completed_at,
-            cache_status="bypass" if refresh else "miss",
+            cache_status=cache_status,
+            duration_ms=duration_ms,
+            discovery_duration_ms=discovery_duration_ms,
+            query_duration_ms=query_duration_ms,
+            normalization_duration_ms=normalization_duration_ms,
         )
         partial_failures = _build_partial_failures(source_results)
 
@@ -1887,7 +2224,7 @@ async def query_view(
             partial_failures=partial_failures,
             bounds=bounds,
             refreshed_at=completed_at,
-            cache_status="bypass" if refresh else "miss",
+            cache_status=cache_status,
         )
 
     async def query_attribution(
@@ -3738,6 +4075,10 @@ def _build_diagnostics(
         started_at: datetime,
         completed_at: datetime,
         cache_status: Literal["miss", "bypass"],
+        duration_ms: int | None = None,
+        discovery_duration_ms: int = 0,
+        query_duration_ms: int = 0,
+        normalization_duration_ms: int = 0,
     ) -> QueryDiagnostics:
         successful = sum(1 for result in results if result.status == "success")
         partial = sum(1 for result in results if result.status == "partial")
@@ -3746,11 +4087,15 @@ def _build_diagnostics(
             for result in results
             if result.status in ("timeout", "throttled", "error")
         )
-        duration_ms = max(int((completed_at - started_at).total_seconds() * 1000), 0)
+        if duration_ms is None:
+            duration_ms = int((completed_at - started_at).total_seconds() * 1000)
         return QueryDiagnostics(
             started_at=started_at,
             completed_at=completed_at,
-            duration_ms=duration_ms,
+            duration_ms=max(duration_ms, 0),
+            discovery_duration_ms=max(discovery_duration_ms, 0),
+            query_duration_ms=max(query_duration_ms, 0),
+            normalization_duration_ms=max(normalization_duration_ms, 0),
             source_count=len(results),
             successful_sources=successful,
             partial_sources=partial,
diff --git a/src/agentops/agent/observe/ui.py b/src/agentops/agent/observe/ui.py
index 8ef78b69..302d1659 100644
--- a/src/agentops/agent/observe/ui.py
+++ b/src/agentops/agent/observe/ui.py
@@ -2865,12 +2865,42 @@ def render_trace_detail_shell(
   background: color-mix(in srgb, var(--observe-accent) 26%, transparent);
 }
 .observe-refresh-button:hover { border-color: var(--observe-border-strong); }
+.observe-refresh-button:disabled {
+  cursor: not-allowed;
+  opacity: 0.45;
+}
 .observe-apply-button:focus-visible,
 .observe-refresh-button:focus-visible {
   outline: 2px solid var(--observe-accent);
   outline-offset: 2px;
 }
 .observe-refresh-status { color: var(--observe-muted); font-size: 12px; }
+.observe-page-controls {
+  display: flex;
+  align-items: center;
+  flex-wrap: wrap;
+  gap: 8px;
+  margin: 0 0 8px;
+}
+.observe-page-controls input,
+.observe-page-controls select {
+  box-sizing: border-box;
+  min-height: 32px;
+  border: 1px solid var(--observe-border);
+  border-radius: 8px;
+  background: var(--observe-card-bg);
+  color: var(--observe-fg);
+  font: inherit;
+  font-size: 13px;
+  padding: 5px 9px;
+}
+.observe-page-controls input { min-width: 220px; }
+.observe-page-status {
+  color: var(--observe-muted);
+  font-size: 12px;
+  min-width: 48px;
+  text-align: center;
+}
 .observe-scope { margin: 0 0 10px; font-size: 12px; color: var(--observe-muted); }
 
 .observe-attribution-summary {
@@ -3089,12 +3119,43 @@ def render_trace_detail_shell(
 .observe-badge {
   display: inline-flex;
   align-items: center;
+  justify-content: center;
+  min-height: 24px;
+  box-sizing: border-box;
   border-radius: 999px;
-  padding: 1px 8px;
-  font-size: 11px;
-  font-weight: 600;
-  line-height: 1.6;
-  border: 1px solid currentColor;
+  padding: 3px 9px;
+  font-size: 12px;
+  font-weight: 700;
+  line-height: 1.25;
+  letter-spacing: 0.01em;
+  white-space: nowrap;
+  border: 1px solid var(--observe-border-strong);
+  background: var(--observe-card-hi);
+}
+.observe-badge.observe-tone-ok {
+  border-color: color-mix(in srgb, var(--observe-ok) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-ok) 16%, var(--observe-card-bg));
+  color: var(--observe-ok);
+}
+.observe-badge.observe-tone-warn {
+  border-color: color-mix(in srgb, var(--observe-warn) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-warn) 14%, var(--observe-card-bg));
+  color: var(--observe-warn);
+}
+.observe-badge.observe-tone-crit {
+  border-color: color-mix(in srgb, var(--observe-crit) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-crit) 14%, var(--observe-card-bg));
+  color: var(--observe-crit);
+}
+.observe-badge.observe-tone-info {
+  border-color: color-mix(in srgb, var(--observe-accent) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-accent) 14%, var(--observe-card-bg));
+  color: var(--observe-accent);
+}
+.observe-badge.observe-tone-muted {
+  border-color: var(--observe-border-strong);
+  background: color-mix(in srgb, var(--observe-muted) 10%, var(--observe-card-bg));
+  color: var(--observe-muted);
 }
 .observe-tone-ok { color: var(--observe-ok); }
 .observe-tone-warn { color: var(--observe-warn); }
@@ -3230,6 +3291,7 @@ def render_trace_detail_shell(
   var ATTRIBUTION_COST_UNAVAILABLE = "Cost attribution is unavailable. Configure a valid cost model and allocatable cost before selecting Cost.";
   var AUTO_REFRESH_MS = 300000; // five minutes
   var DEFAULT_RANGE_MS = 24 * 60 * 60 * 1000; // trailing 24 hours
+  var CACHE_WINDOW_MS = 60 * 1000; // align default windows across browser sessions
   // Mirrors MAX_TREND_POINTS in ui.py: even though the backend is expected
   // to already bound each trend series (T053), the client re-bounds
   // defensively so a chart never renders unbounded markup regardless of
@@ -3239,6 +3301,39 @@ def render_trace_detail_shell(
   // from contracts/observe-api.openapi.yaml (mirrors OBSERVE_VIEW_WIRE_NAMES
   // in ui.py -- the internal "usage" id is spelled "models" on the wire).
   var VIEW_WIRE_NAMES = { overview: "overview", agents: "agents", usage: "models", tools: "tools", runs: "runs" };
+  var SERVER_SORT_KEYS = {
+    "observe-agents-table": {
+      "Agent": "agent_name", "Source": "source_kind", "Model": "model",
+      "Last seen": "last_seen", "Invocations": "invocations",
+      "Failure rate": "failure_rate", "p95 latency": "p95_latency_ms",
+      "Input tokens": "input_tokens", "Output tokens": "output_tokens",
+      "Total tokens": "total_tokens", "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens"
+    },
+    "observe-usage-table": {
+      "Model": "model", "Deployment": "deployment", "Requests": "requests",
+      "Failure rate": "failure_rate", "p95 latency": "p95_latency_ms",
+      "Input tokens": "input_tokens", "Output tokens": "output_tokens",
+      "Total tokens": "total_tokens", "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens",
+      "Last seen": "last_seen"
+    },
+    "observe-tools-table": {
+      "Tool": "tool_name", "Agent": "agent_name", "Source": "source_id",
+      "Runtime": "source_kind", "Last seen": "last_seen",
+      "Invocations": "invocations", "Failures": "failures",
+      "p95 latency": "p95_latency_ms"
+    },
+    "observe-runs-table": {
+      "Run key": "run_key", "Correlation": "run_key_kind", "Agent": "agent_name",
+      "Source": "source_id", "Runtime": "source_kind", "Started in range": "started_at",
+      "Duration in range": "duration_ms", "Status": "status", "Turns in range": "turns",
+      "Tool invocations": "tool_invocations", "Input tokens": "input_tokens",
+      "Output tokens": "output_tokens", "Total tokens": "total_tokens",
+      "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens"
+    }
+  };
   // Best-effort, human-friendly labels for *documented* portal link keys
   // (mirrors _KNOWN_PORTAL_LABELS in ui.py). Any key not listed here still
   // renders (title-cased) rather than being dropped -- the "best-effort
@@ -3257,6 +3352,11 @@ def render_trace_detail_shell(
   var draftFilters = {};
   var appliedFilters = {};
   var currentView = "overview";
+  var currentPage = 1;
+  var currentPageSize = 50;
+  var currentSearch = "";
+  var currentSortBy = "";
+  var currentSortDirection = "desc";
   var requestToken = 0;
   var activeController = null;
   var refreshTimer = null;
@@ -3299,11 +3399,22 @@ def render_trace_detail_shell(
     if (currentView === "cost" && !document.getElementById("cost")) {
       currentView = "overview";
     }
+    currentPage = Math.max(1, parseInt(params.get("page") || "1", 10) || 1);
+    currentPageSize = [25, 50, 100].indexOf(parseInt(params.get("page_size") || "50", 10)) >= 0
+      ? parseInt(params.get("page_size") || "50", 10)
+      : 50;
+    currentSearch = String(params.get("search") || "").slice(0, 200);
+    currentSortBy = String(params.get("sort_by") || "").slice(0, 64);
+    currentSortDirection = params.get("sort_direction") === "asc" ? "asc" : "desc";
     if (!applied.start || !applied.end) {
-      var end = new Date();
-      var start = new Date(end.getTime() - DEFAULT_RANGE_MS);
-      applied.start = applied.start || start.toISOString();
-      applied.end = applied.end || end.toISOString();
+      var end = applied.end
+        ? new Date(applied.end)
+        : new Date(Math.floor(Date.now() / CACHE_WINDOW_MS) * CACHE_WINDOW_MS);
+      var start = applied.start
+        ? new Date(applied.start)
+        : new Date(end.getTime() - DEFAULT_RANGE_MS);
+      applied.start = start.toISOString();
+      applied.end = end.toISOString();
     }
     return applied;
   }
@@ -3330,6 +3441,13 @@ def render_trace_detail_shell(
       });
     }
     params.set("view", currentView);
+    if (isPagedView(currentView)) {
+      params.set("page", String(currentPage));
+      params.set("page_size", String(currentPageSize));
+      if (currentSearch) params.set("search", currentSearch);
+      if (currentSortBy) params.set("sort_by", currentSortBy);
+      params.set("sort_direction", currentSortDirection);
+    }
     params.set("theme", document.documentElement.getAttribute("data-theme") || "dark");
     return window.location.pathname + "?" + params.toString();
   }
@@ -3812,13 +3930,27 @@ def render_trace_detail_shell(
       var label = header.dataset.label || String(header.textContent || "").trim() ||
         "Column " + (columnIndex + 1);
       var helpText = header.dataset.help || "";
+      var serverSortKey = header.dataset.serverSortKey || "";
       clearChildren(header);
-      header.setAttribute("aria-sort", "none");
+      var activeDirection = serverSortKey && currentSortBy === serverSortKey
+        ? (currentSortDirection === "asc" ? "ascending" : "descending")
+        : "none";
+      header.setAttribute("aria-sort", activeDirection);
       var button = makeEl("button", "observe-sort-button", label);
       button.type = "button";
       button.title = "Sort by " + label;
       button.setAttribute("aria-label", "Sort by " + label);
       button.addEventListener("click", function () {
+        if (serverSortKey) {
+          currentSortDirection = currentSortBy === serverSortKey && currentSortDirection === "asc"
+            ? "desc"
+            : "asc";
+          currentSortBy = serverSortKey;
+          currentPage = 1;
+          syncUrl();
+          fetchObserveData(false);
+          return;
+        }
         body.querySelectorAll("[data-observe-drilldown-row]").forEach(function (detailRow) {
           detailRow.remove();
         });
@@ -3889,6 +4021,7 @@ def render_trace_detail_shell(
     table.setAttribute("aria-label", ariaLabel);
     var thead = document.createElement("thead");
     var headRow = document.createElement("tr");
+    var sortKeys = SERVER_SORT_KEYS[className] || {};
     columns.forEach(function (column) {
       var definition = typeof column === "string" ? { label: column } : column;
       var th = makeEl("th", null, definition.label);
@@ -3897,6 +4030,9 @@ def render_trace_detail_shell(
       if (definition.help) {
         th.dataset.help = definition.help;
       }
+      if (sortKeys[definition.label]) {
+        th.dataset.serverSortKey = sortKeys[definition.label];
+      }
       headRow.appendChild(th);
     });
     thead.appendChild(headRow);
@@ -4142,12 +4278,110 @@ def render_trace_detail_shell(
 
   function boundsNoticeNode(bounds, rowsShown) {
     var total = bounds && bounds.rows_total_in_scope;
-    var text = total === null || total === undefined
-      ? "Showing " + rowsShown + " rows."
-      : "Showing " + rowsShown + " of " + total + " rows in scope.";
+    var page = bounds && bounds.page ? bounds.page : currentPage;
+    var pageSize = bounds && bounds.page_size ? bounds.page_size : currentPageSize;
+    var first = rowsShown ? ((page - 1) * pageSize) + 1 : 0;
+    var last = rowsShown ? first + rowsShown - 1 : 0;
+    var text;
+    if (!rowsShown && page > 1) {
+      text = "No rows are available on page " + page + ".";
+    } else if (bounds && bounds.truncated) {
+      text = "Showing rows " + first + "\u2013" + last + " from the highest-ranked results";
+      text += total === null || total === undefined
+        ? "."
+        : " (" + total + " rows in scope).";
+    } else {
+      text = total === null || total === undefined
+        ? "Showing rows " + first + "\u2013" + last + "."
+        : "Showing rows " + first + "\u2013" + last + " of " + total + ".";
+    }
     return makeEl("p", "observe-hint observe-bounds-notice", text);
   }
 
+  function isPagedView(view) {
+    return ["agents", "usage", "tools", "runs"].indexOf(view) >= 0;
+  }
+
+  function resetPaging(resetQuery) {
+    currentPage = 1;
+    if (resetQuery) {
+      currentSearch = "";
+      currentSortBy = "";
+      currentSortDirection = "desc";
+    }
+  }
+
+  function paginationToolbar(bounds) {
+    var toolbar = makeEl("form", "observe-page-controls");
+    toolbar.setAttribute("role", "search");
+    toolbar.addEventListener("submit", function (event) {
+      event.preventDefault();
+      currentSearch = String(search.value || "").trim().slice(0, 200);
+      currentPageSize = parseInt(pageSize.value, 10) || 50;
+      currentPage = 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+
+    var search = document.createElement("input");
+    search.type = "search";
+    search.value = currentSearch;
+    search.placeholder = "Search this view";
+    search.setAttribute("aria-label", "Search this view");
+    toolbar.appendChild(search);
+
+    var pageSize = document.createElement("select");
+    pageSize.setAttribute("aria-label", "Rows per page");
+    [25, 50, 100].forEach(function (size) {
+      var option = document.createElement("option");
+      option.value = String(size);
+      option.textContent = size + " rows";
+      option.selected = size === currentPageSize;
+      pageSize.appendChild(option);
+    });
+    toolbar.appendChild(pageSize);
+    toolbar.appendChild(makeEl("button", "observe-refresh-button", "Search"));
+    var clear = makeEl("button", "observe-refresh-button", "Clear");
+    clear.type = "button";
+    clear.disabled = !currentSearch;
+    clear.addEventListener("click", function () {
+      currentSearch = "";
+      currentPage = 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(clear);
+
+    var previous = makeEl("button", "observe-refresh-button", "Previous");
+    previous.type = "button";
+    previous.disabled = !(bounds && bounds.has_previous_page);
+    previous.addEventListener("click", function () {
+      currentPage = Math.max(1, currentPage - 1);
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(previous);
+
+    var status = makeEl(
+      "span",
+      "observe-page-status",
+      "Page " + String((bounds && bounds.page) || currentPage)
+    );
+    status.setAttribute("aria-live", "polite");
+    toolbar.appendChild(status);
+
+    var next = makeEl("button", "observe-refresh-button", "Next");
+    next.type = "button";
+    next.disabled = !(bounds && bounds.has_next_page);
+    next.addEventListener("click", function () {
+      currentPage += 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(next);
+    return toolbar;
+  }
+
   function renderOverview(data, diagnostics) {
     var metrics = overviewMetricsFrom(data);
     if (!metrics.length) {
@@ -4180,10 +4414,12 @@ def render_trace_detail_shell(
     setViewContent("overview", [grid]);
   }
 
-  function renderAgents(data, diagnostics) {
+  function renderAgents(data, diagnostics, bounds) {
     var agents = agentsFrom(data);
+    var notice = boundsNoticeNode(bounds, agents.length);
+    var controls = paginationToolbar(bounds);
     if (!agents.length) {
-      setViewContent("agents", [emptyStateNode("No data found for the selected filters.")]);
+      setViewContent("agents", [controls, notice, emptyStateNode("No data found for the selected filters.")]);
       return;
     }
     var rows = agents.map(function (agent) {
@@ -4262,7 +4498,7 @@ def render_trace_detail_shell(
         "\u2014"
       ]
     );
-    setViewContent("agents", [table]);
+    setViewContent("agents", [controls, notice, table]);
   }
 
   // ---------------------------------------------------------------------
@@ -4535,10 +4771,12 @@ def render_trace_detail_shell(
       });
   }
 
-  function renderUsage(data, diagnostics) {
+  function renderUsage(data, diagnostics, bounds) {
     var usage = modelsFrom(data);
+    var notice = boundsNoticeNode(bounds, usage.length);
+    var controls = paginationToolbar(bounds);
     if (!usage.length) {
-      setViewContent("usage", [emptyStateNode("No data found for the selected filters.")]);
+      setViewContent("usage", [controls, notice, emptyStateNode("No data found for the selected filters.")]);
       return;
     }
     var rows = usage.map(function (entry) {
@@ -4612,14 +4850,16 @@ def render_trace_detail_shell(
         "\u2014"
       ]
     );
-    setViewContent("usage", [table]);
+    setViewContent("usage", [controls, notice, table]);
   }
 
   function renderTools(data, diagnostics, bounds) {
     var tools = toolsFrom(data);
     var notice = boundsNoticeNode(bounds, tools.length);
+    var controls = paginationToolbar(bounds);
     if (!tools.length) {
       setViewContent("tools", [
+        controls,
         notice,
         emptyStateNode("No tool activity was found for the selected filters. Tool attribution may not be reported for this selection."),
       ]);
@@ -4675,14 +4915,16 @@ def render_trace_detail_shell(
         "\u2014"
       ]
     );
-    setViewContent("tools", [notice, table]);
+    setViewContent("tools", [controls, notice, table]);
   }
 
   function renderRuns(data, diagnostics, bounds) {
     var runs = runsFrom(data);
     var notice = boundsNoticeNode(bounds, runs.length);
+    var controls = paginationToolbar(bounds);
     if (!runs.length) {
       setViewContent("runs", [
+        controls,
         notice,
         emptyStateNode("No runs could be correlated for the selected filters. Run correlation may not be reported for this selection."),
       ]);
@@ -4763,7 +5005,7 @@ def render_trace_detail_shell(
         renderMaybeMissing(sumReported(runs, "reasoning_tokens"), { missingText: "\u2014" })
       ]
     );
-    setViewContent("runs", [notice, table]);
+    setViewContent("runs", [controls, notice, table]);
   }
 
   function costLabel(value) {
@@ -5657,9 +5899,9 @@ def render_trace_detail_shell(
     if (view === "overview") {
       renderOverview(body.data, body.diagnostics);
     } else if (view === "agents") {
-      renderAgents(body.data, body.diagnostics);
+      renderAgents(body.data, body.diagnostics, body.bounds);
     } else if (view === "usage") {
-      renderUsage(body.data, body.diagnostics);
+      renderUsage(body.data, body.diagnostics, body.bounds);
     } else if (view === "tools") {
       renderTools(body.data, body.diagnostics, body.bounds);
     } else if (view === "runs") {
@@ -5774,6 +6016,11 @@ def render_trace_detail_shell(
         end: appliedFilters.end,
       },
       refresh: manual === true,
+      page: currentPage,
+      page_size: currentPageSize,
+      search: currentSearch || null,
+      sort_by: currentSortBy || null,
+      sort_direction: currentSortDirection,
     };
     payload = currentView === "cost" ? buildCostPayload(manual) : payload;
     if (currentView === "departments") {
@@ -5888,6 +6135,7 @@ def render_trace_detail_shell(
           appliedFilters.attribution_cost_period_id = preservedAttributionFilters.attribution_cost_period_id;
           appliedFilters.attribution_cost_component_id = preservedAttributionFilters.attribution_cost_component_id;
         }
+        resetPaging(false);
         syncUrl();
         fetchObserveData(true);
       });
@@ -5964,7 +6212,9 @@ def render_trace_detail_shell(
     document.querySelectorAll("[data-observe-nav-link]").forEach(function (link) {
       link.addEventListener("click", function (event) {
         event.preventDefault();
-        activateView(link.getAttribute("data-observe-nav-link"));
+        var nextView = link.getAttribute("data-observe-nav-link");
+        if (nextView !== currentView) resetPaging(true);
+        activateView(nextView);
         pushUrl();
         // Switching views queries a different `ObserveQuery.view`, so the
         // newly active view must be fetched -- otherwise it would only ever
diff --git a/src/agentops/core/observe.py b/src/agentops/core/observe.py
index d4d08a3c..db8b65e0 100644
--- a/src/agentops/core/observe.py
+++ b/src/agentops/core/observe.py
@@ -47,8 +47,11 @@
     "protected_or_unavailable",
 ]
 
-# This mirrors the bounded Observe query contract without coupling core to agent code.
-MAX_ROWS_PER_QUERY = 500
+# Aggregate queries can retain enough rows for large fleets while API responses
+# remain independently page-bounded.
+MAX_ROWS_PER_QUERY = 5000
+DEFAULT_PAGE_SIZE = 50
+MAX_PAGE_SIZE = 100
 
 _SUBSCRIPTION_RE = re.compile(r"^/subscriptions/[^/]+$", re.IGNORECASE)
 _RESOURCE_GROUP_RE = re.compile(
@@ -426,6 +429,23 @@ class ObserveQueryRequest(ContractModel):
     view: ObserveView
     filters: ObserveFilterState
     refresh: bool = False
+    page: int = Field(default=1, ge=1, le=1000)
+    page_size: int = Field(default=DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE)
+    search: str | None = Field(default=None, max_length=200)
+    sort_by: str | None = Field(
+        default=None,
+        max_length=64,
+        pattern=r"^[a-z][a-z0-9_]*$",
+    )
+    sort_direction: Literal["asc", "desc"] = "desc"
+
+    @field_validator("search")
+    @classmethod
+    def _normalize_search(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        normalized = value.strip()
+        return normalized or None
 
 
 class AgentDetailRequest(ContractModel):
@@ -482,7 +502,10 @@ class QueryDiagnostics(ContractModel):
     started_at: datetime
     completed_at: datetime
     duration_ms: int = Field(ge=0)
-    source_count: int = Field(ge=0, le=10)
+    discovery_duration_ms: int = Field(default=0, ge=0)
+    query_duration_ms: int = Field(default=0, ge=0)
+    normalization_duration_ms: int = Field(default=0, ge=0)
+    source_count: int = Field(ge=0)
     successful_sources: int = Field(ge=0)
     partial_sources: int = Field(ge=0)
     failed_sources: int = Field(ge=0)
@@ -629,6 +652,10 @@ class ResultBounds(ContractModel):
     rows_shown: int = Field(ge=0, le=MAX_ROWS_PER_QUERY)
     rows_total_in_scope: int | None = Field(default=None, ge=0)
     truncated: bool = False
+    page: int | None = Field(default=None, ge=1)
+    page_size: int | None = Field(default=None, ge=1, le=MAX_PAGE_SIZE)
+    has_previous_page: bool = False
+    has_next_page: bool = False
 
     @model_validator(mode="after")
     def _validate_bounds(self) -> "ResultBounds":
@@ -637,10 +664,8 @@ def _validate_bounds(self) -> "ResultBounds":
             and self.rows_total_in_scope < self.rows_shown
         ):
             raise ValueError("rows_total_in_scope cannot be less than rows_shown")
-        if self.truncated and self.rows_shown != MAX_ROWS_PER_QUERY:
-            raise ValueError(
-                "truncated results must show exactly MAX_ROWS_PER_QUERY rows"
-            )
+        if (self.page is None) != (self.page_size is None):
+            raise ValueError("page and page_size must be reported together")
         return self
 
 
diff --git a/tests/unit/__snapshots__/observe_overview.html b/tests/unit/__snapshots__/observe_overview.html
index 9bbe6ebe..26bc5159 100644
--- a/tests/unit/__snapshots__/observe_overview.html
+++ b/tests/unit/__snapshots__/observe_overview.html
@@ -310,12 +310,42 @@
   background: color-mix(in srgb, var(--observe-accent) 26%, transparent);
 }
 .observe-refresh-button:hover { border-color: var(--observe-border-strong); }
+.observe-refresh-button:disabled {
+  cursor: not-allowed;
+  opacity: 0.45;
+}
 .observe-apply-button:focus-visible,
 .observe-refresh-button:focus-visible {
   outline: 2px solid var(--observe-accent);
   outline-offset: 2px;
 }
 .observe-refresh-status { color: var(--observe-muted); font-size: 12px; }
+.observe-page-controls {
+  display: flex;
+  align-items: center;
+  flex-wrap: wrap;
+  gap: 8px;
+  margin: 0 0 8px;
+}
+.observe-page-controls input,
+.observe-page-controls select {
+  box-sizing: border-box;
+  min-height: 32px;
+  border: 1px solid var(--observe-border);
+  border-radius: 8px;
+  background: var(--observe-card-bg);
+  color: var(--observe-fg);
+  font: inherit;
+  font-size: 13px;
+  padding: 5px 9px;
+}
+.observe-page-controls input { min-width: 220px; }
+.observe-page-status {
+  color: var(--observe-muted);
+  font-size: 12px;
+  min-width: 48px;
+  text-align: center;
+}
 .observe-scope { margin: 0 0 10px; font-size: 12px; color: var(--observe-muted); }
 
 .observe-attribution-summary {
@@ -534,12 +564,43 @@
 .observe-badge {
   display: inline-flex;
   align-items: center;
+  justify-content: center;
+  min-height: 24px;
+  box-sizing: border-box;
   border-radius: 999px;
-  padding: 1px 8px;
-  font-size: 11px;
-  font-weight: 600;
-  line-height: 1.6;
-  border: 1px solid currentColor;
+  padding: 3px 9px;
+  font-size: 12px;
+  font-weight: 700;
+  line-height: 1.25;
+  letter-spacing: 0.01em;
+  white-space: nowrap;
+  border: 1px solid var(--observe-border-strong);
+  background: var(--observe-card-hi);
+}
+.observe-badge.observe-tone-ok {
+  border-color: color-mix(in srgb, var(--observe-ok) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-ok) 16%, var(--observe-card-bg));
+  color: var(--observe-ok);
+}
+.observe-badge.observe-tone-warn {
+  border-color: color-mix(in srgb, var(--observe-warn) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-warn) 14%, var(--observe-card-bg));
+  color: var(--observe-warn);
+}
+.observe-badge.observe-tone-crit {
+  border-color: color-mix(in srgb, var(--observe-crit) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-crit) 14%, var(--observe-card-bg));
+  color: var(--observe-crit);
+}
+.observe-badge.observe-tone-info {
+  border-color: color-mix(in srgb, var(--observe-accent) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-accent) 14%, var(--observe-card-bg));
+  color: var(--observe-accent);
+}
+.observe-badge.observe-tone-muted {
+  border-color: var(--observe-border-strong);
+  background: color-mix(in srgb, var(--observe-muted) 10%, var(--observe-card-bg));
+  color: var(--observe-muted);
 }
 .observe-tone-ok { color: var(--observe-ok); }
 .observe-tone-warn { color: var(--observe-warn); }
@@ -777,6 +838,7 @@ <h2 id="runs-heading">Runs</h2>
   var ATTRIBUTION_COST_UNAVAILABLE = "Cost attribution is unavailable. Configure a valid cost model and allocatable cost before selecting Cost.";
   var AUTO_REFRESH_MS = 300000; // five minutes
   var DEFAULT_RANGE_MS = 24 * 60 * 60 * 1000; // trailing 24 hours
+  var CACHE_WINDOW_MS = 60 * 1000; // align default windows across browser sessions
   // Mirrors MAX_TREND_POINTS in ui.py: even though the backend is expected
   // to already bound each trend series (T053), the client re-bounds
   // defensively so a chart never renders unbounded markup regardless of
@@ -786,6 +848,39 @@ <h2 id="runs-heading">Runs</h2>
   // from contracts/observe-api.openapi.yaml (mirrors OBSERVE_VIEW_WIRE_NAMES
   // in ui.py -- the internal "usage" id is spelled "models" on the wire).
   var VIEW_WIRE_NAMES = { overview: "overview", agents: "agents", usage: "models", tools: "tools", runs: "runs" };
+  var SERVER_SORT_KEYS = {
+    "observe-agents-table": {
+      "Agent": "agent_name", "Source": "source_kind", "Model": "model",
+      "Last seen": "last_seen", "Invocations": "invocations",
+      "Failure rate": "failure_rate", "p95 latency": "p95_latency_ms",
+      "Input tokens": "input_tokens", "Output tokens": "output_tokens",
+      "Total tokens": "total_tokens", "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens"
+    },
+    "observe-usage-table": {
+      "Model": "model", "Deployment": "deployment", "Requests": "requests",
+      "Failure rate": "failure_rate", "p95 latency": "p95_latency_ms",
+      "Input tokens": "input_tokens", "Output tokens": "output_tokens",
+      "Total tokens": "total_tokens", "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens",
+      "Last seen": "last_seen"
+    },
+    "observe-tools-table": {
+      "Tool": "tool_name", "Agent": "agent_name", "Source": "source_id",
+      "Runtime": "source_kind", "Last seen": "last_seen",
+      "Invocations": "invocations", "Failures": "failures",
+      "p95 latency": "p95_latency_ms"
+    },
+    "observe-runs-table": {
+      "Run key": "run_key", "Correlation": "run_key_kind", "Agent": "agent_name",
+      "Source": "source_id", "Runtime": "source_kind", "Started in range": "started_at",
+      "Duration in range": "duration_ms", "Status": "status", "Turns in range": "turns",
+      "Tool invocations": "tool_invocations", "Input tokens": "input_tokens",
+      "Output tokens": "output_tokens", "Total tokens": "total_tokens",
+      "Cache read": "cache_read_tokens",
+      "Cache write": "cache_write_tokens", "Reasoning": "reasoning_tokens"
+    }
+  };
   // Best-effort, human-friendly labels for *documented* portal link keys
   // (mirrors _KNOWN_PORTAL_LABELS in ui.py). Any key not listed here still
   // renders (title-cased) rather than being dropped -- the "best-effort
@@ -804,6 +899,11 @@ <h2 id="runs-heading">Runs</h2>
   var draftFilters = {};
   var appliedFilters = {};
   var currentView = "overview";
+  var currentPage = 1;
+  var currentPageSize = 50;
+  var currentSearch = "";
+  var currentSortBy = "";
+  var currentSortDirection = "desc";
   var requestToken = 0;
   var activeController = null;
   var refreshTimer = null;
@@ -846,11 +946,22 @@ <h2 id="runs-heading">Runs</h2>
     if (currentView === "cost" && !document.getElementById("cost")) {
       currentView = "overview";
     }
+    currentPage = Math.max(1, parseInt(params.get("page") || "1", 10) || 1);
+    currentPageSize = [25, 50, 100].indexOf(parseInt(params.get("page_size") || "50", 10)) >= 0
+      ? parseInt(params.get("page_size") || "50", 10)
+      : 50;
+    currentSearch = String(params.get("search") || "").slice(0, 200);
+    currentSortBy = String(params.get("sort_by") || "").slice(0, 64);
+    currentSortDirection = params.get("sort_direction") === "asc" ? "asc" : "desc";
     if (!applied.start || !applied.end) {
-      var end = new Date();
-      var start = new Date(end.getTime() - DEFAULT_RANGE_MS);
-      applied.start = applied.start || start.toISOString();
-      applied.end = applied.end || end.toISOString();
+      var end = applied.end
+        ? new Date(applied.end)
+        : new Date(Math.floor(Date.now() / CACHE_WINDOW_MS) * CACHE_WINDOW_MS);
+      var start = applied.start
+        ? new Date(applied.start)
+        : new Date(end.getTime() - DEFAULT_RANGE_MS);
+      applied.start = start.toISOString();
+      applied.end = end.toISOString();
     }
     return applied;
   }
@@ -877,6 +988,13 @@ <h2 id="runs-heading">Runs</h2>
       });
     }
     params.set("view", currentView);
+    if (isPagedView(currentView)) {
+      params.set("page", String(currentPage));
+      params.set("page_size", String(currentPageSize));
+      if (currentSearch) params.set("search", currentSearch);
+      if (currentSortBy) params.set("sort_by", currentSortBy);
+      params.set("sort_direction", currentSortDirection);
+    }
     params.set("theme", document.documentElement.getAttribute("data-theme") || "dark");
     return window.location.pathname + "?" + params.toString();
   }
@@ -1359,13 +1477,27 @@ <h2 id="runs-heading">Runs</h2>
       var label = header.dataset.label || String(header.textContent || "").trim() ||
         "Column " + (columnIndex + 1);
       var helpText = header.dataset.help || "";
+      var serverSortKey = header.dataset.serverSortKey || "";
       clearChildren(header);
-      header.setAttribute("aria-sort", "none");
+      var activeDirection = serverSortKey && currentSortBy === serverSortKey
+        ? (currentSortDirection === "asc" ? "ascending" : "descending")
+        : "none";
+      header.setAttribute("aria-sort", activeDirection);
       var button = makeEl("button", "observe-sort-button", label);
       button.type = "button";
       button.title = "Sort by " + label;
       button.setAttribute("aria-label", "Sort by " + label);
       button.addEventListener("click", function () {
+        if (serverSortKey) {
+          currentSortDirection = currentSortBy === serverSortKey && currentSortDirection === "asc"
+            ? "desc"
+            : "asc";
+          currentSortBy = serverSortKey;
+          currentPage = 1;
+          syncUrl();
+          fetchObserveData(false);
+          return;
+        }
         body.querySelectorAll("[data-observe-drilldown-row]").forEach(function (detailRow) {
           detailRow.remove();
         });
@@ -1436,6 +1568,7 @@ <h2 id="runs-heading">Runs</h2>
     table.setAttribute("aria-label", ariaLabel);
     var thead = document.createElement("thead");
     var headRow = document.createElement("tr");
+    var sortKeys = SERVER_SORT_KEYS[className] || {};
     columns.forEach(function (column) {
       var definition = typeof column === "string" ? { label: column } : column;
       var th = makeEl("th", null, definition.label);
@@ -1444,6 +1577,9 @@ <h2 id="runs-heading">Runs</h2>
       if (definition.help) {
         th.dataset.help = definition.help;
       }
+      if (sortKeys[definition.label]) {
+        th.dataset.serverSortKey = sortKeys[definition.label];
+      }
       headRow.appendChild(th);
     });
     thead.appendChild(headRow);
@@ -1689,12 +1825,110 @@ <h2 id="runs-heading">Runs</h2>
 
   function boundsNoticeNode(bounds, rowsShown) {
     var total = bounds && bounds.rows_total_in_scope;
-    var text = total === null || total === undefined
-      ? "Showing " + rowsShown + " rows."
-      : "Showing " + rowsShown + " of " + total + " rows in scope.";
+    var page = bounds && bounds.page ? bounds.page : currentPage;
+    var pageSize = bounds && bounds.page_size ? bounds.page_size : currentPageSize;
+    var first = rowsShown ? ((page - 1) * pageSize) + 1 : 0;
+    var last = rowsShown ? first + rowsShown - 1 : 0;
+    var text;
+    if (!rowsShown && page > 1) {
+      text = "No rows are available on page " + page + ".";
+    } else if (bounds && bounds.truncated) {
+      text = "Showing rows " + first + "–" + last + " from the highest-ranked results";
+      text += total === null || total === undefined
+        ? "."
+        : " (" + total + " rows in scope).";
+    } else {
+      text = total === null || total === undefined
+        ? "Showing rows " + first + "–" + last + "."
+        : "Showing rows " + first + "–" + last + " of " + total + ".";
+    }
     return makeEl("p", "observe-hint observe-bounds-notice", text);
   }
 
+  function isPagedView(view) {
+    return ["agents", "usage", "tools", "runs"].indexOf(view) >= 0;
+  }
+
+  function resetPaging(resetQuery) {
+    currentPage = 1;
+    if (resetQuery) {
+      currentSearch = "";
+      currentSortBy = "";
+      currentSortDirection = "desc";
+    }
+  }
+
+  function paginationToolbar(bounds) {
+    var toolbar = makeEl("form", "observe-page-controls");
+    toolbar.setAttribute("role", "search");
+    toolbar.addEventListener("submit", function (event) {
+      event.preventDefault();
+      currentSearch = String(search.value || "").trim().slice(0, 200);
+      currentPageSize = parseInt(pageSize.value, 10) || 50;
+      currentPage = 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+
+    var search = document.createElement("input");
+    search.type = "search";
+    search.value = currentSearch;
+    search.placeholder = "Search this view";
+    search.setAttribute("aria-label", "Search this view");
+    toolbar.appendChild(search);
+
+    var pageSize = document.createElement("select");
+    pageSize.setAttribute("aria-label", "Rows per page");
+    [25, 50, 100].forEach(function (size) {
+      var option = document.createElement("option");
+      option.value = String(size);
+      option.textContent = size + " rows";
+      option.selected = size === currentPageSize;
+      pageSize.appendChild(option);
+    });
+    toolbar.appendChild(pageSize);
+    toolbar.appendChild(makeEl("button", "observe-refresh-button", "Search"));
+    var clear = makeEl("button", "observe-refresh-button", "Clear");
+    clear.type = "button";
+    clear.disabled = !currentSearch;
+    clear.addEventListener("click", function () {
+      currentSearch = "";
+      currentPage = 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(clear);
+
+    var previous = makeEl("button", "observe-refresh-button", "Previous");
+    previous.type = "button";
+    previous.disabled = !(bounds && bounds.has_previous_page);
+    previous.addEventListener("click", function () {
+      currentPage = Math.max(1, currentPage - 1);
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(previous);
+
+    var status = makeEl(
+      "span",
+      "observe-page-status",
+      "Page " + String((bounds && bounds.page) || currentPage)
+    );
+    status.setAttribute("aria-live", "polite");
+    toolbar.appendChild(status);
+
+    var next = makeEl("button", "observe-refresh-button", "Next");
+    next.type = "button";
+    next.disabled = !(bounds && bounds.has_next_page);
+    next.addEventListener("click", function () {
+      currentPage += 1;
+      syncUrl();
+      fetchObserveData(false);
+    });
+    toolbar.appendChild(next);
+    return toolbar;
+  }
+
   function renderOverview(data, diagnostics) {
     var metrics = overviewMetricsFrom(data);
     if (!metrics.length) {
@@ -1727,10 +1961,12 @@ <h2 id="runs-heading">Runs</h2>
     setViewContent("overview", [grid]);
   }
 
-  function renderAgents(data, diagnostics) {
+  function renderAgents(data, diagnostics, bounds) {
     var agents = agentsFrom(data);
+    var notice = boundsNoticeNode(bounds, agents.length);
+    var controls = paginationToolbar(bounds);
     if (!agents.length) {
-      setViewContent("agents", [emptyStateNode("No data found for the selected filters.")]);
+      setViewContent("agents", [controls, notice, emptyStateNode("No data found for the selected filters.")]);
       return;
     }
     var rows = agents.map(function (agent) {
@@ -1809,7 +2045,7 @@ <h2 id="runs-heading">Runs</h2>
         "—"
       ]
     );
-    setViewContent("agents", [table]);
+    setViewContent("agents", [controls, notice, table]);
   }
 
   // ---------------------------------------------------------------------
@@ -2082,10 +2318,12 @@ <h2 id="runs-heading">Runs</h2>
       });
   }
 
-  function renderUsage(data, diagnostics) {
+  function renderUsage(data, diagnostics, bounds) {
     var usage = modelsFrom(data);
+    var notice = boundsNoticeNode(bounds, usage.length);
+    var controls = paginationToolbar(bounds);
     if (!usage.length) {
-      setViewContent("usage", [emptyStateNode("No data found for the selected filters.")]);
+      setViewContent("usage", [controls, notice, emptyStateNode("No data found for the selected filters.")]);
       return;
     }
     var rows = usage.map(function (entry) {
@@ -2159,14 +2397,16 @@ <h2 id="runs-heading">Runs</h2>
         "—"
       ]
     );
-    setViewContent("usage", [table]);
+    setViewContent("usage", [controls, notice, table]);
   }
 
   function renderTools(data, diagnostics, bounds) {
     var tools = toolsFrom(data);
     var notice = boundsNoticeNode(bounds, tools.length);
+    var controls = paginationToolbar(bounds);
     if (!tools.length) {
       setViewContent("tools", [
+        controls,
         notice,
         emptyStateNode("No tool activity was found for the selected filters. Tool attribution may not be reported for this selection."),
       ]);
@@ -2222,14 +2462,16 @@ <h2 id="runs-heading">Runs</h2>
         "—"
       ]
     );
-    setViewContent("tools", [notice, table]);
+    setViewContent("tools", [controls, notice, table]);
   }
 
   function renderRuns(data, diagnostics, bounds) {
     var runs = runsFrom(data);
     var notice = boundsNoticeNode(bounds, runs.length);
+    var controls = paginationToolbar(bounds);
     if (!runs.length) {
       setViewContent("runs", [
+        controls,
         notice,
         emptyStateNode("No runs could be correlated for the selected filters. Run correlation may not be reported for this selection."),
       ]);
@@ -2310,7 +2552,7 @@ <h2 id="runs-heading">Runs</h2>
         renderMaybeMissing(sumReported(runs, "reasoning_tokens"), { missingText: "—" })
       ]
     );
-    setViewContent("runs", [notice, table]);
+    setViewContent("runs", [controls, notice, table]);
   }
 
   function costLabel(value) {
@@ -3204,9 +3446,9 @@ <h2 id="runs-heading">Runs</h2>
     if (view === "overview") {
       renderOverview(body.data, body.diagnostics);
     } else if (view === "agents") {
-      renderAgents(body.data, body.diagnostics);
+      renderAgents(body.data, body.diagnostics, body.bounds);
     } else if (view === "usage") {
-      renderUsage(body.data, body.diagnostics);
+      renderUsage(body.data, body.diagnostics, body.bounds);
     } else if (view === "tools") {
       renderTools(body.data, body.diagnostics, body.bounds);
     } else if (view === "runs") {
@@ -3321,6 +3563,11 @@ <h2 id="runs-heading">Runs</h2>
         end: appliedFilters.end,
       },
       refresh: manual === true,
+      page: currentPage,
+      page_size: currentPageSize,
+      search: currentSearch || null,
+      sort_by: currentSortBy || null,
+      sort_direction: currentSortDirection,
     };
     payload = currentView === "cost" ? buildCostPayload(manual) : payload;
     if (currentView === "departments") {
@@ -3435,6 +3682,7 @@ <h2 id="runs-heading">Runs</h2>
           appliedFilters.attribution_cost_period_id = preservedAttributionFilters.attribution_cost_period_id;
           appliedFilters.attribution_cost_component_id = preservedAttributionFilters.attribution_cost_component_id;
         }
+        resetPaging(false);
         syncUrl();
         fetchObserveData(true);
       });
@@ -3511,7 +3759,9 @@ <h2 id="runs-heading">Runs</h2>
     document.querySelectorAll("[data-observe-nav-link]").forEach(function (link) {
       link.addEventListener("click", function (event) {
         event.preventDefault();
-        activateView(link.getAttribute("data-observe-nav-link"));
+        var nextView = link.getAttribute("data-observe-nav-link");
+        if (nextView !== currentView) resetPaging(true);
+        activateView(nextView);
         pushUrl();
         // Switching views queries a different `ObserveQuery.view`, so the
         // newly active view must be fetched -- otherwise it would only ever
diff --git a/tests/unit/__snapshots__/observe_styles.css b/tests/unit/__snapshots__/observe_styles.css
index ddaf748b..bf8f4a9c 100644
--- a/tests/unit/__snapshots__/observe_styles.css
+++ b/tests/unit/__snapshots__/observe_styles.css
@@ -304,12 +304,42 @@
   background: color-mix(in srgb, var(--observe-accent) 26%, transparent);
 }
 .observe-refresh-button:hover { border-color: var(--observe-border-strong); }
+.observe-refresh-button:disabled {
+  cursor: not-allowed;
+  opacity: 0.45;
+}
 .observe-apply-button:focus-visible,
 .observe-refresh-button:focus-visible {
   outline: 2px solid var(--observe-accent);
   outline-offset: 2px;
 }
 .observe-refresh-status { color: var(--observe-muted); font-size: 12px; }
+.observe-page-controls {
+  display: flex;
+  align-items: center;
+  flex-wrap: wrap;
+  gap: 8px;
+  margin: 0 0 8px;
+}
+.observe-page-controls input,
+.observe-page-controls select {
+  box-sizing: border-box;
+  min-height: 32px;
+  border: 1px solid var(--observe-border);
+  border-radius: 8px;
+  background: var(--observe-card-bg);
+  color: var(--observe-fg);
+  font: inherit;
+  font-size: 13px;
+  padding: 5px 9px;
+}
+.observe-page-controls input { min-width: 220px; }
+.observe-page-status {
+  color: var(--observe-muted);
+  font-size: 12px;
+  min-width: 48px;
+  text-align: center;
+}
 .observe-scope { margin: 0 0 10px; font-size: 12px; color: var(--observe-muted); }
 
 .observe-attribution-summary {
@@ -528,12 +558,43 @@ tfoot td {
 .observe-badge {
   display: inline-flex;
   align-items: center;
+  justify-content: center;
+  min-height: 24px;
+  box-sizing: border-box;
   border-radius: 999px;
-  padding: 1px 8px;
-  font-size: 11px;
-  font-weight: 600;
-  line-height: 1.6;
-  border: 1px solid currentColor;
+  padding: 3px 9px;
+  font-size: 12px;
+  font-weight: 700;
+  line-height: 1.25;
+  letter-spacing: 0.01em;
+  white-space: nowrap;
+  border: 1px solid var(--observe-border-strong);
+  background: var(--observe-card-hi);
+}
+.observe-badge.observe-tone-ok {
+  border-color: color-mix(in srgb, var(--observe-ok) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-ok) 16%, var(--observe-card-bg));
+  color: var(--observe-ok);
+}
+.observe-badge.observe-tone-warn {
+  border-color: color-mix(in srgb, var(--observe-warn) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-warn) 14%, var(--observe-card-bg));
+  color: var(--observe-warn);
+}
+.observe-badge.observe-tone-crit {
+  border-color: color-mix(in srgb, var(--observe-crit) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-crit) 14%, var(--observe-card-bg));
+  color: var(--observe-crit);
+}
+.observe-badge.observe-tone-info {
+  border-color: color-mix(in srgb, var(--observe-accent) 52%, var(--observe-border));
+  background: color-mix(in srgb, var(--observe-accent) 14%, var(--observe-card-bg));
+  color: var(--observe-accent);
+}
+.observe-badge.observe-tone-muted {
+  border-color: var(--observe-border-strong);
+  background: color-mix(in srgb, var(--observe-muted) 10%, var(--observe-card-bg));
+  color: var(--observe-muted);
 }
 .observe-tone-ok { color: var(--observe-ok); }
 .observe-tone-warn { color: var(--observe-warn); }
diff --git a/tests/unit/test_cockpit.py b/tests/unit/test_cockpit.py
index 7638ee96..747a5fc1 100644
--- a/tests/unit/test_cockpit.py
+++ b/tests/unit/test_cockpit.py
@@ -1675,6 +1675,11 @@ async def query(
         filters,
         refresh=False,
         user_context=None,
+        page=1,
+        page_size=50,
+        search=None,
+        sort_by=None,
+        sort_direction="desc",
     ):
         return {
             "view": view,
@@ -1710,6 +1715,71 @@ async def attribution(self, *, request, user_context=None):
         }
 
 
+def test_observe_query_forwards_paging_and_reports_server_timing(tmp_path: Path) -> None:
+    from fastapi.testclient import TestClient
+
+    from agentops.agent.cockpit import create_app
+
+    class TimingObserveService:
+        def __init__(self) -> None:
+            self.request: dict = {}
+
+        async def query(self, **kwargs):
+            self.request = kwargs
+            return {
+                "view": kwargs["view"],
+                "data": [],
+                "cache_status": "miss",
+                "diagnostics": {
+                    "discovery_duration_ms": 12.5,
+                    "query_duration_ms": 34.5,
+                    "normalization_duration_ms": 2.0,
+                },
+            }
+
+    service = TimingObserveService()
+    client = TestClient(
+        create_app(
+            tmp_path,
+            mode="local",
+            observe_scope={
+                "version": 1,
+                "mode": "projects",
+                "project_resource_ids": [
+                    "/subscriptions/sub/resourceGroups/rg/providers/"
+                    "Microsoft.CognitiveServices/accounts/a/projects/p"
+                ],
+            },
+            observe_service=service,
+        )
+    )
+
+    response = client.post(
+        "/api/observe/query",
+        json={
+            "view": "agents",
+            "filters": _OBSERVE_FILTERS,
+            "page": 3,
+            "page_size": 25,
+            "search": "planner",
+            "sort_by": "invocations",
+            "sort_direction": "asc",
+        },
+    )
+
+    assert response.status_code == 200
+    assert service.request["page"] == 3
+    assert service.request["page_size"] == 25
+    assert service.request["search"] == "planner"
+    assert service.request["sort_by"] == "invocations"
+    assert service.request["sort_direction"] == "asc"
+    timing = response.headers["server-timing"]
+    assert "total;dur=" in timing
+    assert "discovery;dur=12.5" in timing
+    assert "monitor;dur=34.5" in timing
+    assert "normalize;dur=2.0" in timing
+
+
 _OBSERVE_FILTERS = {
     "start": "2026-08-01T00:00:00Z",
     "end": "2026-09-01T00:00:00Z",
diff --git a/tests/unit/test_observe_adapters.py b/tests/unit/test_observe_adapters.py
index af95b2bf..11fa6c5d 100644
--- a/tests/unit/test_observe_adapters.py
+++ b/tests/unit/test_observe_adapters.py
@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import builtins
 import sys
 from datetime import datetime, timedelta, timezone
@@ -761,6 +762,50 @@ async def test_azure_query_client_chunks_more_than_ten_sources() -> None:
     assert all(result.status == "success" for result in results)
 
 
+@pytest.mark.asyncio
+async def test_azure_query_client_limits_chunk_concurrency_and_preserves_order() -> None:
+    class DelayedLogsClient(_FakeLogsClient):
+        def __init__(self) -> None:
+            super().__init__()
+            self.active = 0
+            self.max_active = 0
+
+        async def query_batch(self, requests):
+            batch_index = len(self.batches)
+            self.batches.append(list(requests))
+            self.active += 1
+            self.max_active = max(self.max_active, self.active)
+            await asyncio.sleep(0.01 * (3 - batch_index))
+            self.active -= 1
+            return [
+                SimpleNamespace(
+                    error=None,
+                    partial_error=None,
+                    tables=[],
+                    status="SUCCESS",
+                )
+                for _ in requests
+            ]
+
+    client = AzureQueryClient(
+        credential="fake-credential",
+        max_concurrent_batches=2,
+    )
+    logs_client = DelayedLogsClient()
+    client._logs_client = logs_client
+    sources = [
+        _make_source(f"source-{index}")
+        for index in range((MAX_SOURCES_PER_BATCH * 3) - 1)
+    ]
+
+    results = await client.query(sources, _make_filters(), view="overview")
+
+    assert logs_client.max_active == 2
+    assert [result.source_id for result in results] == [
+        source.source_id for source in sources
+    ]
+
+
 @pytest.mark.asyncio
 async def test_azure_query_client_skips_sources_without_workspace_id() -> None:
     client = AzureQueryClient(credential="fake-credential")
diff --git a/tests/unit/test_observe_cache.py b/tests/unit/test_observe_cache.py
index f75e1810..1698d17d 100644
--- a/tests/unit/test_observe_cache.py
+++ b/tests/unit/test_observe_cache.py
@@ -36,6 +36,20 @@ def test_ttl_identity_keying_refresh_and_eviction() -> None:
     assert cache.get("c") == 3
 
 
+def test_lookup_distinguishes_fresh_stale_and_miss() -> None:
+    clock = FakeClock()
+    cache = ObserveCache(ttl_seconds=10, clock=clock)
+    cache.set("key", {"value": 1})
+
+    assert cache.lookup("key").state == "fresh"
+    clock.now += 11
+    stale = cache.lookup("key", max_stale_seconds=5)
+    assert stale.state == "stale"
+    assert stale.value == {"value": 1}
+    clock.now += 5
+    assert cache.lookup("key", max_stale_seconds=5).state == "miss"
+
+
 @pytest.mark.parametrize(
     "value",
     [
diff --git a/tests/unit/test_observe_facade.py b/tests/unit/test_observe_facade.py
index 3edb6396..76b19427 100644
--- a/tests/unit/test_observe_facade.py
+++ b/tests/unit/test_observe_facade.py
@@ -828,6 +828,10 @@ async def test_query_tools_view_normalizes_rows_and_serializes_bounds() -> None:
         "rows_shown": 1,
         "rows_total_in_scope": 1,
         "truncated": False,
+        "page": 1,
+        "page_size": 50,
+        "has_previous_page": False,
+        "has_next_page": False,
     }
 
 
@@ -1287,6 +1291,10 @@ async def test_query_cost_preserves_explained_partial_and_missing_amount_states(
         "rows_shown": 2,
         "rows_total_in_scope": 3,
         "truncated": False,
+        "page": None,
+        "page_size": None,
+        "has_previous_page": False,
+        "has_next_page": False,
     }
     assert result["coverage"][0]["state"] == "partial"
     assert result["coverage"][0]["component_id"] == "gpt-ptu-prod"
@@ -2049,11 +2057,9 @@ def set(self, key, value):  # type: ignore[override]
     assert result["input_messages"] == ["hello"]
     assert FakeLogsQueryAdapter.instances, "expected the fake logs adapter to be used"
     assert FakeLogsQueryAdapter.instances[0].closed is True
-    # get_inventory (non-sensitive) may cache; the protected content itself
-    # must never be written through ObserveCache.set.
-    for call_args in cache.set_calls, cache._entries:  # pragma: no cover - sanity only
-        pass
-    assert cache.set_calls == 1  # only the inventory cache write from get_inventory
+    # Inventory uses a dedicated cache; protected content never writes to the
+    # shared aggregate cache.
+    assert cache.set_calls == 0
 
 
 @pytest.mark.asyncio
diff --git a/tests/unit/test_observe_models.py b/tests/unit/test_observe_models.py
index ac537b73..5c12b59c 100644
--- a/tests/unit/test_observe_models.py
+++ b/tests/unit/test_observe_models.py
@@ -251,16 +251,16 @@ def test_attribution_filter_tokens_are_nullable_trimmed_and_url_safe() -> None:
 
 def test_result_bounds_enforce_total_and_truncation_invariants() -> None:
     assert ResultBounds(rows_shown=0).rows_total_in_scope is None
-    assert ResultBounds(rows_shown=500, rows_total_in_scope=501, truncated=True)
+    assert ResultBounds(rows_shown=5000, rows_total_in_scope=5001, truncated=True)
 
     with pytest.raises(ValidationError, match="less than"):
         ResultBounds(rows_shown=2, rows_total_in_scope=1)
-    with pytest.raises(ValidationError, match="MAX_ROWS_PER_QUERY"):
-        ResultBounds(rows_shown=499, truncated=True)
+    with pytest.raises(ValidationError, match="page and page_size"):
+        ResultBounds(rows_shown=50, page=1)
     with pytest.raises(ValidationError):
         ResultBounds(rows_shown=1, unexpected=True)
     with pytest.raises(ValidationError):
-        ResultBounds(rows_shown=501)
+        ResultBounds(rows_shown=5001)
 
 
 def test_observe_api_requests_are_strict_and_canonical() -> None:
@@ -287,6 +287,9 @@ def test_observe_api_requests_are_strict_and_canonical() -> None:
     )
 
     assert query.refresh is False
+    assert query.page == 1
+    assert query.page_size == 50
+    assert query.sort_direction == "desc"
     assert "cost" in get_args(ObserveView)
     assert ObserveQueryRequest(view="cost", filters=filters).view == "cost"
     assert ObserveQueryRequest(view="tools", filters=filters).view == "tools"
@@ -297,6 +300,8 @@ def test_observe_api_requests_are_strict_and_canonical() -> None:
     assert content.source_resource_id == PROJECT.lower()
     with pytest.raises(ValidationError):
         ObserveQueryRequest(view="unknown", filters=filters)
+    with pytest.raises(ValidationError):
+        ObserveQueryRequest(view="agents", filters=filters, page_size=101)
     with pytest.raises(ValidationError):
         ObserveDrilldownRequest(
             view="tools",
diff --git a/tests/unit/test_observe_service.py b/tests/unit/test_observe_service.py
index c22662f6..5ebffc5f 100644
--- a/tests/unit/test_observe_service.py
+++ b/tests/unit/test_observe_service.py
@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal
@@ -17,6 +18,7 @@
 
 import pytest
 
+import agentops.agent.observe.service as observe_service_module
 from agentops.agent.observe.cache import ObserveCache, SensitiveValueError
 from agentops.agent.observe.queries import MAX_ROWS_PER_QUERY, SourceResult
 from agentops.agent.observe.service import (
@@ -53,11 +55,18 @@
     derive_pseudonymous_user_key,
     issue_user_filter_token,
 )
-from agentops.core.cost import CostComponent, CostModel, CostPeriod, CostUsageObservation
+from agentops.core.cost import (
+    MAX_COST_ROWS,
+    CostComponent,
+    CostModel,
+    CostPeriod,
+    CostUsageObservation,
+)
 from agentops.core.observe import (
     AttributionQueryRequest,
     ModelUsage,
     ObserveFilterState,
+    ObserveQueryRequest,
     ObserveScope,
     ResourceInventory,
     TelemetrySource,
@@ -802,7 +811,8 @@ def test_result_bounds_combines_truncated_multi_source_totals() -> None:
     first = SourceResult(
         source_id="src-a",
         status="success",
-        tables=[{"total_in_scope": 600}] * MAX_ROWS_PER_QUERY,
+        tables=[{"total_in_scope": MAX_ROWS_PER_QUERY + 100}]
+        * MAX_ROWS_PER_QUERY,
     )
     second = SourceResult(
         source_id="src-b",
@@ -812,7 +822,7 @@ def test_result_bounds_combines_truncated_multi_source_totals() -> None:
 
     bounds = _result_bounds([first, second], rows_shown=MAX_ROWS_PER_QUERY)
 
-    assert bounds.rows_total_in_scope == 610
+    assert bounds.rows_total_in_scope == MAX_ROWS_PER_QUERY + 110
     assert bounds.rows_shown == MAX_ROWS_PER_QUERY
     assert bounds.truncated is True
 
@@ -1554,12 +1564,12 @@ async def test_query_cost_keeps_omitted_amounts_when_agent_rows_are_bounded() ->
     assert all(summary.rows_total == 501 for summary in result.data.components)
     assert (
         sum(summary.rows_shown for summary in result.data.components)
-        == MAX_ROWS_PER_QUERY
+        == MAX_COST_ROWS
     )
     assert all(
         summary.omitted_allocated_amount > 0 for summary in result.data.components
     )
-    assert len(result.data.rows) == MAX_ROWS_PER_QUERY
+    assert len(result.data.rows) == MAX_COST_ROWS
 
 
 @pytest.mark.asyncio
@@ -2404,6 +2414,178 @@ async def test_query_view_caches_for_two_minutes_and_serves_hits_without_requery
     assert second.refreshed_at == first.refreshed_at
 
 
+@pytest.mark.asyncio
+async def test_query_view_pages_searches_and_sorts_one_cached_aggregate() -> None:
+    clock = FakeDatetimeClock(datetime(2024, 1, 1, tzinfo=timezone.utc))
+    inventory = _inventory([_source()])
+    rows = [
+        {
+            "agent_key": f"agent-{index:03d}",
+            "agent_id": f"agent-{index:03d}",
+            "agent_name": f"Agent {index:03d}",
+            "model": "gpt-4o",
+            "invocations": index,
+            "failures": 0,
+            "last_seen": datetime(2024, 1, 1, 8, tzinfo=timezone.utc),
+            "total_in_scope": 120,
+        }
+        for index in range(120)
+    ]
+    service, _discovery, query = _service(
+        inventory=inventory,
+        results=[SourceResult(source_id="src-1", status="success", tables=rows)],
+        clock=clock,
+    )
+
+    second_page = await service.query_view(
+        _scope(),
+        _filters(),
+        view="agents",
+        page=2,
+        page_size=50,
+    )
+    filtered = await service.query_view(
+        _scope(),
+        _filters(),
+        view="agents",
+        page_size=25,
+        search="Agent 11",
+        sort_by="agent_name",
+        sort_direction="asc",
+    )
+
+    assert len(query.calls) == 1
+    assert [row.agent_id for row in second_page.data[:2]] == [
+        "agent-069",
+        "agent-068",
+    ]
+    assert second_page.bounds is not None
+    assert second_page.bounds.model_dump() == {
+        "rows_shown": 50,
+        "rows_total_in_scope": 120,
+        "truncated": False,
+        "page": 2,
+        "page_size": 50,
+        "has_previous_page": True,
+        "has_next_page": True,
+    }
+    assert [row.agent_id for row in filtered.data] == [
+        f"agent-{index:03d}" for index in range(110, 120)
+    ]
+    assert filtered.bounds is not None
+    assert filtered.bounds.rows_total_in_scope == 10
+
+
+def test_every_allowlisted_view_sort_field_satisfies_request_contract() -> None:
+    for view, fields in observe_service_module._VIEW_SORT_FIELDS.items():
+        for sort_field in fields:
+            request = ObserveQueryRequest(
+                view=view,
+                filters=_filters(),
+                sort_by=sort_field,
+            )
+            assert request.sort_by == sort_field
+
+
+@pytest.mark.asyncio
+async def test_truncated_paging_stops_at_last_materialized_row(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(observe_service_module, "MAX_ROWS_PER_QUERY", 3)
+    rows = [
+        {
+            "agent_key": f"agent-{index}",
+            "agent_id": f"agent-{index}",
+            "agent_name": f"Agent {index}",
+            "model": "gpt-4o",
+            "invocations": index,
+            "failures": 0,
+            "last_seen": datetime(2024, 1, 1, 8, tzinfo=timezone.utc),
+            "total_in_scope": 5,
+        }
+        for index in range(5)
+    ]
+    service, _discovery, query = _service(
+        inventory=_inventory([_source()]),
+        results=[SourceResult(source_id="src-1", status="success", tables=rows)],
+        clock=FakeDatetimeClock(datetime(2024, 1, 1, tzinfo=timezone.utc)),
+    )
+
+    first_page = await service.query_view(
+        _scope(),
+        _filters(),
+        view="agents",
+        page=1,
+        page_size=2,
+    )
+    last_page = await service.query_view(
+        _scope(),
+        _filters(),
+        view="agents",
+        page=2,
+        page_size=2,
+    )
+    filtered_last_page = await service.query_view(
+        _scope(),
+        _filters(),
+        view="agents",
+        page=2,
+        page_size=2,
+        search="Agent",
+    )
+
+    assert len(query.calls) == 1
+    assert first_page.bounds is not None
+    assert first_page.bounds.has_next_page is True
+    assert last_page.bounds is not None
+    assert last_page.bounds.rows_shown == 1
+    assert last_page.bounds.rows_total_in_scope == 5
+    assert last_page.bounds.truncated is True
+    assert last_page.bounds.has_next_page is False
+    assert filtered_last_page.bounds is not None
+    assert filtered_last_page.bounds.rows_total_in_scope is None
+    assert filtered_last_page.bounds.has_next_page is False
+
+
+@pytest.mark.asyncio
+async def test_query_view_coalesces_concurrent_identical_misses() -> None:
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    class BlockingQueryClient(FakeQueryClient):
+        async def query(self, sources, filters, *, view):
+            self.calls.append((tuple(source.source_id for source in sources), view))
+            entered.set()
+            await release.wait()
+            return self.results
+
+    inventory = _inventory([_source()])
+    discovery = FakeDiscoveryClient(inventory)
+    query = BlockingQueryClient([_agent_rows_result()])
+    service = ObserveService(
+        discovery_client=discovery,
+        query_client=query,
+        runtime=FakeRuntime(),
+        clock=FakeDatetimeClock(datetime(2024, 1, 1, tzinfo=timezone.utc)),
+        cache=ObserveCache(ttl_seconds=CACHE_TTL_SECONDS),
+    )
+
+    first_task = asyncio.create_task(
+        service.query_view(_scope(), _filters(), view="agents")
+    )
+    await entered.wait()
+    second_task = asyncio.create_task(
+        service.query_view(_scope(), _filters(), view="agents")
+    )
+    await asyncio.sleep(0)
+    assert len(query.calls) == 1
+    release.set()
+    first, second = await asyncio.gather(first_task, second_task)
+
+    assert {first.cache_status, second.cache_status} == {"miss", "hit"}
+    assert first.data == second.data
+
+
 @pytest.mark.asyncio
 async def test_query_view_refresh_bypasses_cache_and_requeries() -> None:
     clock = FakeDatetimeClock(
@@ -2425,7 +2607,7 @@ async def test_query_view_refresh_bypasses_cache_and_requeries() -> None:
 
 
 @pytest.mark.asyncio
-async def test_query_view_expires_after_ttl() -> None:
+async def test_query_view_serves_stale_after_ttl_and_refreshes_in_background() -> None:
     clock = FakeDatetimeClock(datetime(2024, 1, 1, tzinfo=timezone.utc))
     cache = ObserveCache(ttl_seconds=CACHE_TTL_SECONDS, clock=lambda: cache_clock.now)
     cache_clock = FakeDatetimeClock(
@@ -2449,8 +2631,10 @@ def __call__(self) -> float:
 
     await service.query_view(_scope(), _filters(), view="agents")
     float_clock.value += CACHE_TTL_SECONDS + 1
-    await service.query_view(_scope(), _filters(), view="agents")
+    stale = await service.query_view(_scope(), _filters(), view="agents")
+    await asyncio.gather(*service._background_tasks)
 
+    assert stale.cache_status == "stale"
     assert len(query.calls) == 2
 
 
diff --git a/tests/unit/test_observe_ui.py b/tests/unit/test_observe_ui.py
index 50f16b42..d4c018da 100644
--- a/tests/unit/test_observe_ui.py
+++ b/tests/unit/test_observe_ui.py
@@ -708,6 +708,22 @@ def test_render_agents_table_shows_source_kind_and_identity_badges() -> None:
     assert "source-a" in html
 
 
+def test_observe_badges_use_filled_high_contrast_treatment() -> None:
+    css = ui._OBSERVE_STYLES
+    assert "min-height: 24px;" in css
+    assert "font-size: 12px;" in css
+    assert "font-weight: 700;" in css
+    assert ".observe-badge.observe-tone-ok {" in css
+    assert (
+        "background: color-mix(in srgb, var(--observe-ok) 16%, "
+        "var(--observe-card-bg));"
+    ) in css
+    assert (
+        "border-color: color-mix(in srgb, var(--observe-ok) 52%, "
+        "var(--observe-border));"
+    ) in css
+
+
 def test_render_agents_table_missing_agent_id_shows_identity_unavailable() -> None:
     html = ui.render_agents_table([_agent(agent_id=None, agent_name=None)])
     assert "Identity unavailable" in html
@@ -2135,7 +2151,7 @@ def test_script_apply_button_is_the_only_place_draft_becomes_applied() -> None:
     script = ui._OBSERVE_SCRIPT
     assert "appliedFilters = draftFilters;" in script
     # Ensure this assignment only happens inside the submit handler block.
-    submit_block = script.split('addEventListener("submit"')[1]
+    submit_block = script.split('form.addEventListener("submit"')[1]
     assert "appliedFilters = draftFilters;" in submit_block
 
 
@@ -2174,6 +2190,27 @@ def test_script_observe_query_payload_matches_observe_query_schema() -> None:
     assert "start: appliedFilters.start" in payload_block
     assert "end: appliedFilters.end" in payload_block
     assert "refresh: manual === true" in payload_block
+    assert "page: currentPage" in payload_block
+    assert "page_size: currentPageSize" in payload_block
+    assert "search: currentSearch || null" in payload_block
+    assert "sort_by: currentSortBy || null" in payload_block
+    assert "sort_direction: currentSortDirection" in payload_block
+
+
+def test_script_pages_large_views_and_round_trips_safe_state_in_url() -> None:
+    script = ui._OBSERVE_SCRIPT
+    assert "function paginationToolbar(bounds)" in script
+    assert "No rows are available on page " in script
+    assert " from the highest-ranked results" in script
+    assert '"Search this view"' in script
+    assert '"Previous"' in script
+    assert '"Next"' in script
+    assert 'params.set("page", String(currentPage))' in script
+    assert 'params.set("page_size", String(currentPageSize))' in script
+    assert 'params.set("search", currentSearch)' in script
+    assert 'params.set("sort_by", currentSortBy)' in script
+    assert "SERVER_SORT_KEYS" in script
+    assert '"p95 latency": "p95_latency_ms"' in script
 
 
 def test_script_manual_refresh_sets_refresh_true_and_auto_refresh_sets_refresh_false() -> None:
@@ -2185,7 +2222,7 @@ def test_script_manual_refresh_sets_refresh_true_and_auto_refresh_sets_refresh_f
     assert "fetchObserveData(false);" in script
     refresh_button_block = script.split('getElementById("observe-refresh-now")')[1].split("}")[0]
     assert "fetchObserveData(true);" in refresh_button_block
-    submit_block = script.split('addEventListener("submit"')[1].split("});")[0]
+    submit_block = script.split('form.addEventListener("submit"')[1].split("});")[0]
     assert "fetchObserveData(true);" in submit_block
 
 
@@ -2275,7 +2312,9 @@ def test_script_runtime_kind_tones_mirror_all_six_python_badges() -> None:
 
 def test_script_tools_and_runs_render_sources_bounds_and_explained_empty_states() -> None:
     script = ui._OBSERVE_SCRIPT
-    agents_block = script.split("function renderAgents(data, diagnostics) {")[1].split("\n  }\n")[0]
+    agents_block = script.split("function renderAgents(data, diagnostics, bounds) {")[1].split(
+        "\n  }\n"
+    )[0]
     assert 'sourceCell.title = agent.source_id || ""' in agents_block
     for function_name, source_field, empty_copy in (
         (
@@ -2326,8 +2365,8 @@ def test_script_defines_response_parsing_and_render_dispatch_functions() -> None
     for name in (
         "function renderObserveResponse(body)",
         "function renderOverview(data, diagnostics)",
-        "function renderAgents(data, diagnostics)",
-        "function renderUsage(data, diagnostics)",
+        "function renderAgents(data, diagnostics, bounds)",
+        "function renderUsage(data, diagnostics, bounds)",
         "function renderTools(data, diagnostics, bounds)",
         "function renderRuns(data, diagnostics, bounds)",
         "function internalViewFromWire(view)",
@@ -2377,8 +2416,8 @@ def test_script_render_dispatch_routes_each_view_to_its_own_renderer_and_field()
     )[0]
     assert "var view = internalViewFromWire(body.view) || currentView;" in dispatch_block
     assert "renderOverview(body.data, body.diagnostics);" in dispatch_block
-    assert "renderAgents(body.data, body.diagnostics);" in dispatch_block
-    assert "renderUsage(body.data, body.diagnostics);" in dispatch_block
+    assert "renderAgents(body.data, body.diagnostics, body.bounds);" in dispatch_block
+    assert "renderUsage(body.data, body.diagnostics, body.bounds);" in dispatch_block
     assert "renderTools(body.data, body.diagnostics, body.bounds);" in dispatch_block
     assert "renderRuns(body.data, body.diagnostics, body.bounds);" in dispatch_block
     assert "renderCoverage(" not in dispatch_block
@@ -2475,7 +2514,8 @@ def test_script_nav_link_click_fetches_the_newly_selected_view() -> None:
     nav_block = script.split(
         'document.querySelectorAll("[data-observe-nav-link]").forEach(function (link) {'
     )[2].split("    });", 1)[0]
-    assert "activateView(link.getAttribute" in script
+    assert 'var nextView = link.getAttribute("data-observe-nav-link");' in nav_block
+    assert "activateView(nextView);" in nav_block
     assert "event.preventDefault();" in script
     assert "pushUrl();" in nav_block
     assert "fetchObserveData(false);" in nav_block
@@ -2588,7 +2628,7 @@ def test_script_defines_agent_detail_functions() -> None:
 
 def test_script_renders_agents_table_with_details_column_and_button() -> None:
     script = ui._OBSERVE_SCRIPT
-    fn_block = script.split("function renderAgents(data, diagnostics) {")[1].split(
+    fn_block = script.split("function renderAgents(data, diagnostics, bounds) {")[1].split(
         "\n  }\n"
     )[0]
     assert '"Details"' in fn_block