hermes: expose provider pool status
All checks were successful
Tests / Declarative: Post Actions passed: 249

This commit is contained in:
jenkins 2026-08-12 02:23:31 -03:00
parent 37502108c4
commit b12b28bdcb
9 changed files with 680 additions and 2 deletions

View File

@ -25,7 +25,7 @@ spec:
ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers
ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback
ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available
ai.bstein.dev/config-rev: "20260812-codex-budget-normalization"
ai.bstein.dev/config-rev: "20260812-provider-status"
vault.hashicorp.com/agent-inject: "true"
vault.hashicorp.com/role: hermes-agent
vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens
@ -827,6 +827,14 @@ spec:
- name: auto-router-plugin
configMap:
name: hermes-auto-router-plugin
items:
- {key: __init__.py, path: __init__.py}
- {key: provider_status.py, path: provider_status.py}
- {key: plugin.yaml, path: plugin.yaml}
- {key: dashboard-manifest.json, path: dashboard/manifest.json}
- {key: dashboard-api.py, path: dashboard/plugin_api.py}
- {key: dashboard-index.js, path: dashboard/dist/index.js}
- {key: dashboard-style.css, path: dashboard/dist/style.css}
- name: image-policy
configMap:
name: hermes-image-policy

View File

@ -86,7 +86,12 @@ configMapGenerator:
namespace: hermes
files:
- __init__.py=plugins/auto-router/__init__.py
- provider_status.py=plugins/auto-router/provider_status.py
- plugin.yaml=plugins/auto-router/plugin.yaml
- dashboard-manifest.json=plugins/auto-router/dashboard/manifest.json
- dashboard-api.py=plugins/auto-router/dashboard/plugin_api.py
- dashboard-index.js=plugins/auto-router/dashboard/dist/index.js
- dashboard-style.css=plugins/auto-router/dashboard/dist/style.css
options:
disableNameSuffixHash: true
- name: hermes-chat-image-plugin

View File

@ -8,6 +8,11 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any
try:
from .provider_status import provider_status_text
except ImportError: # Direct module loading in the small unit-test harness.
from provider_status import provider_status_text
POLICY_PATH = Path("/opt/data/workspace/coordinator/route-policy.json")
SWITCHYARD_PROVIDER = "atlas-switchyard"
@ -378,3 +383,8 @@ def register(ctx: Any) -> None:
description="Show or constrain Switchyard AUTO routing",
args_hint="auto [posture]|status|manual provider effort [model]",
)
ctx.register_command(
"providers",
lambda _raw_args: provider_status_text(),
description="Show Codex, Claude, local, and Switchyard status",
)

View File

