Release: provider-filter + memory diagnostics (#6338 #6363) (#6994)

* fix(#6335): filter unconfigured providers from model selector

When a known provider (e.g. openai-api) is listed in config.yaml's
providers section with only metadata (name, display_name) and no route
configuration or credentials, it should not appear in the chat model
selector dropdown.

Previously, any provider recognised as a 'known provider' (present in
_PROVIDER_MODELS or _PROVIDER_DISPLAY) would be admitted to the picker
simply by being listed in the config.yaml providers block — even after
the user removed its API key.  This caused a persistent ghost provider
entry in the dropdown (#6335).

The fix gates known-provider admission on credential evidence: the
provider must already have been detected from a credential source (env
vars, hermes auth, or credential pool) OR carry explicit route
configuration (api/base_url/api_key/key_env) OR be the active provider
with a configured models allowlist.

* fix(#6338): normalize provider aliases in credential check

The credential check compared _canonical (config key like 'x-ai',
'google') directly against detected_providers (alias-resolved values
like 'xai', 'gemini').  Alias-named authenticated providers were
therefore incorrectly treated as unconfigured and removed from the
model selector.

Fix: resolve _canonical through _resolve_provider_alias() before the
membership test, so config keys like 'x-ai' match credential evidence
reported as 'xai'.  Also check the raw canonical for cases where the
config key is itself the alias target.

Refs: #6335, #6338

* fix(#6338): normalize detected_providers through alias resolver

Normalize both _canonical and detected_providers entries through
_resolve_provider_alias before the membership test, so that a WebUI
canonical form in detected_providers (e.g. x-ai added by a prior
loop iteration) also matches the alias-resolved form used in
credential detection (e.g. xai from hermes_cli).

This is belt-and-braces on the existing _resolve_provider_alias(_canonical)
check: both sides now map into the same alias-resolved namespace,
closing any edge case where detected_providers contains a WebUI-canonical
form that happens to differ from the alias target.

* test(ui): catalog-admission regression for credentialed providers (#6338)

* test(ui): pin config path state in catalog-admission regression (#6338)

* fix(system-health): expose runtime memory diagnostics without blocking owner locks (#6351)

* fix(system-health): keep owner contention local to runtime status (#6351)

* docs(changelog): stamp #6338 #6363 (certified batch)

---------

Co-authored-by: webtecnica <webtecnica@users.noreply.github.com>
Co-authored-by: webtecnica <webtecnica@gmail.com>
Co-authored-by: Rod Boev <rod.boev@gmail.com>
Co-authored-by: nesquena-hermes <nesquena-hermes@users.noreply.github.com>

CHANGELOG.md

@@ -7,6 +7,8 @@
 
 - **The chat view no longer bounces vertically while a reply streams in.** As each word faded in during streaming, the browser replaced the finished word's DOM node with a fresh text node — and every replacement made the browser's native scroll anchoring re-pick an anchor while the transcript was still growing, producing a visible vertical bounce. The two fade-cleanup handlers now keep each faded-in word's node stable (they just clear the transient animation class and timing variable) instead of replacing it, so the scroll position stays put. The settled reply is still rebuilt as plain text exactly as before, and text selection during streaming is steadier as a bonus. Thanks @ruizanthony. (#6257)
 
+- **The model selector no longer lists providers whose API key you've removed.** A provider that stayed in `config.yaml` with only metadata (e.g. `openai-api: {name: "OpenAI API"}`) kept appearing in the chat model dropdown after its credential was removed, so picking it failed. A config-listed provider now appears only when it was actually detected from a credential source (env var, hermes auth, or the credential pool), with provider aliases resolved on both sides (so `x-ai`/`xai`, `google`/`gemini` match correctly). Route-configured and active-model providers are unaffected. Thanks @webtecnica. (#6338)
+
 - **Links ending in Chinese/Japanese full-width punctuation no longer swallow the punctuation into the URL.** When an assistant wrote a URL followed immediately by a CJK full-width mark (`）。，；：！？、`) — common in Chinese/Japanese prose — the autolink pass treated the mark as part of the link, so the rendered href included the trailing punctuation (and often 404'd). Both autolink passes now strip a single trailing CJK full-width mark the same way they already strip ASCII `.,;:!?)`, and a full-width close-paren also terminates the URL. ASCII behavior is unchanged. Thanks @happy5318. (#6792)
 
 - **Mobile: create forms opened from the sidebar drawer are no longer hidden behind it.** On phones the sidebar is a full-screen drawer over the main view; opening a new cron/skill/workspace/profile form from inside it left the drawer covering the form, so the tap looked like a no-op. The panel re-open on a rail tap now happens synchronously (before the panel's async data load, so a form that closes the drawer isn't re-covered), and the four create-form openers now close the drawer the same way their matching detail views already did. Desktop is unaffected. Thanks @silent-reader-cn. (#6921)
@@ -32,6 +34,8 @@
 
 ### Changed
 
+- **The System/health panel now reports WebUI runtime memory diagnostics.** Beyond host CPU/RAM/disk, the system-health response now observes the process-owned caches and stream structures named in #6351 (session-list cache, agent cache, live-stream structures) by reading their existing owners non-blockingly — a collector that never waits on an owner lock, so it can't stall a request. This surfaces sizes/bounds that were previously invisible for diagnosing runaway memory. Thanks @rodboev. (#6363)
+
 - **Slash-command autocomplete now searches skills by keyword.** Typing `/` and a term now matches plain skills on a case-insensitive keyword in their **name or description** (previously only a name prefix), so a skill is discoverable by what it does, not just its exact slug. Built-in, agent/plugin, and bundle commands keep prefix matching and always take precedence over a same-slug skill, and skill suggestions are held until the bundle-metadata request settles so a slow response can't briefly surface a shadowed skill. Thanks @MinhoJJang. (#6932)
 
 - **Extensions can observe turn lifecycle events via `handle.events.on(...)`.** A registered extension (see `window.hermesExt.register`) can now subscribe to session turn-lifecycle events — `turn:start`, `turn:complete`, `turn:error`, and `turn:cancel` — through a frozen `events` accessor on its handle, each delivering a frozen scalar event (`sessionId`, `streamId`, `status`, and start/end timestamps; no message content). Core owns the authoritative live-session stream and emits each event exactly once per real transition (a per-session/stream state machine suppresses duplicate starts and post-terminal events, including on reconnect/replay), with per-listener error isolation so a misbehaving extension cannot break streaming. Subscriptions return an idempotent unsubscribe. This is the B1 observer surface on top of the E0 identity handle — a stable page-local facade instead of extensions polling private DOM state. Thanks @franksong2702. (#6924)

api/config.py

@@ -6960,7 +6960,30 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa
                     and _provider_cfg["models"]
                     and _canonical == _canonicalise_provider_id(active_provider)
                 )
-                if not (_is_known_provider or _has_provider_route or _has_models_only_active_route):
+                # A known provider listed in config.yaml without route
+                # configuration should only appear in the picker when it was
+                # already detected from credential sources (env vars, hermes
+                # auth, credential pool).  Otherwise a provider with
+                # metadata-only entries in config.yaml (e.g.
+                # ``openai-api: {name: "OpenAI API"}``) would still render
+                # in the model selector after the API key is removed (#6335).
+                # Resolve provider aliases on both sides so an alias-named
+                # config key (e.g. ``x-ai`` in providers, ``google`` in
+                # config.yaml) matches credential evidence reported under the
+                # agent's canonical alias (``xai``, ``gemini``) (#6338).
+                # Normalise detected_providers entries into the same
+                # alias-resolved namespace as _canonical so that a WebUI
+                # canonical form in detected_providers (e.g. ``x-ai`` added
+                # by a prior loop iteration) also matches (#6338).
+                _resolved_detected = {
+                    _resolve_provider_alias(_pid) for _pid in detected_providers
+                }
+                _already_credentialed = (
+                    _resolve_provider_alias(_canonical) in _resolved_detected
+                    or _canonical in _resolved_detected
+                )
+                _admit_as_known = _is_known_provider and _already_credentialed
+                if not (_admit_as_known or _has_provider_route or _has_models_only_active_route):
                     continue
 
                 _canonical_to_raw_provider_key.setdefault(_canonical, _pid_key)
@@ -8582,6 +8605,10 @@ def get_sessions_cache_max(config_data: dict | None = None) -> int:
     ``HERMES_WEBUI_SESSIONS_MAX`` env override, then to
     ``DEFAULT_SESSIONS_CACHE_MAX`` — a typo can never disable the bound and
     reintroduce unbounded memory growth.
+
+    This is the sole resolution authority and it is side-effect free. The cap
+    diagnostics report is published by the code that enforces it; see
+    ``_LAST_APPLIED_SESSIONS_CACHE_MAX`` below.
     """
     active_cfg = config_data if isinstance(config_data, dict) else get_config()
     webui_cfg = active_cfg.get("webui", {}) if isinstance(active_cfg, dict) else {}
@@ -8590,7 +8617,10 @@ def get_sessions_cache_max(config_data: dict | None = None) -> int:
         if raw is not None:
             try:
                 value = int(raw)
-            except (TypeError, ValueError):
+            except (TypeError, ValueError, OverflowError):
+                # OverflowError covers YAML's float infinities (`.inf`, `1e400`),
+                # which safe_load resolves to a real float. Without it a typo
+                # would escape the fallback and raise out of every caller.
                 value = None
             if value is not None and value >= 1:
                 return value
@@ -8599,6 +8629,16 @@ def get_sessions_cache_max(config_data: dict | None = None) -> int:
     if isinstance(SESSIONS_MAX, int) and SESSIONS_MAX >= 1:
         return SESSIONS_MAX
     return DEFAULT_SESSIONS_CACHE_MAX
+
+
+# The cap api/models.py::_evict_sessions_over_cap() last enforced. That function
+# publishes it after its own fallback and range normalization, so a nonblocking
+# diagnostics read reports what eviction applied without re-entering config or
+# profile I/O, and nothing else writes this field. Seeded from the config
+# reload_config() already loaded at import (see `cfg` above) through the getter's
+# dict mode, which reads no file and takes no lock, so the value is right before
+# the first eviction pass instead of after it.
+_LAST_APPLIED_SESSIONS_CACHE_MAX: int = get_sessions_cache_max(cfg)
 CHAT_LOCK = threading.Lock()
 
 
@@ -8814,19 +8854,39 @@ class StreamChannel:
                 self._SUBSCRIBER_QUEUE_MAXSIZE,
             )
 
+    def _diagnostic_counters_locked(self) -> dict[str, object]:
+        """Return the counter dict. CALLER CONTRACT: ``self._lock`` is held."""
+        return {
+            "subscriber_count": len(self._subscribers),
+            "offline_buffered_events": len(self._offline_buffer),
+            # Cumulative over the channel lifetime (ops visibility), vs. the
+            # per-cycle count subscribe_with_snapshot() reports for truncation.
+            "offline_dropped_events": self._offline_dropped_total,
+            # Cumulative per-subscriber queue drops (replay + broadcast) over
+            # the channel lifetime — surfaces slow/backpressured tabs.
+            "subscriber_dropped_events": self._subscriber_dropped_total,
+        }
+
     def diagnostic_snapshot(self) -> dict[str, object]:
         """Return non-sensitive stream observation counters for health checks."""
         with self._lock:
-            return {
+            return self._diagnostic_counters_locked()
-                "subscriber_count": len(self._subscribers),
+
-                "offline_buffered_events": len(self._offline_buffer),
+    def try_diagnostic_snapshot(self) -> dict[str, object] | None:
-                # Cumulative over the channel lifetime (ops visibility), vs. the
+        """Return the same counters without waiting, or ``None`` when busy.
-                # per-cycle count subscribe_with_snapshot() reports for truncation.
+
-                "offline_dropped_events": self._offline_dropped_total,
+        An aggregate health poll must never stall behind one channel's producer
-                # Cumulative per-subscriber queue drops (replay + broadcast) over
+        or subscriber work, so a contended channel is reported as unavailable
-                # the channel lifetime — surfaces slow/backpressured tabs.
+        instead of waited on. ``diagnostic_snapshot()`` keeps its blocking
-                "subscriber_dropped_events": self._subscriber_dropped_total,
+        contract for the per-stream ``/health?deep=1`` view, which needs the
-            }
+        counters of every stream rather than a best-effort aggregate.
+        """
+        if not self._lock.acquire(blocking=False):
+            return None
+        try:
+            return self._diagnostic_counters_locked()
+        finally:
+            self._lock.release()
 
 
 def create_stream_channel() -> StreamChannel:
@@ -9949,6 +10009,70 @@ if _settings_file_exists:
 # ── SESSIONS in-memory cache (LRU OrderedDict) ───────────────────────────────
 SESSIONS: collections.OrderedDict = collections.OrderedDict()
 
+
+def get_runtime_diagnostics_snapshot() -> dict[str, dict[str, object]]:
+    """Return nonblocking scalar observations owned by the config module."""
+    result = {
+        "sessions": {"available": False, "resident": 0, "cap": 0},
+        "models_cache": {
+            "available": False,
+            "groups": 0,
+            "models": 0,
+            "age_seconds": None,
+        },
+    }
+    try:
+        if LOCK.acquire(blocking=False):
+            try:
+                # Held-section discipline: len(), arithmetic, and owner-held
+                # scalars only. Never call anything here that can resolve config
+                # or a profile, touch the filesystem, import a module, or wait on
+                # another lock — the cap is the scalar _evict_sessions_over_cap()
+                # published, precisely so this section stays leaf-nonblocking.
+                result["sessions"] = {
+                    "available": True,
+                    "resident": max(0, int(len(SESSIONS))),
+                    "cap": max(0, int(_LAST_APPLIED_SESSIONS_CACHE_MAX)),
+                }
+            finally:
+                LOCK.release()
+    except Exception:
+        pass
+    try:
+        if _available_models_cache_lock.acquire(blocking=False):
+            try:
+                # Same held-section discipline: len(), isinstance, float(), and
+                # time.monotonic() only. _available_models_cache_lock is an RLock
+                # (see its definition), so a nonblocking acquire from a thread
+                # that already holds it would report available mid-build; safe
+                # here because health collection is never nested inside a
+                # catalog build, and nothing may be added that changes that.
+                snapshot = _available_models_cache
+                groups = snapshot.get("groups") if isinstance(snapshot, dict) else None
+                group_count = len(groups) if isinstance(groups, list) else 0
+                model_count = 0
+                if isinstance(groups, list):
+                    for group in groups:
+                        if isinstance(group, dict):
+                            for bucket in ("models", "extra_models"):
+                                models = group.get(bucket)
+                                if isinstance(models, list):
+                                    model_count += len(models)
+                age = None
+                if snapshot is not None and _available_models_cache_ts:
+                    age = max(0.0, time.monotonic() - float(_available_models_cache_ts))
+                result["models_cache"] = {
+                    "available": True,
+                    "groups": max(0, int(group_count)),
+                    "models": max(0, int(model_count)),
+                    "age_seconds": age,
+                }
+            finally:
+                _available_models_cache_lock.release()
+    except Exception:
+        pass
+    return result
+
 # ── Profile state initialisation ────────────────────────────────────────────
 # Must run after all imports are resolved to correctly patch module-level caches
 try:

api/models.py

@@ -4624,6 +4624,10 @@ def _evict_sessions_over_cap(cap: int | None = None) -> int:
     CALLER CONTRACT: the global ``LOCK`` MUST already be held (every call site
     mutates ``SESSIONS`` under ``LOCK``). This function never acquires ``LOCK``
     or any stream lock itself, so it cannot introduce a lock-ordering deadlock.
+    Under that same held ``LOCK`` it publishes the cap it enforced into
+    ``api.config._LAST_APPLIED_SESSIONS_CACHE_MAX`` for nonblocking diagnostics
+    (#6351); any future edit that can change ``cap`` after that point must move
+    the publish down with it.
 
     Returns the number of sessions evicted. If every over-cap candidate is
     active/unsaved, the cache may temporarily exceed ``cap`` — that is the
@@ -4636,6 +4640,11 @@ def _evict_sessions_over_cap(cap: int | None = None) -> int:
             cap = SESSIONS_MAX
     if not isinstance(cap, int) or cap < 1:
         cap = SESSIONS_MAX if isinstance(SESSIONS_MAX, int) and SESSIONS_MAX >= 1 else 1
+    # Diagnostics owns the field; this function owns the decision. Publishing the
+    # normalized cap here, rather than from the resolver, is what makes the health
+    # payload report a cap eviction actually applied — including the getter-failure
+    # fallback and explicit/normalized calls, which never reach the resolver (#6351).
+    _cfg._LAST_APPLIED_SESSIONS_CACHE_MAX = cap
     evicted = 0
     # Iterate over a snapshot of ids in LRU order (oldest first). We stop as
     # soon as we are at/below the cap. Skipping a non-evictable oldest entry and

api/route_session_list_cache.py

@@ -44,6 +44,41 @@ _SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION = 0
 _SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION: dict[str, int] = {}
 
 
+def get_session_list_cache_snapshot() -> dict[str, object]:
+    """Return scalar cache occupancy without waiting or changing LRU state.
+
+    Held-section discipline: between the nonblocking acquire and the release,
+    only ``len()`` and module-constant reads are permitted. Nothing that can
+    resolve config, resolve a profile, touch the filesystem, import a module, or
+    wait on another lock may be added here. ``_SESSIONS_CACHE_LOCK`` is an
+    ``RLock``, so a nonblocking acquire from a thread already holding it would
+    report available mid-mutation; the health collector is this helper's only
+    caller and never runs nested inside a cache rebuild.
+    """
+    result = {
+        "available": False,
+        "entries": 0,
+        "inflight_rebuilds": 0,
+        "cap": 0,
+    }
+    acquired = False
+    try:
+        acquired = _SESSIONS_CACHE_LOCK.acquire(blocking=False)
+        if not acquired:
+            return result
+        return {
+            "available": True,
+            "entries": max(0, int(len(_SESSIONS_CACHE))),
+            "inflight_rebuilds": max(0, int(len(_SESSIONS_CACHE_INFLIGHT))),
+            "cap": max(0, int(_SESSIONS_CACHE_MAX_ENTRIES)),
+        }
+    except Exception:
+        return result
+    finally:
+        if acquired:
+            _SESSIONS_CACHE_LOCK.release()
+
+
 def _session_list_cache_session_dir() -> Path:
     try:
         import api.routes as _routes

api/streaming.py

@@ -85,6 +85,61 @@ from api.process_event_utils import (
 )
 
 
+def get_stream_runtime_snapshot() -> dict[str, object]:
+    """Return aggregate stream observations without waiting on the registry.
+
+    The registry is copied under a nonblocking ``STREAMS_LOCK`` acquire and the
+    lock is released before any channel lock is touched, so registry and channel
+    locks are never nested. Each channel is then read through its nonblocking
+    owner method: a channel busy with its own producer or subscriber work counts
+    as one unavailable channel and the loop keeps summing its siblings.
+    """
+    result = {
+        "available": False,
+        "active": 0,
+        "agent_instances": 0,
+        "subscribers": 0,
+        "offline_buffered_events": 0,
+        "offline_dropped_events": 0,
+        "subscriber_dropped_events": 0,
+        "unavailable_channels": 0,
+    }
+    try:
+        if not STREAMS_LOCK.acquire(blocking=False):
+            return result
+        try:
+            channels = list(STREAMS.values())
+            agent_count = len(AGENT_INSTANCES)
+        finally:
+            STREAMS_LOCK.release()
+        result["available"] = True
+        result["active"] = len(channels)
+        result["agent_instances"] = agent_count
+        for channel in channels:
+            try:
+                snapshot = channel.try_diagnostic_snapshot()
+                if snapshot is None:
+                    result["unavailable_channels"] += 1
+                    continue
+                result["subscribers"] += max(0, int(snapshot.get("subscriber_count", 0)))
+                result["offline_buffered_events"] += max(
+                    0, int(snapshot.get("offline_buffered_events", 0))
+                )
+                result["offline_dropped_events"] += max(
+                    0, int(snapshot.get("offline_dropped_events", 0))
+                )
+                result["subscriber_dropped_events"] += max(
+                    0, int(snapshot.get("subscriber_dropped_events", 0))
+                )
+            except Exception:
+                result["unavailable_channels"] += 1
+    except Exception:
+        # Keep the aggregates already summed from channels that read cleanly; a
+        # late unexpected failure must not discard successful sibling counts.
+        return result
+    return result
+
+
 def _session_payload_with_full_messages(session, *, tool_calls=None):
     """Return compact session metadata plus the embedded full transcript.
 

api/system_health.py

@@ -159,6 +159,113 @@ def _safe_error(metric: str, exc: Exception) -> dict[str, str]:
     return {"metric": metric, "code": type(exc).__name__}
 
 
+def _zero_webui_runtime_payload() -> dict[str, Any]:
+    return {
+        "sessions": {"available": False, "resident": 0, "cap": 0},
+        "streams": {
+            "available": False, "active": 0, "agent_instances": 0,
+            "subscribers": 0, "offline_buffered_events": 0,
+            "offline_dropped_events": 0, "subscriber_dropped_events": 0,
+            "unavailable_channels": 0,
+        },
+        "session_list_cache": {
+            "available": False, "entries": 0, "inflight_rebuilds": 0, "cap": 0,
+        },
+        "models_cache": {
+            "available": False, "groups": 0, "models": 0, "age_seconds": None,
+        },
+    }
+
+
+def _safe_runtime_int(value: Any) -> int:
+    try:
+        return max(0, int(value))
+    except (TypeError, ValueError):
+        return 0
+
+
+def _webui_runtime_payload(errors: list[dict[str, str]]) -> dict[str, Any]:
+    runtime = _zero_webui_runtime_payload()
+    # The three owner modules stay lazily imported so this small diagnostics
+    # module cannot create an eager import cycle. The import-lock wait that
+    # implies is only reachable while a first import is in flight, and
+    # api/routes.py imports all three at load, before any request can dispatch.
+    try:
+        from api import config
+    except Exception as exc:
+        config = None
+        config_import_error = exc
+    else:
+        config_import_error = None
+    try:
+        from api import route_session_list_cache
+    except Exception as exc:
+        route_session_list_cache = None
+        cache_import_error = exc
+    else:
+        cache_import_error = None
+    try:
+        from api import streaming
+    except Exception as exc:
+        streaming = None
+        streaming_import_error = exc
+    else:
+        streaming_import_error = None
+
+    # Collect the config owner exactly once per payload: two calls would double
+    # the blocking edge and let `sessions` and `models_cache` come from two
+    # different snapshots of the same process.
+    config_snapshot: Any = None
+    config_error = config_import_error
+    if config is not None and config_error is None:
+        try:
+            config_snapshot = config.get_runtime_diagnostics_snapshot()
+        except Exception as exc:
+            config_error = exc
+
+    def _config_section(section: str):
+        def project():
+            if not isinstance(config_snapshot, dict):
+                raise TypeError("runtime_snapshot_invalid")
+            return config_snapshot[section]
+
+        return project
+
+    sources = {
+        "sessions": (_config_section("sessions"), config_error),
+        "models_cache": (_config_section("models_cache"), config_error),
+        "streams": (
+            (lambda: streaming.get_stream_runtime_snapshot())
+            if streaming is not None else None,
+            streaming_import_error,
+        ),
+        "session_list_cache": (
+            (lambda: route_session_list_cache.get_session_list_cache_snapshot())
+            if route_session_list_cache is not None else None,
+            cache_import_error,
+        ),
+    }
+    for name, (collect, import_error) in sources.items():
+        try:
+            if import_error is not None:
+                raise import_error
+            source = collect()
+            if not isinstance(source, dict):
+                raise TypeError("runtime_snapshot_invalid")
+            target = runtime[name]
+            for key in target:
+                if key == "available":
+                    target[key] = bool(source.get(key, False))
+                elif key == "age_seconds":
+                    age = source.get(key)
+                    target[key] = None if age is None else max(0.0, float(age))
+                else:
+                    target[key] = _safe_runtime_int(source.get(key, 0))
+        except Exception as exc:
+            errors.append(_safe_error(f"webui_runtime.{name}", exc))
+    return runtime
+
+
 def build_system_health_payload() -> dict[str, Any]:
     metrics: dict[str, Any] = {"cpu": None, "memory": None, "disk": None}
     errors: list[dict[str, str]] = []
@@ -182,6 +289,15 @@ def build_system_health_payload() -> dict[str, Any]:
         except Exception as exc:
             errors.append(_safe_error(name, exc))
 
+    try:
+        runtime = _webui_runtime_payload(errors)
+    except Exception as exc:
+        # Terminal fail-open: an unexpected failure in the runtime compositor
+        # itself must not fail the whole authenticated response, which returned
+        # host metrics before this subtree existed.
+        runtime = _zero_webui_runtime_payload()
+        errors.append(_safe_error("webui_runtime", exc))
+
     available = any(metrics[name] is not None for name in metrics)
     status = "ok" if available and not errors else "partial" if available else "unavailable"
     return {
@@ -191,5 +307,6 @@ def build_system_health_payload() -> dict[str, Any]:
         "cpu": metrics["cpu"],
         "memory": metrics["memory"],
         "disk": metrics["disk"],
+        "webui_runtime": runtime,
         "errors": errors,
     }

tests/test_issue6335_catalog_admission_regression.py

@@ -0,0 +1,246 @@
+"""Regression test for #6335 / #6338 catalog admission gate.
+
+Issue #6335: after removing a provider's API key, a metadata-only
+``providers.<name>: {name: ...}`` entry in config.yaml still rendered the
+provider in the chat model selector.  The fix gates known-provider admission
+on credential evidence: a provider listed in config.yaml is only admitted
+when it was *already* detected from a credential source (env vars, hermes
+auth, credential pool) — or carries an explicit route (``api`` /
+``base_url`` / ``api_key`` / ``key_env``), or is the active provider with a
+configured models allowlist.
+
+Follow-up #6338: Hermes reports detected credentials under canonical alias
+names (``xai``, ``gemini``) while WebUI preserves config.yaml keys as
+``x-ai``, ``google``.  The credential check must compare in an
+alias-normalised identity space or authenticated providers disappear from
+the selector.
+
+This test drives ``api/config.py::get_available_models()`` end-to-end with
+a stubbed ``hermes_cli`` (deterministic, offline) and asserts the
+catalog-builder admission behaviour:
+
+1. Mocked Hermes provider detection reports authenticated ``xai`` and
+   ``gemini`` (the agent's canonical alias names).
+2. ``providers.x-ai`` and ``providers.google`` are metadata-only entries
+   (name only — no route, no models allowlist).
+3. The active provider is a *different* provider (``openai-codex``), so the
+   test cannot pass through the unconditional active-provider path.
+4. The xAI and Google groups remain admitted (alias-normalised credential
+   match).
+5. A metadata-only known provider with no matching credential, route field,
+   or configured model list (``openai-api``) stays absent.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import types
+
+import api.config as config
+import api.profiles as profiles
+
+
+def _install_fake_hermes_cli(monkeypatch):
+    """Stub hermes_cli so detection is deterministic and offline.
+
+    ``list_available_providers()`` reports authenticated ``xai`` and
+    ``gemini`` — the agent's canonical alias names — so the ONLY way the
+    metadata-only ``providers.x-ai`` / ``providers.google`` entries can be
+    admitted is via the alias-normalised ``_already_credentialed`` gate
+    under test.
+    """
+    fake_pkg = types.ModuleType("hermes_cli")
+    fake_pkg.__path__ = []
+
+    fake_models = types.ModuleType("hermes_cli.models")
+    fake_models.list_available_providers = lambda: [
+        {"id": "xai", "authenticated": True},
+        {"id": "gemini", "authenticated": True},
+    ]
+    fake_models.provider_model_ids = lambda pid: []
+    fake_models._PROVIDER_ALIASES = {
+        "x-ai": "xai",
+        "x.ai": "xai",
+        "grok": "xai",
+        "google": "gemini",
+        "google-gemini": "gemini",
+        "google-ai-studio": "gemini",
+    }
+
+    fake_auth = types.ModuleType("hermes_cli.auth")
+    fake_auth.get_auth_status = lambda _pid: {}
+
+    monkeypatch.setitem(sys.modules, "hermes_cli", fake_pkg)
+    monkeypatch.setitem(sys.modules, "hermes_cli.models", fake_models)
+    monkeypatch.setitem(sys.modules, "hermes_cli.auth", fake_auth)
+
+    # Remove the real agent.credential_pool so no pool evidence leaks in
+    # from the host environment.
+    monkeypatch.delitem(sys.modules, "agent.credential_pool", raising=False)
+    monkeypatch.delitem(sys.modules, "agent", raising=False)
+
+
+def _call_get_available_models(monkeypatch, tmp_path):
+    _install_fake_hermes_cli(monkeypatch)
+
+    # Active provider is a DIFFERENT provider (openai-codex), so the
+    # x-ai/google entries can only be admitted through the credential gate,
+    # never through the active-provider path.
+    (tmp_path / "auth.json").write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "providers": {},
+                "active_provider": "openai-codex",
+                "credential_pool": {},
+            }
+        ),
+        encoding="utf-8",
+    )
+    # config.yaml must exist on disk: get_available_models() reloads the
+    # config from disk when the path/mtime changed, wiping in-memory cfg
+    # overrides.  Metadata-only entries only (name) — no api/base_url/
+    # api_key/key_env route fields, no models allowlist.
+    (tmp_path / "config.yaml").write_text(
+        "model:\n  provider: openai-codex\nproviders:\n"
+        "  x-ai:\n    name: xAI\n"
+        "  google:\n    name: Google\n"
+        "  openai-api:\n    name: OpenAI API\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+
+    for var in (
+        "OPENAI_API_KEY",
+        "HERMES_API_KEY",
+        "HERMES_OPENAI_API_KEY",
+        "LOCAL_API_KEY",
+        "OPENROUTER_API_KEY",
+        "API_KEY",
+        "XAI_API_KEY",
+        "GOOGLE_API_KEY",
+        "GEMINI_API_KEY",
+    ):
+        monkeypatch.delenv(var, raising=False)
+
+    old_cfg = dict(config.cfg)
+    old_mtime = config._cfg_mtime
+    old_path = config._cfg_path
+    old_fingerprint = config._cfg_fingerprint
+    config.cfg.clear()
+    config.cfg["model"] = {"provider": "openai-codex"}
+    config.cfg["providers"] = {
+        "x-ai": {"name": "xAI"},
+        "google": {"name": "Google"},
+        "openai-api": {"name": "OpenAI API"},
+    }
+    cfg_path = config.Path(config._get_config_path())
+    try:
+        config._cfg_mtime = cfg_path.stat().st_mtime
+    except Exception:
+        config._cfg_mtime = 0.0
+    # Pin the path/fingerprint globals to THIS fixture so the path_changed /
+    # fingerprint staleness checks cannot force a disk reload that wipes the
+    # in-memory overrides above.  A preceding config test (e.g.
+    # test_byok_model_dropdown.py) can leave _cfg_path pointing at its own
+    # deleted temp config; without the pin, get_available_models() sees
+    # path_changed=True and reloads from disk, dropping the fixture's x-ai /
+    # google entries (gate certification 2026-08-07: byok -> 6335 = 1 failed).
+    config._cfg_path = cfg_path
+    config._cfg_fingerprint = config._fingerprint_config(config.cfg)
+
+    config.invalidate_models_cache()
+    try:
+        return config.get_available_models()
+    finally:
+        config.cfg.clear()
+        config.cfg.update(old_cfg)
+        config._cfg_mtime = old_mtime
+        config._cfg_path = old_path
+        config._cfg_fingerprint = old_fingerprint
+        config.invalidate_models_cache()
+
+
+def _group_ids(result):
+    return [g.get("provider_id") for g in result.get("groups", [])]
+
+
+def _group_names(result):
+    return [g.get("provider") for g in result.get("groups", [])]
+
+
+def test_predecessor_leaked_cfg_path_fingerprint_does_not_break_admission(
+    monkeypatch, tmp_path
+):
+    """Predecessor-order regression: leaked ``_cfg_path``/``_cfg_fingerprint``
+    must not trigger a disk reload that wipes the fixture providers.
+
+    A preceding config test (``tests/test_byok_model_dropdown.py`` sorts
+    before this file) repoints ``_get_config_path`` at its own temporary
+    config and leaves ``config._cfg_path`` / ``config._cfg_fingerprint``
+    pointing at that now-deleted file.  The next
+    ``get_available_models()`` call then sees ``path_changed=True`` and
+    reloads from disk, dropping the fixture's ``x-ai`` / ``google`` entries
+    (reproduced by the 2026-08-07 gate certification: byok -> 6335 = 1
+    failed, ``got ['openai-codex', 'gemini']``).
+
+    This test runs FIRST in this file, deliberately simulates that exact
+    leaked state, and asserts the helper pins over it — proving the two
+    admission tests below also pass after a leaking predecessor.
+    """
+    import api.config as config
+
+    leaked_path = tmp_path / "deleted-predecessor-config.yaml"
+    old_path, old_fingerprint = config._cfg_path, config._cfg_fingerprint
+    try:
+        config._cfg_path = leaked_path
+        config._cfg_fingerprint = "stale-predecessor-fingerprint"
+        result = _call_get_available_models(monkeypatch, tmp_path)
+        ids = _group_ids(result)
+        assert "openai-codex" in ids, f"active provider group missing; got {ids}"
+        assert "x-ai" in ids, f"leaked _cfg_path must not wipe x-ai; got {ids}"
+        assert "google" in ids, f"leaked _cfg_path must not wipe google; got {ids}"
+        assert "xAI" in _group_names(result)
+        assert "Google" in _group_names(result)
+    finally:
+        config._cfg_path = old_path
+        config._cfg_fingerprint = old_fingerprint
+
+
+def test_credentialed_alias_config_providers_stay_admitted(monkeypatch, tmp_path):
+    """Authenticated xai/gemini keep metadata-only x-ai/google entries in the picker.
+
+    Covers review points 1-4: detection reports authenticated ``xai`` /
+    ``gemini`` (alias names), config has metadata-only ``x-ai`` / ``google``
+    entries, the active provider is ``openai-codex`` (a different provider),
+    and the xAI / Google groups remain admitted through the alias-normalised
+    credential check.
+    """
+    result = _call_get_available_models(monkeypatch, tmp_path)
+    ids = _group_ids(result)
+
+    # Point 3 sanity: the active provider is a different provider, so these
+    # groups cannot be explained by the unconditional active-provider path.
+    assert "openai-codex" in ids, f"active provider group missing; got {ids}"
+
+    # Point 4: the alias-matched credential evidence keeps both groups.
+    assert "x-ai" in ids, f"authenticated xai must admit configured x-ai group; got {ids}"
+    assert "google" in ids, f"authenticated gemini must admit configured google group; got {ids}"
+    assert "xAI" in _group_names(result)
+    assert "Google" in _group_names(result)
+
+
+def test_keyless_metadata_only_known_provider_stays_absent(monkeypatch, tmp_path):
+    """A metadata-only known provider with no credential/route/models is dropped.
+
+    Review point 5: ``providers.openai-api: {name: ...}`` is a *known*
+    provider (in ``_PROVIDER_MODELS`` / ``_PROVIDER_DISPLAY``) but carries no
+    credential evidence, no route field, and no configured model list — the
+    #6335 bug.  It must not render in the model selector.
+    """
+    result = _call_get_available_models(monkeypatch, tmp_path)
+    ids = _group_ids(result)
+    assert "openai-api" not in ids, (
+        f"keyless metadata-only openai-api must stay absent; got {ids}"
+    )

tests/test_issue693_system_health_panel.py

@@ -2,10 +2,17 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
+import os
 import pathlib
+import queue
+import subprocess
 import sys
+import threading
+import time
 from types import SimpleNamespace
+from unittest.mock import Mock, patch
 from urllib.parse import urlparse
 
 
@@ -302,3 +309,807 @@ def test_system_health_backend_uses_no_shell_or_private_process_sources():
     assert "/proc/self/environ" not in src
     for private_field in ("argv", "cmdline", "username", "mountpoint"):
         assert private_field not in src
+
+
+# #6351 runtime-diagnostics hygiene rule for every test below: a test that
+# touches a process-owned shared mapping copies it by value, performs the
+# clear/seed and the restore inside the mapping's owning lock, and restores the
+# original key order and value identities. Contested locks are always held from
+# a worker thread, which is the only schedule production can produce.
+_LOCK_HOLD_SECONDS = 3.0
+# A nonblocking collector does no I/O, so a second is generous even on a loaded
+# CI box, while staying far enough below the hold that a blocking regression
+# (which waits the full _LOCK_HOLD_SECONDS) cannot slip past.
+_NONBLOCKING_BUDGET_SECONDS = 1.0
+
+
+@contextlib.contextmanager
+def _held_from_worker(lock, hold_seconds=_LOCK_HOLD_SECONDS):
+    """Hold ``lock`` from a worker thread for the body of the with-block."""
+    entered = threading.Event()
+    release = threading.Event()
+
+    def hold():
+        with lock:
+            entered.set()
+            release.wait(hold_seconds)
+
+    worker = threading.Thread(target=hold, daemon=True)
+    worker.start()
+    try:
+        # Inside the try so a worker that never acquires still gets released and
+        # joined; otherwise it takes the lock after this test fails and starves
+        # the next contention test for the full hold.
+        assert entered.wait(5), "worker thread never acquired the contested lock"
+        yield
+    finally:
+        release.set()
+        worker.join(hold_seconds + 5)
+
+
+@contextlib.contextmanager
+def _seeded_mapping(lock, mapping, entries):
+    """Seed a process-owned mapping under its owning lock and restore it exactly."""
+    with lock:
+        original = dict(mapping)
+        mapping.clear()
+        mapping.update(entries)
+    try:
+        yield original
+    finally:
+        with lock:
+            mapping.clear()
+            mapping.update(original)
+
+
+def _seed_channel_counters(
+    channel,
+    *,
+    subscribers=0,
+    offline_buffered=0,
+    offline_dropped=0,
+    subscriber_dropped=0,
+):
+    """Seed a real StreamChannel's diagnostic counters under its own lock."""
+    with channel._lock:
+        channel._subscribers.extend(queue.Queue() for _ in range(subscribers))
+        channel._offline_buffer.extend(
+            ("token", {"seq": index}, str(index)) for index in range(offline_buffered)
+        )
+        channel._offline_dropped_total = offline_dropped
+        channel._subscriber_dropped_total = subscriber_dropped
+
+
+def _pin_config_module_cache(monkeypatch):
+    """Let pytest restore the config globals a forced reload or eviction rebinds."""
+    from api import config
+
+    # _LAST_APPLIED_SESSIONS_CACHE_MAX belongs here because a real
+    # _evict_sessions_over_cap() pass rewrites it outside monkeypatch's view;
+    # pinning first records the true pre-test value.
+    for name in (
+        "cfg",
+        "_cfg_cache",
+        "_cfg_mtime",
+        "_cfg_path",
+        "_cfg_fingerprint",
+        "_LAST_APPLIED_SESSIONS_CACHE_MAX",
+    ):
+        monkeypatch.setattr(config, name, getattr(config, name), raising=False)
+
+
+def _fixed_host_metrics(monkeypatch):
+    from api import system_health
+
+    monkeypatch.setattr(system_health, "_cpu_percent", lambda: 12.0)
+    monkeypatch.setattr(
+        system_health,
+        "_memory_usage",
+        lambda: {"used_bytes": 1, "total_bytes": 4, "percent": 25.0},
+    )
+    monkeypatch.setattr(
+        system_health,
+        "_disk_usage",
+        lambda: {"used_bytes": 4, "total_bytes": 8, "percent": 50.0},
+    )
+
+
+def _assert_host_metrics_intact(payload):
+    assert payload["cpu"] == {"percent": 12.0}
+    assert payload["memory"] == {"used_bytes": 1, "total_bytes": 4, "percent": 25.0}
+    assert payload["disk"] == {"used_bytes": 4, "total_bytes": 8, "percent": 50.0}
+
+
+def test_runtime_diagnostics_absence_presence_and_privacy(monkeypatch):
+    from api import system_health
+
+    monkeypatch.setattr(system_health, "_cpu_percent", lambda: 1.0)
+    monkeypatch.setattr(system_health, "_memory_usage", lambda: {"used_bytes": 1, "total_bytes": 2, "percent": 50})
+    monkeypatch.setattr(system_health, "_disk_usage", lambda: {"used_bytes": 1, "total_bytes": 2, "percent": 50})
+    payload = system_health.build_system_health_payload()
+    assert set(payload["webui_runtime"]) == {"sessions", "streams", "session_list_cache", "models_cache"}
+    assert "stream_id" not in repr(payload)
+    assert "provider" not in repr(payload)
+    assert "secret-model" not in repr(payload)
+    assert "private-provider" not in repr(payload)
+
+
+def test_health_route_completes_while_config_cache_lock_is_held(monkeypatch, tmp_path):
+    from api import config, routes
+
+    _fixed_host_metrics(monkeypatch)
+    _pin_config_module_cache(monkeypatch)
+
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text("webui:\n  sessions_cache_max: 41\n", encoding="utf-8")
+    monkeypatch.setenv("HERMES_CONFIG_PATH", str(config_path))
+    # Cold cache: get_config() takes the reload branch, which waits on _cfg_lock.
+    monkeypatch.setattr(config, "_cfg_cache", {})
+    monkeypatch.setattr(config, "_cfg_mtime", 0.0)
+    monkeypatch.setattr(config, "_cfg_path", None)
+    monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 41, raising=False)
+
+    seeded = {"health-route-a": object(), "health-route-b": object()}
+    with _seeded_mapping(config.LOCK, config.SESSIONS, seeded):
+        with _held_from_worker(config._cfg_lock):
+            handler = _FakeHandler()
+            started = time.monotonic()
+            assert routes.handle_get(handler, urlparse("http://example.test/api/system/health")) is True
+            elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS, (
+        f"health response waited {elapsed:.3f}s on the config cache lock"
+    )
+    payload = handler.json_body()
+    _assert_host_metrics_intact(payload)
+    assert payload["status"] == "ok"
+    runtime = payload["webui_runtime"]
+    assert runtime["sessions"] == {"available": True, "resident": 2, "cap": 41}
+    assert runtime["streams"]["available"] is True
+    assert runtime["session_list_cache"]["available"] is True
+    assert runtime["session_list_cache"]["cap"] == 64
+    assert set(runtime["models_cache"]) == {"available", "groups", "models", "age_seconds"}
+    assert "health-route-a" not in repr(payload)
+
+
+def test_sessions_snapshot_never_resolves_config_or_profile(monkeypatch):
+    from api import config, profiles, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    fired = []
+
+    def trap(*args, **kwargs):
+        fired.append(True)
+        raise AssertionError("health collection must not resolve config or profile state")
+
+    monkeypatch.setattr(config, "get_config", trap)
+    monkeypatch.setattr(config, "_get_config_path", trap)
+    monkeypatch.setattr(profiles, "get_active_hermes_home", trap)
+    monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 57, raising=False)
+
+    with _seeded_mapping(config.LOCK, config.SESSIONS, {"trap-probe": object()}):
+        payload = system_health.build_system_health_payload()
+
+    assert fired == []
+    assert payload["webui_runtime"]["sessions"] == {"available": True, "resident": 1, "cap": 57}
+    assert {"metric": "webui_runtime.sessions", "code": "AssertionError"} not in payload["errors"]
+
+
+def test_health_route_completes_while_one_real_stream_channel_lock_is_held(monkeypatch):
+    from api import config, routes
+
+    _fixed_host_metrics(monkeypatch)
+    healthy = config.StreamChannel()
+    _seed_channel_counters(
+        healthy,
+        subscribers=2,
+        offline_buffered=3,
+        offline_dropped=4,
+        subscriber_dropped=5,
+    )
+    busy = config.StreamChannel()
+
+    streams = {"health-open-stream": healthy, "health-busy-stream": busy}
+    with _seeded_mapping(config.STREAMS_LOCK, config.STREAMS, streams):
+        with _seeded_mapping(config.STREAMS_LOCK, config.AGENT_INSTANCES, {"health-open-stream": object()}):
+            with _held_from_worker(busy._lock):
+                handler = _FakeHandler()
+                started = time.monotonic()
+                assert routes.handle_get(handler, urlparse("http://example.test/api/system/health")) is True
+                elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS, (
+        f"health response waited {elapsed:.3f}s on one busy stream channel"
+    )
+    payload = handler.json_body()
+    _assert_host_metrics_intact(payload)
+    assert payload["status"] == "ok"
+    runtime = payload["webui_runtime"]
+    assert runtime["streams"] == {
+        "available": True,
+        "active": 2,
+        "agent_instances": 1,
+        "subscribers": 2,
+        "offline_buffered_events": 3,
+        "offline_dropped_events": 4,
+        "subscriber_dropped_events": 5,
+        "unavailable_channels": 1,
+    }
+    assert runtime["sessions"]["available"] is True
+    assert runtime["session_list_cache"]["available"] is True
+    assert set(runtime["models_cache"]) == {"available", "groups", "models", "age_seconds"}
+    assert "health-open-stream" not in repr(payload)
+    assert "health-busy-stream" not in repr(payload)
+
+
+def test_stream_channel_try_snapshot_matches_blocking_snapshot():
+    from api import config
+
+    channel = config.StreamChannel()
+    _seed_channel_counters(
+        channel,
+        subscribers=1,
+        offline_buffered=2,
+        offline_dropped=3,
+        subscriber_dropped=4,
+    )
+    blocking = channel.diagnostic_snapshot()
+    nonblocking = channel.try_diagnostic_snapshot()
+    assert nonblocking == blocking
+    assert set(nonblocking) == {
+        "subscriber_count",
+        "offline_buffered_events",
+        "offline_dropped_events",
+        "subscriber_dropped_events",
+    }
+    assert blocking == {
+        "subscriber_count": 1,
+        "offline_buffered_events": 2,
+        "offline_dropped_events": 3,
+        "subscriber_dropped_events": 4,
+    }
+
+    # diagnostic_snapshot() is deliberately not called here: it keeps its
+    # blocking contract for /health?deep=1 and would wait out the whole hold.
+    with _held_from_worker(channel._lock):
+        started = time.monotonic()
+        assert channel.try_diagnostic_snapshot() is None
+        assert time.monotonic() - started < _NONBLOCKING_BUDGET_SECONDS
+
+
+def test_sessions_owner_snapshot_preserves_existing_cache_identities(monkeypatch):
+    from api import config
+
+    monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 37, raising=False)
+    probe_key = "pre-existing-identity-probe"
+    probe = object()
+    with config.LOCK:
+        config.SESSIONS[probe_key] = probe
+    try:
+        seeded = {"first": object(), "second": object()}
+        with _seeded_mapping(config.LOCK, config.SESSIONS, seeded) as original:
+            assert probe_key not in config.SESSIONS
+            assert config.get_runtime_diagnostics_snapshot()["sessions"] == {
+                "available": True, "resident": 2, "cap": 37
+            }
+            assert list(config.SESSIONS) == ["first", "second"]
+
+        assert list(config.SESSIONS) == list(original)
+        for key, value in original.items():
+            assert config.SESSIONS[key] is value
+        assert config.SESSIONS[probe_key] is probe
+    finally:
+        with config.LOCK:
+            config.SESSIONS.pop(probe_key, None)
+
+
+def test_diagnostics_cap_tracks_the_eviction_owner(monkeypatch, tmp_path):
+    from api import config, models
+
+    _pin_config_module_cache(monkeypatch)
+    config_path = tmp_path / "config.yaml"
+    monkeypatch.setenv("HERMES_CONFIG_PATH", str(config_path))
+    fallback = (
+        config.SESSIONS_MAX
+        if isinstance(config.SESSIONS_MAX, int) and config.SESSIONS_MAX >= 1
+        else config.DEFAULT_SESSIONS_CACHE_MAX
+    )
+
+    for body, expected in (
+        ("webui:\n  sessions_cache_max: 23\n", 23),
+        ("webui:\n  sessions_cache_max: 0\n", fallback),
+        ("webui: {}\n", fallback),
+    ):
+        config_path.write_text(body, encoding="utf-8")
+        monkeypatch.setattr(config, "_cfg_cache", {})
+        monkeypatch.setattr(config, "_cfg_path", None)
+        assert config.get_sessions_cache_max() == expected
+        # Poison the memo so only a real eviction pass can restore it.
+        monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", -1)
+        with _seeded_mapping(config.LOCK, config.SESSIONS, {}):
+            with config.LOCK:
+                models._evict_sessions_over_cap()
+            assert config.get_runtime_diagnostics_snapshot()["sessions"]["cap"] == expected
+
+
+def test_diagnostics_cap_reports_the_owner_fallback_when_the_getter_raises(monkeypatch):
+    from api import config, models, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    _pin_config_module_cache(monkeypatch)
+
+    def raiser(*args, **kwargs):
+        raise RuntimeError("cap resolution failed")
+
+    # A distinct prior memo, so no assertion below can pass on the incumbent value.
+    monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 4321, raising=False)
+    # api/models.py resolves the getter through _cfg at call time, so failing the
+    # real attribute is what the production lookup hits, not a shadowed copy.
+    monkeypatch.setattr(config, "get_sessions_cache_max", raiser)
+    # Pinning models' own binding is what proves the published cap came from the
+    # enforcer's fallback rather than from anything the resolver returned.
+    monkeypatch.setattr(models, "SESSIONS_MAX", 29)
+
+    with _seeded_mapping(config.LOCK, config.SESSIONS, {}):
+        with config.LOCK:
+            models._evict_sessions_over_cap()
+        snapshot = config.get_runtime_diagnostics_snapshot()
+        payload = system_health.build_system_health_payload()
+
+    assert snapshot["sessions"]["cap"] != 4321
+    assert snapshot["sessions"]["cap"] == 29
+    assert payload["webui_runtime"]["sessions"]["cap"] == 29
+
+
+def test_diagnostics_cap_reports_the_normalized_explicit_cap(monkeypatch):
+    from api import config, models, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    _pin_config_module_cache(monkeypatch)
+    # An explicit cap never reaches the resolver, so only the enforcer can publish.
+    monkeypatch.setattr(models, "SESSIONS_MAX", 17)
+
+    for explicit, expected in ((0, 17), ("nope", 17), (5, 5)):
+        monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 4321, raising=False)
+        with _seeded_mapping(config.LOCK, config.SESSIONS, {}):
+            with config.LOCK:
+                models._evict_sessions_over_cap(cap=explicit)
+            snapshot = config.get_runtime_diagnostics_snapshot()
+            payload = system_health.build_system_health_payload()
+
+        assert snapshot["sessions"]["cap"] != 4321, explicit
+        assert snapshot["sessions"]["cap"] == expected, explicit
+        assert payload["webui_runtime"]["sessions"]["cap"] == expected, explicit
+
+
+def test_the_cap_getter_does_not_publish_diagnostics_state(monkeypatch, tmp_path):
+    from api import config
+
+    _pin_config_module_cache(monkeypatch)
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text("webui:\n  sessions_cache_max: 88\n", encoding="utf-8")
+    monkeypatch.setenv("HERMES_CONFIG_PATH", str(config_path))
+    monkeypatch.setattr(config, "_cfg_cache", {})
+    monkeypatch.setattr(config, "_cfg_mtime", 0.0)
+    monkeypatch.setattr(config, "_cfg_path", None)
+    monkeypatch.setattr(config, "_LAST_APPLIED_SESSIONS_CACHE_MAX", 4321, raising=False)
+
+    assert config.get_sessions_cache_max({"webui": {"sessions_cache_max": 42}}) == 42
+    assert config._LAST_APPLIED_SESSIONS_CACHE_MAX == 4321
+    assert config.get_sessions_cache_max() == 88
+    assert config._LAST_APPLIED_SESSIONS_CACHE_MAX == 4321
+
+    # Process start: a fresh import must already carry the getter's dict-mode
+    # answer for the config it loaded, not the SESSIONS_MAX/default tier. Read in
+    # a child process because this one's memo has been rewritten by earlier tests.
+    seeded = subprocess.run(
+        [sys.executable, "-c", "import api.config as c; print(c._LAST_APPLIED_SESSIONS_CACHE_MAX)"],
+        cwd=str(REPO_ROOT),
+        env={**os.environ, "HERMES_CONFIG_PATH": str(config_path)},
+        capture_output=True,
+        text=True,
+        timeout=45,
+    )
+    assert seeded.returncode == 0, seeded.stderr
+    printed = seeded.stdout.strip().splitlines()
+    assert printed, f"child printed nothing; stderr={seeded.stderr!r}"
+    assert int(printed[-1]) == 88
+
+
+def test_infinite_cap_falls_back_instead_of_raising(tmp_path):
+    """A YAML float infinity must degrade to the bound, not abort the import.
+
+    ``yaml.safe_load`` resolves ``.inf`` to a real float and ``int(float('inf'))``
+    raises OverflowError, which is neither TypeError nor ValueError. The cap is
+    resolved at module scope, so an uncaught raise here takes down startup.
+    """
+    from api import config
+
+    fallback = (
+        config.SESSIONS_MAX
+        if isinstance(config.SESSIONS_MAX, int) and config.SESSIONS_MAX >= 1
+        else config.DEFAULT_SESSIONS_CACHE_MAX
+    )
+    for raw in (float("inf"), float("-inf"), float("nan")):
+        assert config.get_sessions_cache_max({"webui": {"sessions_cache_max": raw}}) == fallback
+
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text("webui:\n  sessions_cache_max: .inf\n", encoding="utf-8")
+    booted = subprocess.run(
+        [sys.executable, "-c", "import api.config as c; print(c._LAST_APPLIED_SESSIONS_CACHE_MAX)"],
+        cwd=str(REPO_ROOT),
+        env={**os.environ, "HERMES_CONFIG_PATH": str(config_path)},
+        capture_output=True,
+        text=True,
+        timeout=45,
+    )
+    assert booted.returncode == 0, booted.stderr
+    printed = booted.stdout.strip().splitlines()
+    assert printed, f"child printed nothing; stderr={booted.stderr!r}"
+    assert int(printed[-1]) == fallback
+
+
+def test_sessions_lock_busy_does_not_block_sibling_owner():
+    from api import config
+
+    with _held_from_worker(config.LOCK):
+        started = time.monotonic()
+        snapshot = config.get_runtime_diagnostics_snapshot()
+        elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS
+    assert snapshot["sessions"] == {"available": False, "resident": 0, "cap": 0}
+    assert "models_cache" in snapshot
+
+
+def test_health_status_stays_ok_when_sessions_owner_lock_is_busy(monkeypatch):
+    from api import config, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    with _held_from_worker(config.LOCK):
+        started = time.monotonic()
+        payload = system_health.build_system_health_payload()
+        elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS
+    assert payload["status"] == "ok"
+    assert payload["webui_runtime"]["sessions"]["available"] is False
+    assert payload["webui_runtime"]["models_cache"]["available"] is True
+
+
+def test_models_cache_snapshot_counts_published_groups_without_names():
+    from api import config
+
+    groups = [
+        {
+            "name": "private-provider",
+            "models": [{"id": "secret-model"}],
+            "extra_models": [{"id": "overflow-secret"}, {"id": "other"}],
+        },
+        {"models": [], "extra_models": [{"id": "overflow-two"}]},
+    ]
+    baseline = time.monotonic()
+    with config._available_models_cache_lock:
+        original_cache = config._available_models_cache
+        original_ts = config._available_models_cache_ts
+        config._available_models_cache = {"groups": groups}
+        config._available_models_cache_ts = baseline - 30.0
+    try:
+        result = config.get_runtime_diagnostics_snapshot()["models_cache"]
+    finally:
+        with config._available_models_cache_lock:
+            config._available_models_cache = original_cache
+            config._available_models_cache_ts = original_ts
+
+    assert result["available"] is True
+    assert result["groups"] == 2
+    assert result["models"] == 4
+    # Real clock, bounded range: patching time.monotonic would freeze the stdlib
+    # clock for every other thread in the process.
+    assert 30.0 <= result["age_seconds"] < 30.0 + _NONBLOCKING_BUDGET_SECONDS
+    assert "private-provider" not in repr(result)
+    assert "secret-model" not in repr(result)
+
+
+def test_stream_owner_snapshot_aggregates_channels():
+    from api import config, streaming
+
+    first = config.StreamChannel()
+    _seed_channel_counters(
+        first,
+        subscribers=2,
+        offline_buffered=3,
+        offline_dropped=4,
+        subscriber_dropped=5,
+    )
+    second = config.StreamChannel()
+    _seed_channel_counters(second, subscribers=1, subscriber_dropped=7)
+
+    streams = {"secret-id": first, "other-secret-id": second}
+    with _seeded_mapping(config.STREAMS_LOCK, config.STREAMS, streams):
+        with _seeded_mapping(config.STREAMS_LOCK, config.AGENT_INSTANCES, {"secret-id": object()}):
+            result = streaming.get_stream_runtime_snapshot()
+
+    assert result == {
+        "available": True, "active": 2, "agent_instances": 1,
+        "subscribers": 3, "offline_buffered_events": 3,
+        "offline_dropped_events": 4, "subscriber_dropped_events": 12,
+        "unavailable_channels": 0,
+    }
+    assert "secret-id" not in repr(result)
+
+
+def test_stream_registry_lock_busy_isolated():
+    from api import config, streaming
+
+    with _held_from_worker(config.STREAMS_LOCK):
+        started = time.monotonic()
+        result = streaming.get_stream_runtime_snapshot()
+        elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS
+    assert result["available"] is False
+    assert result["active"] == 0
+
+
+def test_session_list_cache_snapshot_reads_scalars_without_lru_mutation():
+    from api import route_session_list_cache as cache
+
+    probe_key = ("pre-existing-identity-probe",)
+    probe = (0.0, (), {})
+    with cache._SESSIONS_CACHE_LOCK:
+        cache._SESSIONS_CACHE[probe_key] = probe
+    try:
+        with _seeded_mapping(
+            cache._SESSIONS_CACHE_LOCK, cache._SESSIONS_CACHE, {("one",): (0.0, (), {})}
+        ) as original:
+            result = cache.get_session_list_cache_snapshot()
+            assert result == {"available": True, "entries": 1, "inflight_rebuilds": 0, "cap": 64}
+            assert list(cache._SESSIONS_CACHE) == [("one",)]
+
+        assert list(cache._SESSIONS_CACHE) == list(original)
+        assert cache._SESSIONS_CACHE[probe_key] is probe
+    finally:
+        with cache._SESSIONS_CACHE_LOCK:
+            cache._SESSIONS_CACHE.pop(probe_key, None)
+
+
+def test_runtime_unexpected_exception_fail_open(monkeypatch):
+    from api import config, system_health
+
+    monkeypatch.setattr(system_health, "_cpu_percent", lambda: 1.0)
+    monkeypatch.setattr(config, "get_runtime_diagnostics_snapshot", lambda: (_ for _ in ()).throw(RecursionError("secret")))
+    payload = system_health.build_system_health_payload()
+    assert payload["status"] == "partial"
+    assert payload["webui_runtime"]["sessions"]["available"] is False
+    assert {"metric": "webui_runtime.sessions", "code": "RecursionError"} in payload["errors"]
+    assert "secret" not in repr(payload)
+
+
+def test_runtime_owner_failure_preserves_sibling_metrics(monkeypatch):
+    from api import config, route_session_list_cache, streaming, system_health
+
+    monkeypatch.setattr(config, "get_runtime_diagnostics_snapshot", lambda: (_ for _ in ()).throw(RecursionError("secret config")))
+    monkeypatch.setattr(streaming, "get_stream_runtime_snapshot", lambda: {"available": True, "active": 4, "agent_instances": 3, "subscribers": 0, "offline_buffered_events": 0, "offline_dropped_events": 0, "subscriber_dropped_events": 0, "unavailable_channels": 0})
+    monkeypatch.setattr(route_session_list_cache, "get_session_list_cache_snapshot", lambda: {"available": True, "entries": 2, "inflight_rebuilds": 0, "cap": 64})
+    payload = system_health.build_system_health_payload()
+    runtime = payload["webui_runtime"]
+    assert runtime["sessions"]["available"] is False
+    assert runtime["models_cache"]["available"] is False
+    assert runtime["streams"]["active"] == 4
+    assert runtime["session_list_cache"]["entries"] == 2
+    assert "secret config" not in repr(payload)
+
+
+def test_config_owner_collected_once_per_health_payload(monkeypatch):
+    from api import config, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    real = config.get_runtime_diagnostics_snapshot
+    calls = []
+
+    def counting_wrapper():
+        calls.append(True)
+        snapshot = real()
+        # Successive calls disagree, so a second collection would show up as a
+        # sessions/models_cache skew in the response.
+        snapshot["sessions"]["resident"] = 10 * len(calls)
+        snapshot["models_cache"]["groups"] = 10 * len(calls)
+        return snapshot
+
+    monkeypatch.setattr(config, "get_runtime_diagnostics_snapshot", counting_wrapper)
+    payload = system_health.build_system_health_payload()
+
+    assert len(calls) == 1
+    runtime = payload["webui_runtime"]
+    assert runtime["sessions"]["resident"] == 10
+    assert runtime["models_cache"]["groups"] == 10
+
+
+def test_webui_runtime_compositor_failure_preserves_host_metrics(monkeypatch):
+    from api import system_health
+
+    _fixed_host_metrics(monkeypatch)
+
+    def boom(errors):
+        raise RecursionError("secret compositor path")
+
+    monkeypatch.setattr(system_health, "_webui_runtime_payload", boom)
+    payload = system_health.build_system_health_payload()
+
+    _assert_host_metrics_intact(payload)
+    assert payload["status"] == "partial"
+    assert payload["webui_runtime"] == system_health._zero_webui_runtime_payload()
+    assert {"metric": "webui_runtime", "code": "RecursionError"} in payload["errors"]
+    assert "secret compositor path" not in repr(payload)
+
+
+def test_cancel_cleanup_updates_runtime_diagnostics():
+    from api import config, streaming
+
+    stream_id = "production-cleanup-stream"
+    session_id = "production-cleanup-session"
+    agent = Mock()
+    agent.session_id = session_id
+    session = SimpleNamespace(
+        session_id=session_id,
+        active_stream_id=stream_id,
+        pending_user_message=None,
+        pending_user_source=None,
+        pending_attachments=[],
+        pending_started_at=None,
+        messages=[],
+        save=Mock(),
+    )
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(
+            _seeded_mapping(config.STREAMS_LOCK, config.STREAMS, {stream_id: config.StreamChannel()})
+        )
+        stack.enter_context(
+            _seeded_mapping(config.STREAMS_LOCK, config.AGENT_INSTANCES, {stream_id: agent})
+        )
+        stack.enter_context(
+            _seeded_mapping(config.STREAMS_LOCK, config.CANCEL_FLAGS, {stream_id: threading.Event()})
+        )
+        stack.enter_context(_seeded_mapping(config.ACTIVE_RUNS_LOCK, config.ACTIVE_RUNS, {}))
+        before = streaming.get_stream_runtime_snapshot()
+        with patch("api.streaming.get_session", return_value=session):
+            assert streaming.cancel_stream(stream_id) is True
+        after = streaming.get_stream_runtime_snapshot()
+
+    assert before["active"] == 1
+    assert before["agent_instances"] == 1
+    assert after["active"] == 0
+    assert after["agent_instances"] == 0
+    agent.interrupt.assert_called_once_with("Cancelled by user")
+
+
+def test_models_cache_lock_busy():
+    from api import config
+
+    with _held_from_worker(config._available_models_cache_lock):
+        started = time.monotonic()
+        snapshot = config.get_runtime_diagnostics_snapshot()
+        elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS
+    assert snapshot["models_cache"]["available"] is False
+
+
+def test_session_list_cache_lock_busy():
+    from api import route_session_list_cache as cache
+
+    with _held_from_worker(cache._SESSIONS_CACHE_LOCK):
+        started = time.monotonic()
+        result = cache.get_session_list_cache_snapshot()
+        elapsed = time.monotonic() - started
+
+    assert elapsed < _NONBLOCKING_BUDGET_SECONDS
+    assert result["available"] is False
+
+
+def test_stream_snapshot_failure():
+    from api import config, streaming
+
+    class _FailingChannel(config.StreamChannel):
+        def try_diagnostic_snapshot(self):
+            raise RecursionError("secret path")
+
+    healthy = config.StreamChannel()
+    _seed_channel_counters(
+        healthy,
+        subscribers=2,
+        offline_buffered=3,
+        offline_dropped=4,
+        subscriber_dropped=5,
+    )
+    streams = {"healthy-secret-id": healthy, "failing-secret-id": _FailingChannel()}
+    with _seeded_mapping(config.STREAMS_LOCK, config.STREAMS, streams):
+        with _seeded_mapping(config.STREAMS_LOCK, config.AGENT_INSTANCES, {"healthy-secret-id": object()}):
+            result = streaming.get_stream_runtime_snapshot()
+
+    assert result == {
+        "available": True, "active": 2, "agent_instances": 1,
+        "subscribers": 2, "offline_buffered_events": 3,
+        "offline_dropped_events": 4, "subscriber_dropped_events": 5,
+        "unavailable_channels": 1,
+    }
+    rendered = repr(result)
+    assert "secret path" not in rendered
+    assert "RecursionError" not in rendered
+    assert "secret-id" not in rendered
+
+
+def test_runtime_route_auth_and_privacy(monkeypatch):
+    from api import auth as _auth
+    from api.auth import check_auth
+    from api import system_health
+
+    monkeypatch.setenv("HERMES_WEBUI_PASSWORD", "test-password")
+    _auth._invalidate_password_hash_cache()
+    handler = _FakeHandler()
+    try:
+        assert check_auth(handler, SimpleNamespace(path="/api/system/health", query="")) is False
+        assert handler.status in (302, 401)
+    finally:
+        monkeypatch.delenv("HERMES_WEBUI_PASSWORD", raising=False)
+        _auth._invalidate_password_hash_cache()
+
+    monkeypatch.setattr(system_health, "_cpu_percent", lambda: 1.0)
+    payload = system_health.build_system_health_payload()
+    assert payload["webui_runtime"]["streams"].keys() >= {
+        "available", "active", "agent_instances", "subscribers"
+    }
+    assert all(key not in repr(payload) for key in ("stream-id", "session-id", "secret-model"))
+
+
+def test_runtime_owner_adapter_routing(monkeypatch):
+    from api import config, route_session_list_cache, streaming, system_health
+
+    monkeypatch.setattr(config, "get_runtime_diagnostics_snapshot", lambda: {
+        "sessions": {"available": True, "resident": 9, "cap": 11},
+        "models_cache": {"available": True, "groups": 3, "models": 4, "age_seconds": 5},
+    })
+    monkeypatch.setattr(streaming, "get_stream_runtime_snapshot", lambda: {
+        "available": True, "active": 7, "agent_instances": 6, "subscribers": 5,
+        "offline_buffered_events": 4, "offline_dropped_events": 3,
+        "subscriber_dropped_events": 2, "unavailable_channels": 1,
+    })
+    monkeypatch.setattr(route_session_list_cache, "get_session_list_cache_snapshot", lambda: {
+        "available": True, "entries": 8, "inflight_rebuilds": 1, "cap": 64,
+    })
+    runtime = system_health._webui_runtime_payload([])
+    assert runtime["sessions"]["resident"] == 9
+    assert runtime["streams"]["active"] == 7
+    assert runtime["session_list_cache"]["entries"] == 8
+
+
+def test_runtime_diagnostics_ignore_profile_storage(monkeypatch, tmp_path):
+    from api import config, profiles, system_health
+
+    _fixed_host_metrics(monkeypatch)
+    hermes_home = tmp_path / ".hermes"
+    leftover_profile = hermes_home / "profiles" / "work"
+    (leftover_profile / "sessions").mkdir(parents=True)
+    (leftover_profile / "config.yaml").write_text(
+        "webui:\n  sessions_cache_max: 7\n", encoding="utf-8"
+    )
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    fired = []
+
+    def trap(*args, **kwargs):
+        fired.append(True)
+        raise AssertionError("health collection must not resolve profile storage")
+
+    monkeypatch.setattr(profiles, "get_active_hermes_home", trap)
+    monkeypatch.setattr(config, "_get_config_path", trap)
+
+    payload = system_health.build_system_health_payload()
+
+    assert fired == []
+    assert payload["status"] in {"ok", "partial"}
+    rendered = repr(payload)
+    assert "profiles" not in rendered
+    assert str(hermes_home) not in rendered
+    assert "work" not in rendered