@ -0,0 +1,140 @@
(function () {
"use strict";
const SDK = window.__HERMES_PLUGIN_SDK__;
if (!SDK) return;
const { React } = SDK;
const h = React.createElement;
const { Card, CardContent, Badge, Button } = SDK.components;
const { useCallback, useEffect, useState } = SDK.hooks;
const API = "/api/plugins/auto-router";
function number(value) {
return new Intl.NumberFormat().format(Number(value || 0));
}
function when(value) {
if (!value) return "not reported";
const parsed = new Date(value);
return Number.isNaN(parsed.valueOf()) ? String(value) : parsed.toLocaleString();
}
function Stat(props) {
return h("div", { className: "provider-status-stat" },
h("span", { className: "provider-status-stat-label" }, props.label),
h("strong", null, props.value)
);
}
function ProviderCard(props) {
const item = props.item || {};
const account = item.account || null;
return h(Card, { className: "provider-status-card" },
h(CardContent, { className: "provider-status-card-content" },
h("div", { className: "provider-status-card-heading" },
h("div", null,
h("h2", null, props.title),
account ? h("p", null, "Plan: ", h("strong", null, account.plan || "unknown")) : null
),
h(Badge, { className: "provider-status-badge provider-status-badge--" + (item.state || "unobserved") }, item.state || "unobserved")
),
account ? h("div", { className: "provider-status-auth" },
h("span", { className: account.authenticated ? "is-good" : "is-bad" }, account.authenticated ? "Authentication ready" : "Authentication unavailable"),
account.rate_limit_tier && account.rate_limit_tier !== "unknown" ? h("span", null, "Tier: " + account.rate_limit_tier) : null,
h("span", null, "Access token: " + when(account.token_expires_at)),
account.usage_url ? h("a", {
className: "provider-status-usage-link",
href: account.usage_url,
target: "_blank",
rel: "noopener noreferrer"
}, "Open official usage ↗") : null
) : null,
h("div", { className: "provider-status-stats" },
h(Stat, { label: "Completed", value: number(item.calls) }),
h(Stat, { label: "Errors", value: number(item.errors) }),
h(Stat, { label: "Tokens", value: number(item.total_tokens) }),
h(Stat, { label: "Avg latency", value: item.avg_latency_ms ? number(item.avg_latency_ms) + " ms" : "—" })
),
item.models && item.models.length ? h("div", { className: "provider-status-models" },
item.models.map(function (model) {
return h("div", { className: "provider-status-model", key: model.id },
h("code", null, model.id),
h("span", null, number(model.calls) + " completed · " + number(model.errors) + " errors · " + number(model.total_tokens) + " tokens")
);
})
) : h("p", { className: "provider-status-empty" }, "No calls observed in this router window.")
)
);
}
function ProviderStatus() {
const dataState = useState(null);
const data = dataState[0];
const setData = dataState[1];
const errorState = useState("");
const error = errorState[0];
const setError = errorState[1];
const loadingState = useState(true);
const loading = loadingState[0];
const setLoading = loadingState[1];
const load = useCallback(function () {
setLoading(true);
return SDK.fetchJSON(API + "/status")
.then(function (value) { setData(value); setError(""); })
.catch(function (err) { setError(err && err.message ? err.message : String(err)); })
.finally(function () { setLoading(false); });
}, []);
useEffect(function () {
load();
const timer = window.setInterval(load, 30000);
return function () { window.clearInterval(timer); };
}, [load]);
if (!data && loading) return h("div", { className: "provider-status-loading" }, "Loading provider status…");
const router = data ? data.router : {};
const providers = data ? data.providers : {};
return h("main", { className: "provider-status-page" },
h("header", { className: "provider-status-header" },
h("div", null,
h("h1", null, "Provider Status"),
h("p", null, "The provider, model, fallback, and token activity Switchyard actually observed.")
),
h(Button, { onClick: load, disabled: loading }, loading ? "Refreshing…" : "Refresh")
),
error ? h("div", { className: "provider-status-error" }, error) : null,
data ? h(Card, { className: "provider-status-router" },
h(CardContent, null,
h("div", { className: "provider-status-card-heading" },
h("div", null, h("h2", null, "Switchyard"), h("p", null, data.window)),
h(Badge, { className: "provider-status-badge provider-status-badge--" + router.state }, router.state)
),
h("div", { className: "provider-status-stats" },
h(Stat, { label: "Requests", value: number(router.total_requests) }),
h(Stat, { label: "Errors", value: number(router.total_errors) }),
h(Stat, { label: "Tokens", value: number(router.total_tokens) }),
h(Stat, { label: "Fallbacks", value: number(Object.values(router.fallbacks || {}).reduce(function (a, b) { return a + Number(b || 0); }, 0)) }),
h(Stat, { label: "Classifier calls", value: number(router.classifier_calls) }),
h(Stat, { label: "Classifier errors", value: number(router.classifier_errors) })
)
)
) : null,
data ? h("section", { className: "provider-status-grid" },
h(ProviderCard, { title: "OpenAI Codex", item: providers.codex }),
h(ProviderCard, { title: "Anthropic Claude", item: providers.claude }),
h(ProviderCard, { title: "Local models", item: providers.local })
) : null,
data ? h("aside", { className: "provider-status-note" },
h("strong", null, "About provider quota"),
h("p", null, data.quota_note),
h("p", null, "The native /usage command remains scoped to direct calls made by the current Hermes process. Use /providers for this pool-wide view.")
) : null
);
}
if (window.__HERMES_PLUGINS__ && typeof window.__HERMES_PLUGINS__.register === "function") {
window.__HERMES_PLUGINS__.register("auto-router", ProviderStatus);
}
}());

View File

@ -0,0 +1,117 @@
.provider-status-page {
display: grid;
gap: 1rem;
padding: 1.25rem;
max-width: 1400px;
margin: 0 auto;
}
.provider-status-header,
.provider-status-card-heading {
display: flex;
align-items: flex-start;
justify-content: space-between;
gap: 1rem;
}
.provider-status-header h1,
.provider-status-card-heading h2 {
margin: 0;
}
.provider-status-header p,
.provider-status-card-heading p,
.provider-status-auth,
.provider-status-empty {
color: var(--muted-foreground);
}
.provider-status-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(320px, 1fr));
gap: 1rem;
}
.provider-status-card-content {
display: grid;
gap: 1rem;
}
.provider-status-stats {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(110px, 1fr));
gap: .65rem;
margin-top: 1rem;
}
.provider-status-stat {
display: grid;
gap: .2rem;
padding: .7rem;
border: 1px solid var(--border);
border-radius: .5rem;
background: color-mix(in srgb, var(--muted) 35%, transparent);
}
.provider-status-stat-label {
color: var(--muted-foreground);
font-size: .75rem;
text-transform: uppercase;
letter-spacing: .06em;
}
.provider-status-auth {
display: flex;
flex-wrap: wrap;
gap: .5rem 1rem;
font-size: .85rem;
}
.provider-status-usage-link {
color: var(--primary, #f7c843);
font-weight: 600;
text-decoration: none;
}
.provider-status-usage-link:hover {
text-decoration: underline;
}
.provider-status-auth .is-good { color: #22c55e; }
.provider-status-auth .is-bad { color: #ef4444; }
.provider-status-models {
display: grid;
gap: .5rem;
}
.provider-status-model {
display: grid;
gap: .2rem;
padding-top: .5rem;
border-top: 1px solid var(--border);
font-size: .8rem;
}
.provider-status-model span { color: var(--muted-foreground); }
.provider-status-badge--available { background: #166534; color: white; }
.provider-status-badge--degraded { background: #a16207; color: white; }
.provider-status-badge--unavailable { background: #991b1b; color: white; }
.provider-status-badge--unobserved { background: #374151; color: white; }
.provider-status-note,
.provider-status-error,
.provider-status-loading {
padding: 1rem;
border: 1px solid var(--border);
border-radius: .65rem;
}
.provider-status-note p { margin: .4rem 0 0; color: var(--muted-foreground); }
.provider-status-error { border-color: #991b1b; color: #fca5a5; }
@media (max-width: 640px) {
.provider-status-page { padding: .75rem; }
.provider-status-header { align-items: stretch; flex-direction: column; }
}

View File

@ -0,0 +1,14 @@
{
"name": "auto-router",
"label": "Provider Status",
"description": "Codex, Claude, local model, routing fallback, and token activity observed by Switchyard.",
"icon": "Activity",
"version": "1.0.0",
"tab": {
"path": "/provider-status",
"position": "after:models"
},
"entry": "dist/index.js",
"css": "dist/style.css",
"api": "plugin_api.py"
}

View File

@ -0,0 +1,23 @@
"""Owner-authenticated dashboard API for Switchyard provider telemetry."""
from __future__ import annotations
import sys
from pathlib import Path
from fastapi import APIRouter
# Dashboard API files are loaded as standalone modules by Hermes. Put the
# owning plugin directory on the import path so this API and the TUI command
# share one normalization implementation.
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from provider_status import provider_status_payload
router = APIRouter()
@router.get("/status")
def status() -> dict[str, object]:
"""Return only non-secret account health and observed router counters."""
return provider_status_payload()

View File

@ -0,0 +1,281 @@
"""Read-only provider-pool status derived from Switchyard and local auth metadata."""
from __future__ import annotations
import base64
import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
SWITCHYARD_ROOT = os.environ.get(
"HERMES_SWITCHYARD_STATUS_URL",
"http://hermes-switchyard.hermes.svc.cluster.local:9005",
).rstrip("/")
CODEX_USAGE_URL = "https://chatgpt.com/codex/settings/usage"
CLAUDE_USAGE_URL = "https://claude.ai/settings/usage"
CODEX_AUTH_PATH = Path(
os.environ.get("CODEX_HOME", "/opt/data/home/.codex")
) / "auth.json"
CLAUDE_AUTH_PATH = Path(
os.environ.get("CLAUDE_CONFIG_DIR", "/opt/data/home/.claude")
) / ".credentials.json"
def _read_json(path: Path) -> dict[str, Any]:
"""Return one local JSON object without leaking parsing details."""
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, TypeError, ValueError, json.JSONDecodeError):
return {}
return value if isinstance(value, dict) else {}
def _get_json(url: str, timeout: float = 3.0) -> dict[str, Any]:
"""Fetch a bounded JSON status response from the in-cluster router."""
request = Request(url, headers={"Accept": "application/json"})
try:
with urlopen(request, timeout=timeout) as response:
value = json.load(response)
except (HTTPError, URLError, OSError, TimeoutError, ValueError):
return {}
return value if isinstance(value, dict) else {}
def _jwt_claims(token: Any) -> dict[str, Any]:
"""Decode trusted local token metadata for display; do not verify or return it."""
if not isinstance(token, str):
return {}
try:
encoded = token.split(".")[1]
encoded += "=" * (-len(encoded) % 4)
value = json.loads(base64.urlsafe_b64decode(encoded))
except (
IndexError,
TypeError,
ValueError,
json.JSONDecodeError,
base64.binascii.Error,
):
return {}
return value if isinstance(value, dict) else {}
def _timestamp(value: Any) -> tuple[str | None, bool | None]:
"""Normalize an epoch or ISO-8601 timestamp and report whether it is live."""
if isinstance(value, str):
rendered_value = value.strip()
if rendered_value and not rendered_value.replace(".", "", 1).isdigit():
try:
parsed = datetime.fromisoformat(rendered_value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
except ValueError:
return None, None
rendered = parsed.astimezone(timezone.utc).isoformat()
return rendered, parsed.timestamp() > time.time()
try:
raw = float(value)
except (TypeError, ValueError):
return None, None
if raw > 100_000_000_000:
raw /= 1000
try:
rendered = datetime.fromtimestamp(raw, tz=timezone.utc).isoformat()
except (OverflowError, OSError, ValueError):
return None, None
return rendered, raw > time.time()
def _codex_account() -> dict[str, Any]:
"""Return non-secret Codex plan and credential health metadata."""
auth = _read_json(CODEX_AUTH_PATH)
tokens = auth.get("tokens") if isinstance(auth.get("tokens"), dict) else {}
access = _jwt_claims(tokens.get("access_token"))
identity = _jwt_claims(tokens.get("id_token"))
claims = identity.get("https://api.openai.com/auth")
claims = claims if isinstance(claims, dict) else {}
expires_at, token_live = _timestamp(access.get("exp"))
subscription_until, subscription_live = _timestamp(
claims.get("chatgpt_subscription_active_until")
)
authenticated = bool(tokens.get("access_token")) and token_live is not False
return {
"authenticated": authenticated,
"auth_mode": auth.get("auth_mode") or "unknown",
"plan": claims.get("chatgpt_plan_type") or "unknown",
"token_expires_at": expires_at,
"subscription_active": subscription_live,
"subscription_until": subscription_until,
"last_refresh": auth.get("last_refresh"),
"quota_reported": False,
"usage_url": CODEX_USAGE_URL,
}
def _claude_account() -> dict[str, Any]:
"""Return non-secret Claude plan and credential health metadata."""
auth = _read_json(CLAUDE_AUTH_PATH)
oauth = auth.get("claudeAiOauth")
oauth = oauth if isinstance(oauth, dict) else {}
expires_at, token_live = _timestamp(oauth.get("expiresAt"))
refresh_expires_at, refresh_live = _timestamp(oauth.get("refreshTokenExpiresAt"))
authenticated = bool(oauth.get("accessToken")) and token_live is not False
return {
"authenticated": authenticated,
"plan": oauth.get("subscriptionType") or "unknown",
"rate_limit_tier": oauth.get("rateLimitTier") or "unknown",
"token_expires_at": expires_at,
"refresh_expires_at": refresh_expires_at,
"refresh_live": refresh_live,
"quota_reported": False,
"usage_url": CLAUDE_USAGE_URL,
}
def _provider_for_model(model_id: str) -> str:
"""Map Switchyard target identifiers to their operator-facing provider."""
value = model_id.lower()
if "/codex/" in value or value.startswith("openai"):
return "codex"
if "/claude/" in value or value.startswith("anthropic"):
return "claude"
if "/local/" in value or "qwen" in value or "ollama" in value:
return "local"
return "other"
def _number(value: Any) -> int:
"""Coerce non-negative counters from Switchyard's JSON safely."""
try:
return max(0, int(value or 0))
except (TypeError, ValueError):
return 0
def _provider_summary(name: str, models: dict[str, Any]) -> dict[str, Any]:
"""Aggregate successful calls, errors, tokens, and latency for one lane."""
selected: list[dict[str, Any]] = []
totals = {
"calls": 0,
"errors": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"reasoning_tokens": 0,
"total_tokens": 0,
}
latency_weight = 0.0
latency_calls = 0
for model_id, raw in sorted(models.items()):
if _provider_for_model(str(model_id)) != name or not isinstance(raw, dict):
continue
item = {"id": str(model_id)}
for key in totals:
item[key] = _number(raw.get(key))
totals[key] += item[key]
try:
item["avg_latency_ms"] = round(float(raw.get("avg_latency_ms") or 0), 1)
except (TypeError, ValueError):
item["avg_latency_ms"] = 0.0
latency_weight += item["avg_latency_ms"] * item["calls"]
latency_calls += item["calls"]
selected.append(item)
calls = totals["calls"]
errors = totals["errors"]
if calls and errors:
state = "degraded"
elif calls:
state = "available"
elif errors:
state = "unavailable"
else:
state = "unobserved"
return {
"state": state,
**totals,
"avg_latency_ms": round(latency_weight / latency_calls, 1)
if latency_calls
else 0.0,
"models": selected,
}
def provider_status_payload() -> dict[str, Any]:
"""Build the owner-safe status document shared by dashboard and TUI."""
health = _get_json(f"{SWITCHYARD_ROOT}/health")
stats = _get_json(f"{SWITCHYARD_ROOT}/v1/stats")
router_ok = health.get("status") == "ok" and bool(stats)
models = stats.get("models") if isinstance(stats.get("models"), dict) else {}
providers = {
name: _provider_summary(name, models)
for name in ("codex", "claude", "local")
}
providers["codex"]["account"] = _codex_account()
providers["claude"]["account"] = _claude_account()
classifier = stats.get("classifier")
classifier = classifier if isinstance(classifier, dict) else {}
fallbacks = stats.get("routing_fallbacks")
fallbacks = fallbacks if isinstance(fallbacks, dict) else {}
return {
"generated_at": datetime.now(timezone.utc).isoformat(),
"window": "Since the last Switchyard restart",
"quota_note": (
"Codex and Claude subscription balances are not exposed to this "
"router. Open the provider usage page for authoritative remaining "
"capacity; the counters here show actual work observed by Switchyard."
),
"router": {
"state": "available" if router_ok else "unavailable",
"total_requests": _number(stats.get("total_requests")),
"total_errors": _number(stats.get("total_errors")),
"total_tokens": _number(
(stats.get("total_tokens") or {}).get("total")
if isinstance(stats.get("total_tokens"), dict)
else 0
),
"fallbacks": {str(key): _number(value) for key, value in fallbacks.items()},
"classifier_calls": _number(classifier.get("total_requests")),
"classifier_errors": _number(classifier.get("total_errors")),
},
"providers": providers,
}
def provider_status_text() -> str:
"""Render the same status as a compact native Hermes command response."""
payload = provider_status_payload()
router = payload["router"]
labels = {"codex": "Codex", "claude": "Claude", "local": "Local"}
lines = [
f"Provider pool: Switchyard {router['state']}",
str(payload["window"]),
]
for key in ("codex", "claude", "local"):
item = payload["providers"][key]
account = item.get("account") or {}
plan = f" · plan {account.get('plan', 'unknown')}" if account else ""
auth = ""
if account:
auth = " · auth ready" if account.get("authenticated") else " · auth unavailable"
lines.append(
f"{labels[key]}: {item['state']} · {item['calls']} completed · "
f"{item['errors']} errors · {item['total_tokens']} tokens{plan}{auth}"
)
lines.extend(
[
f"Fallbacks: {sum(router['fallbacks'].values())} · "
f"classifier: {router['classifier_calls']} calls / "
f"{router['classifier_errors']} errors",
str(payload["quota_note"]),
f"Codex usage: {CODEX_USAGE_URL}",
f"Claude usage: {CLAUDE_USAGE_URL}",
"Dashboard: Provider Status",
]
)
return "\n".join(lines)

View File

@ -10,6 +10,8 @@ from types import SimpleNamespace
SOURCE = Path(__file__).parents[2] / "services/hermes/plugins/auto-router/__init__.py"
PLUGIN_ROOT = SOURCE.parent
sys.path.insert(0, str(PLUGIN_ROOT.parent))
SPEC = importlib.util.spec_from_file_location("hermes_auto_router", SOURCE)
assert SPEC and SPEC.loader
router = importlib.util.module_from_spec(SPEC)
@ -145,11 +147,89 @@ def test_registers_every_model_call_boundary_and_route_command():
"pre_subagent_route",
}
assert "route" in commands
assert "providers" in commands
def test_adapter_contains_no_content_classifier_or_direct_ollama_call():
source = SOURCE.read_text(encoding="utf-8")
assert "jetson_decision" not in source
assert "urllib.request" not in source
assert "ollama.ai.svc" not in source
def test_provider_status_separates_observed_activity_from_plan_quota(monkeypatch):
status = sys.modules["hermes_auto_router"].provider_status_text.__globals__
module = sys.modules[status["provider_status_payload"].__module__]
monkeypatch.setattr(module, "_get_json", lambda url, timeout=3.0: (
{"status": "ok"} if url.endswith("/health") else {
"total_requests": 3,
"total_errors": 1,
"total_tokens": {"total": 900},
"models": {
"route/codex/sol/xhigh": {
"calls": 1,
"errors": 0,
"total_tokens": 600,
"avg_latency_ms": 1200,
},
"route/claude/sonnet/high": {
"calls": 0,
"errors": 1,
"total_tokens": 0,
},
"route/local/qwen2.5-14b/medium": {
"calls": 1,
"errors": 0,
"total_tokens": 300,
},
},
"routing_fallbacks": {"unavailable": 1},
"classifier": {"total_requests": 3, "total_errors": 0},
}
))
monkeypatch.setattr(module, "_codex_account", lambda: {
"authenticated": True, "plan": "plus", "quota_reported": False,
})
monkeypatch.setattr(module, "_claude_account", lambda: {
"authenticated": True, "plan": "max", "quota_reported": False,
})
payload = module.provider_status_payload()
assert payload["router"]["state"] == "available"
assert payload["providers"]["codex"]["calls"] == 1
assert payload["providers"]["claude"]["state"] == "unavailable"
assert payload["providers"]["local"]["total_tokens"] == 300
assert payload["providers"]["codex"]["account"]["quota_reported"] is False
assert "not exposed" in payload["quota_note"]
def test_provider_status_accepts_iso_and_epoch_credential_expiry():
status = sys.modules["hermes_auto_router"].provider_status_text.__globals__
module = sys.modules[status["provider_status_payload"].__module__]
iso_value, iso_live = module._timestamp("2999-01-01T00:00:00Z")
epoch_value, epoch_live = module._timestamp(32_472_192_000)
assert iso_value == "2999-01-01T00:00:00+00:00"
assert iso_live is True
assert epoch_value and epoch_value.startswith("2999-01-01T")
assert epoch_live is True
def test_agent_mounts_provider_status_dashboard_into_auto_router_plugin():
import yaml
root = Path(__file__).parents[2]
deployment = yaml.safe_load(
(root / "services/hermes/agent-deployment.yaml").read_text(encoding="utf-8")
)
volume = next(
item for item in deployment["spec"]["template"]["spec"]["volumes"]
if item["name"] == "auto-router-plugin"
)
paths = {item["path"] for item in volume["configMap"]["items"]}
assert "provider_status.py" in paths
assert "dashboard/manifest.json" in paths
assert "dashboard/plugin_api.py" in paths
assert "dashboard/dist/index.js" in paths