hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 31 Packages Projects Releases Wiki Activity

refactor(plugins/hermes-achievements,dashboard_auth): unify snapshot/scan_meta/result builders, drop dead evaluate_boolean + import shims, compact module docstrings, AST-neutral layout

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:43:13 -07:00
parent e86d3fab86
commit eccfcb5f8e
7 changed files with 242 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
plugins/dashboard_auth/_shared.py
View File

@@ -109,12 +109,10 @@ def pkce_login_start(authorize_url: str, *, client_id: str, scope: str, redirect
params = {
"response_type": "code", "client_id": client_id, "redirect_uri": redirect_uri, "scope": scope, "state": state,
"code_challenge": b64url_no_pad(hashlib.sha256(code_verifier.encode("ascii")).digest()),
"code_challenge_method": "S256",
}
"code_challenge_method": "S256"}
return LoginStart(
redirect_url=f"{authorize_url}?{urllib.parse.urlencode(params)}",
cookie_payload={"hermes_session_pkce": f"state={state};verifier={code_verifier}"},
)
cookie_payload={"hermes_session_pkce": f"state={state};verifier={code_verifier}"})
def parse_json_body(response: httpx.Response) -> Dict[str, Any]:
@@ -130,8 +128,7 @@ def parse_json_body(response: httpx.Response) -> Dict[str, Any]:
def exchange_token(
url: str, data: Dict[str, str], *, headers: Optional[Dict[str, str]] = None, bad_request_exc: type[Exception],
idp: str, endpoint: str, token_key: str, missing_msg: str,
) -> tuple[str, Dict[str, Any]]:
idp: str, endpoint: str, token_key: str, missing_msg: str) -> tuple[str, Dict[str, Any]]:
"""POST a token grant and return ``(token, payload)``.
A 400 (OAuth-shaped error envelope) raises ``bad_request_exc`` — ``InvalidCodeError``
@@ -168,16 +165,14 @@ def refresh_token_from(payload: Dict[str, Any], fallback: str = "") -> str:
def session_from_claims(
provider: str, claims: Dict[str, Any], *, access_token: str, refresh_token: str,
label: str = "token", email: str = "", display_name: str = "", org_id: str = "",
) -> Session:
label: str = "token", email: str = "", display_name: str = "", org_id: str = "") -> Session:
"""Map verified JWT claims onto a Session; ``sub`` is mandatory."""
user_id = str(claims.get("sub", ""))
if not user_id:
raise ProviderError(f"{label} missing 'sub' (user_id) claim")
return Session(
user_id=user_id, email=email, display_name=display_name, org_id=org_id, provider=provider,
expires_at=int(claims["exp"]), access_token=access_token, refresh_token=refresh_token,
)
expires_at=int(claims["exp"]), access_token=access_token, refresh_token=refresh_token)
# ---- JWT verification ----
@@ -189,13 +184,11 @@ def make_jwks_client(jwks_url: str) -> Any:
return PyJWKClient(
jwks_url, cache_keys=True, lifespan=JWKS_CACHE_SECONDS,
headers={"Accept": "application/json", "User-Agent": "HermesAgent/1.0"},
)
headers={"Accept": "application/json", "User-Agent": "HermesAgent/1.0"})
def verify_jwt(
token: str, jwks_client: Any, *, algorithms: list[str], audience: str, issuer: str, label: str,
) -> Dict[str, Any]:
token: str, jwks_client: Any, *, algorithms: list[str], audience: str, issuer: str, label: str) -> Dict[str, Any]:
"""Verify ``token`` against ``jwks_client`` with pinned ``aud``/``iss``.
Unreachable JWKS → ``ProviderError`` (503); a bearer that is not one of our JWTs
@@ -213,8 +206,7 @@ def verify_jwt(
try:
return jwt.decode(
token, signing_key.key, algorithms=algorithms, audience=audience, issuer=issuer,
options={"require": ["exp", "iat", "aud", "iss", "sub"]},
)
options={"require": ["exp", "iat", "aud", "iss", "sub"]})
except jwt.ExpiredSignatureError as exc:
raise InvalidCodeError(f"{label} expired: {exc}") from exc
except jwt.InvalidTokenError as exc:
@@ -225,8 +217,7 @@ def verify_jwt(
unverified = jwt.decode(token, options={"verify_signature": False, "verify_exp": False})
details = (
f" [token iss={unverified.get('iss')!r} aud={unverified.get('aud')!r}; "
f"expected iss={issuer!r} aud={audience!r}]"
)
f"expected iss={issuer!r} aud={audience!r}]")
except Exception:
pass
raise ProviderError(f"{label} verification failed: {exc}{details}") from exc

59
plugins/dashboard_auth/basic/__init__.py
View File

@@ -1,28 +1,11 @@
"""BasicAuthProvider — username/password dashboard auth (no OAuth IDP).
Same ``DashboardAuthProvider`` framework as the OAuth providers, but login is a credential
form (``supports_password = True`` + ``complete_password_login``); cookies, verify, refresh,
ws-tickets and logout are shared. No IDP and no database: sessions are stateless
HMAC-signed tokens this provider mints and verifies. Passwords use stdlib scrypt;
``complete_password_login`` always performs a hash even for an unknown username, so the
endpoint is not a username-enumeration timing oracle. ``LAST_SKIP_REASON`` is read by
the gate's fail-closed branch when the plugin loads but declines to register.
Configuration (env wins over config.yaml when set non-empty)::
dashboard:
basic_auth:
username: admin # required
password_hash: "scrypt$..." # preferred — see hash_password()
password: "s3cret" # OR plaintext, hashed in-memory at load
secret: "<32+ random bytes, base64 or hex>" # optional signing key
session_ttl_seconds: 43200 # optional access-token lifetime (12h)
HERMES_DASHBOARD_BASIC_AUTH_USERNAME / _PASSWORD_HASH (preferred) / _PASSWORD (plaintext)
HERMES_DASHBOARD_BASIC_AUTH_SECRET / _TTL_SECONDS
Without ``secret`` a random per-process key is generated: sessions then don't
survive a restart or span multiple worker processes.
Login is a credential form (``supports_password`` + ``complete_password_login``); cookies,
verify, refresh, ws-tickets and logout are the shared framework. Sessions are stateless
HMAC-signed tokens (no IDP, no database); passwords use stdlib scrypt and login always hashes
even for an unknown username (no username-enumeration timing oracle). Config: ``dashboard.
basic_auth.{username,password_hash|password,secret,session_ttl_seconds}`` or the
``HERMES_DASHBOARD_BASIC_AUTH_*`` env vars (env wins when non-empty; see ``_settings``).
"""
from __future__ import annotations
@@ -37,7 +20,8 @@ import secrets
import time
from typing import Optional
from hermes_cli.dashboard_auth import DashboardAuthProvider, InvalidCredentialsError, LoginStart, RefreshExpiredError, Session
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCredentialsError, LoginStart, RefreshExpiredError, Session)
from plugins.dashboard_auth._shared import SkipRegistration, load_config_section, register_provider, resolve_env_or_cfg
logger = logging.getLogger(__name__)
@@ -69,8 +53,10 @@ def hash_password(password: str) -> str:
precompute ``password_hash`` for config.yaml (the plaintext then never sits at rest):
``python -c "from plugins.dashboard_auth.basic import hash_password; print(hash_password('pw'))"``."""
salt = secrets.token_bytes(_SCRYPT_SALT_BYTES)
dk = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_DKLEN, maxmem=0)
return f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${base64.b64encode(salt).decode()}${base64.b64encode(dk).decode()}"
dk = hashlib.scrypt(
password.encode("utf-8"), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_DKLEN, maxmem=0)
salt_b64, dk_b64 = base64.b64encode(salt).decode(), base64.b64encode(dk).decode()
return f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${salt_b64}${dk_b64}"
def _verify_password(password: str, encoded: str) -> bool:
@@ -149,8 +135,7 @@ class BasicAuthProvider(DashboardAuthProvider):
def start_login(self, *, redirect_uri: str) -> LoginStart:
raise NotImplementedError(
"BasicAuthProvider is password-only; there is no OAuth redirect flow. "
"The login page POSTs to /auth/password-login instead."
)
"The login page POSTs to /auth/password-login instead.")
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
raise NotImplementedError("BasicAuthProvider is password-only; use complete_password_login.")
@@ -195,14 +180,12 @@ class BasicAuthProvider(DashboardAuthProvider):
return self._session(
user_id, exp,
_sign({"sub": user_id, "kind": "access", "exp": exp}, self._secret),
_sign({"sub": user_id, "kind": "refresh", "exp": now + _REFRESH_TTL_SECONDS}, self._secret),
)
_sign({"sub": user_id, "kind": "refresh", "exp": now + _REFRESH_TTL_SECONDS}, self._secret))
def _session(self, user_id: str, exp: int, access_token: str, refresh_token: str) -> Session:
return Session(
user_id=user_id, email="", display_name=user_id, org_id="", provider=self.name,
expires_at=exp, access_token=access_token, refresh_token=refresh_token,
)
expires_at=exp, access_token=access_token, refresh_token=refresh_token)
# ---- Plugin entry point ----
@@ -221,8 +204,7 @@ def _resolve_secret(cfg_section: dict) -> bytes:
"dashboard-auth-basic: no 'secret' configured; generating a random "
"per-process signing key. Sessions will not survive a restart or span "
"multiple workers. Set dashboard.basic_auth.secret (or "
"HERMES_DASHBOARD_BASIC_AUTH_SECRET) for stable sessions."
)
"HERMES_DASHBOARD_BASIC_AUTH_SECRET) for stable sessions.")
return secrets.token_bytes(32)
for decoder in (base64.b64decode, bytes.fromhex):
try:
@@ -250,15 +232,13 @@ def _settings() -> dict:
"dashboard.basic_auth.username is not set (and HERMES_DASHBOARD_BASIC_AUTH_USERNAME "
"is empty). Set a username and a password (or password_hash) under "
"dashboard.basic_auth in config.yaml to enable username/password dashboard "
"login, or use the OAuth provider, or pass --insecure to skip the auth gate."
)
"login, or use the OAuth provider, or pass --insecure to skip the auth gate.")
if not password_hash and not plaintext:
raise SkipRegistration(
"dashboard.basic_auth.username is set but neither password_hash nor password "
"is configured. Provide one of them (password_hash is preferred — compute it "
"with plugins.dashboard_auth.basic.hash_password).",
level="warning",
)
level="warning")
# Precedence: env password (hashed in-memory) overrides any config password_hash so
# operators can rotate without editing config; a config password_hash wins over a
# config-only plaintext password (preferred at-rest form).
@@ -271,8 +251,7 @@ def _settings() -> dict:
logger.info(
"dashboard-auth-basic: hashed plaintext password in-memory. "
"For production, precompute dashboard.basic_auth.password_hash "
"and remove the plaintext password from config."
)
"and remove the plaintext password from config.")
try:
ttl = int(ttl_raw) if ttl_raw else _DEFAULT_TTL_SECONDS
except ValueError:

29
plugins/dashboard_auth/drain/__init__.py
View File

@@ -1,16 +1,11 @@
"""DrainSecretProvider — shared-bearer-secret auth for the drain-control endpoint.
Uses the non-interactive token capability of the ``DashboardAuthProvider`` ABC
(``supports_token`` / ``verify_token`` + the route-agnostic ``token_auth`` middleware
seam). ``nous-account-service`` (NAS) provisions a **per-agent unique** shared secret into
each agent's environment; this provider verifies an inbound bearer against it with a
constant-time compare and vouches for the caller as the ``drain-control`` principal. No
login/cookie/session/refresh — the interactive ABC methods raise ``NotImplementedError``.
Fail-CLOSED entropy gate at registration: a weak/short/low-entropy secret is never
silently accepted (>= 43 url-safe-base64 chars ~= 256 bits, enough distinct characters,
Shannon entropy floor). The secret is a CREDENTIAL, so it is env-only
(``HERMES_DASHBOARD_DRAIN_SECRET``); knobs ``scope`` / ``min_secret_chars`` live under
``dashboard.drain_auth`` in config.yaml. Unset env var → no-op (records a skip reason).
Non-interactive token capability of the ``DashboardAuthProvider`` ABC (``verify_token`` +
the ``token_auth`` middleware seam): ``nous-account-service`` provisions a per-agent unique
secret (``HERMES_DASHBOARD_DRAIN_SECRET``, env-only — it is a credential); an inbound bearer
is compared constant-time and vouched for as the ``drain-control`` principal. Fail-CLOSED
entropy gate at registration (length, distinct chars, Shannon bits); interactive ABC methods
raise. Knobs ``scope`` / ``min_secret_chars`` live under ``dashboard.drain_auth``.
"""
from __future__ import annotations
@@ -60,8 +55,7 @@ def assess_secret_strength(secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_
return (
f"secret too short: {len(secret)} chars (need >= {min_chars}; "
"use a >=256-bit value, e.g. `python -c \"import secrets; "
"print(secrets.token_urlsafe(32))\"`)"
)
"print(secrets.token_urlsafe(32))\"`)")
distinct = len(set(secret))
if distinct < _MIN_DISTINCT_CHARS:
return f"secret has only {distinct} distinct characters (need >= {_MIN_DISTINCT_CHARS}); looks structured/low-entropy"
@@ -130,8 +124,7 @@ def _settings() -> dict:
raise SkipRegistration(
"HERMES_DASHBOARD_DRAIN_SECRET is not set. Set a per-agent >=256-bit secret "
"(e.g. `python -c \"import secrets; print(secrets.token_urlsafe(32))\"`) to enable "
"NAS-driven drain coordination; leave it unset to disable the drain endpoint."
)
"NAS-driven drain coordination; leave it unset to disable the drain endpoint.")
section = _load_config_drain_auth_section()
scope = str(section.get("scope", "drain") or "drain").strip() or "drain"
try:
@@ -142,8 +135,7 @@ def _settings() -> dict:
if reason is not None:
raise SkipRegistration(
f"HERMES_DASHBOARD_DRAIN_SECRET rejected — {reason}. The drain endpoint stays disabled (fail-closed).",
level="warning",
)
level="warning")
return {"secret": secret, "scope": scope}
@@ -166,5 +158,4 @@ def register(ctx) -> None:
logger.warning("dashboard-auth-drain: could not register token route %s: %s", DRAIN_ROUTE_PATH, exc)
logger.info(
"dashboard-auth-drain: registered drain service-credential provider (scope=%s, route=%s)",
kwargs["scope"], DRAIN_ROUTE_PATH,
)
kwargs["scope"], DRAIN_ROUTE_PATH)

58
plugins/dashboard_auth/nous/__init__.py
View File

@@ -1,18 +1,11 @@
"""NousDashboardAuthProvider — Nous Portal OAuth (authorization-code + PKCE).
Implements ``nous-account-service/docs/agent-dashboard-oauth-contract.md``. Bundled and
auto-loaded, but only registers when a client_id is configured (``dashboard.oauth.client_id``
or ``HERMES_DASHBOARD_OAUTH_CLIENT_ID``, shape ``agent:{agent_instance_id}``; optional
``portal_url`` / ``HERMES_DASHBOARD_PORTAL_URL``). Env wins over config when set non-empty.
Contract points: the client_id suffix is cross-checked against the token's
``agent_instance_id`` claim; scope is ``agent_dashboard:access`` only, audience is the bare
client_id; access tokens are RS256 JWTs verified against ``/.well-known/jwks.json`` (cached
5 min). Portal issues a 24h *rotating* refresh token with reuse detection: the middleware
MUST persist ``Session.refresh_token`` back to the cookie on every refresh, or the next
refresh replays a rotated token and (outside a 60s grace) revokes the whole session; a 400
on refresh → ``RefreshExpiredError`` → re-login. ``oauth_contract_version``: missing →
warn + proceed; ``!= 1`` → refuse. ``LAST_SKIP_REASON`` is read by the gate's fail-closed branch.
Implements ``nous-account-service/docs/agent-dashboard-oauth-contract.md``; registers only
when a client_id (``dashboard.oauth.client_id`` / ``HERMES_DASHBOARD_OAUTH_CLIENT_ID``, shape
``agent:{instance_id}``) is configured. Access tokens are RS256 JWTs verified against the
Portal JWKS with ``aud`` = bare client_id. Portal issues a 24h *rotating* refresh token with
reuse detection: the middleware MUST persist ``Session.refresh_token`` back to the cookie on
every refresh or the next refresh replays a rotated token and revokes the whole session.
"""
from __future__ import annotations
@@ -20,7 +13,8 @@ from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from hermes_cli.dashboard_auth import DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session)
from plugins.dashboard_auth._shared import (
SkipRegistration,
exchange_token,
@@ -32,8 +26,7 @@ from plugins.dashboard_auth._shared import (
resolve_env_or_cfg,
session_from_claims,
validate_redirect_uri,
verify_jwt,
)
verify_jwt)
logger = logging.getLogger(__name__)
_TAG = "dashboard-auth-nous"
@@ -74,10 +67,8 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
return self._token_grant(
{
"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
"client_id": self._client_id, "code_verifier": code_verifier,
},
bad_request_exc=InvalidCodeError,
)
"client_id": self._client_id, "code_verifier": code_verifier},
bad_request_exc=InvalidCodeError)
def refresh_session(self, *, refresh_token: str) -> Session:
if not refresh_token:
@@ -88,8 +79,7 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
return self._token_grant(
{"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token},
headers={"x-nous-refresh-token": refresh_token},
bad_request_exc=RefreshExpiredError,
)
bad_request_exc=RefreshExpiredError)
def verify_session(self, *, access_token: str) -> Optional[Session]:
# None on expiry/invalidity (middleware then tries refresh); a ProviderError
@@ -113,8 +103,7 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
access_token, payload = exchange_token(
self._token_url, data, headers=headers, bad_request_exc=bad_request_exc,
idp="Portal", endpoint="Portal token endpoint", token_key="access_token",
missing_msg="Portal token response missing access_token",
)
missing_msg="Portal token response missing access_token")
# Rotating RT the caller MUST persist back to the cookie.
return self._session(access_token, refresh_token_from(payload), self._verify_jwt(access_token))
@@ -127,25 +116,21 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
claims = verify_jwt(
access_token, self._get_jwks_client(), algorithms=["RS256"],
audience=self._client_id, # contract C2: bare client_id
issuer=self._portal_url, label="access token",
)
issuer=self._portal_url, label="access token")
# Contract C9: agent_instance_id is "should" not "must" — tolerated when absent
# (the aud check already binds the token to this instance).
token_instance_id = claims.get("agent_instance_id")
if token_instance_id is not None and token_instance_id != self._agent_instance_id:
raise ProviderError(
f"agent_instance_id mismatch: token={token_instance_id!r} vs configured={self._agent_instance_id!r}"
)
f"agent_instance_id mismatch: token={token_instance_id!r} vs configured={self._agent_instance_id!r}")
contract_version = claims.get("oauth_contract_version")
if contract_version is None:
logger.warning(
"Nous Portal token missing oauth_contract_version claim (contract says it should be %d); proceeding anyway.",
_EXPECTED_CONTRACT_VERSION,
)
_EXPECTED_CONTRACT_VERSION)
elif contract_version != _EXPECTED_CONTRACT_VERSION:
raise ProviderError(
f"unsupported oauth_contract_version={contract_version!r}, expected {_EXPECTED_CONTRACT_VERSION}"
)
f"unsupported oauth_contract_version={contract_version!r}, expected {_EXPECTED_CONTRACT_VERSION}")
return claims
def _session(self, access_token: str, refresh_token: str, claims: Dict[str, Any]) -> Session:
@@ -173,15 +158,13 @@ def _settings() -> dict:
"'agent:{instance_id}') when it deploys a Hermes Agent instance — set it to "
"your provisioned client id (either as an env var or under "
"dashboard.oauth.client_id in config.yaml), or pass --insecure to skip the "
"OAuth gate entirely."
)
"OAuth gate entirely.")
if not client_id.startswith("agent:"):
raise SkipRegistration(
f"HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id!r} doesn't match the contract "
f"shape 'agent:{{instance_id}}'. The Nous Portal provisions this value at deploy "
f"time; check your Fly app's secrets or override with the value from the Portal admin UI.",
level="warning",
)
level="warning")
return {"client_id": client_id, "portal_url": portal_url}
@@ -191,4 +174,5 @@ def register(ctx) -> None:
LAST_SKIP_REASON = ""
kwargs, LAST_SKIP_REASON = register_provider(ctx, logger, _TAG, NousDashboardAuthProvider, _settings)
if kwargs is not None:
logger.info("dashboard-auth-nous: registered provider (client_id=%s, portal=%s)", kwargs["client_id"], kwargs["portal_url"])
logger.info(
"dashboard-auth-nous: registered provider (client_id=%s, portal=%s)", kwargs["client_id"], kwargs["portal_url"])

77
plugins/dashboard_auth/self_hosted/__init__.py
View File

@@ -1,32 +1,11 @@
"""SelfHostedOIDCProvider — generic self-hosted OpenID Connect dashboard auth.
A standards-compliant OIDC Relying Party for the ``hermes dashboard`` gate (Authentik,
Keycloak, Zitadel, Authelia, Auth0, Okta, Google, …). The HTTP round trip, cookies, CSRF
``state`` check and ``redirect_uri`` reconstruction are owned by
``hermes_cli/dashboard_auth/routes.py``; this provider discovers endpoints from
``{issuer}/.well-known/openid-configuration``, builds the PKCE (S256) ``/authorize`` URL,
exchanges the code at the discovered ``token_endpoint``, and verifies the **ID token**
against the discovered ``jwks_uri`` with ``iss``/``aud`` pinned (the access token's format
is opaque per spec; the ``nous`` provider verifies its access token only because Portal
mints a custom JWT). Public (PKCE-only) and confidential (PKCE + ``client_secret`` via
``client_secret_basic`` / ``client_secret_post``) clients are both supported — the secret is
layered on top of PKCE, never a replacement (OAuth 2.1 / RFC 9700).
Configuration (env wins over config.yaml when set non-empty)::
dashboard:
oauth:
provider: self-hosted
self_hosted:
issuer: https://auth.example.com/application/o/hermes/ # required
client_id: hermes-dashboard # required
scopes: "openid profile email" # optional
# client_secret: confidential clients only — prefer the env var.
HERMES_DASHBOARD_OIDC_ISSUER / _CLIENT_ID / _SCOPES (optional)
HERMES_DASHBOARD_OIDC_CLIENT_SECRET # optional; .env is its canonical home
On skip (missing issuer / client_id) ``LAST_SKIP_REASON`` carries the reason for the gate.
A plain OIDC Relying Party (Authentik, Keycloak, Zitadel, Authelia, Auth0, Okta, …): discovers
endpoints from ``{issuer}/.well-known/openid-configuration``, builds the PKCE (S256) authorize
URL, exchanges the code, and verifies the **ID token** (the access token is opaque per spec)
against the discovered ``jwks_uri`` with ``iss``/``aud`` pinned. Public and confidential
(``client_secret`` layered on top of PKCE, never replacing it) clients both work. Config:
``dashboard.oauth.self_hosted.{issuer,client_id,scopes,client_secret}`` or ``HERMES_DASHBOARD_OIDC_*``.
"""
from __future__ import annotations
@@ -40,7 +19,8 @@ from typing import Any, Dict, Optional
import httpx
from hermes_cli.dashboard_auth import DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session)
from plugins.dashboard_auth._shared import (
JSON_HEADERS,
TOKEN_ENDPOINT_TIMEOUT_SEC as _TOKEN_ENDPOINT_TIMEOUT_SEC,
@@ -55,8 +35,7 @@ from plugins.dashboard_auth._shared import (
resolve_env_or_cfg,
session_from_claims,
validate_redirect_uri,
verify_jwt,
)
verify_jwt)
logger = logging.getLogger(__name__)
_TAG = "dashboard-auth-self-hosted"
@@ -120,18 +99,15 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
validate_redirect_uri(redirect_uri)
disco = self._get_discovery()
return pkce_login_start(
disco["authorization_endpoint"], client_id=self._client_id, scope=self._scopes, redirect_uri=redirect_uri,
)
disco["authorization_endpoint"], client_id=self._client_id, scope=self._scopes, redirect_uri=redirect_uri)
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
# ``state`` is verified by the auth-route layer before this call.
return self._exchange(
{
"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
"client_id": self._client_id, "code_verifier": code_verifier,
},
bad_request_exc=InvalidCodeError,
)
"client_id": self._client_id, "code_verifier": code_verifier},
bad_request_exc=InvalidCodeError)
def refresh_session(self, *, refresh_token: str) -> Session:
if not refresh_token:
@@ -141,11 +117,9 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token,
# Re-request the same scopes so the rotated ID token keeps its identity
# claims (some IDPs narrow scope on refresh otherwise).
"scope": self._scopes,
},
"scope": self._scopes},
bad_request_exc=RefreshExpiredError,
previous_refresh_token=refresh_token,
)
previous_refresh_token=refresh_token)
def verify_session(self, *, access_token: str) -> Optional[Session]:
# The session cookie carries the ID token in the access-token slot (see _session)
@@ -207,9 +181,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
missing_msg=(
"OIDC token response missing id_token — ensure the 'openid' "
"scope is configured and the client is allowed to receive an "
"ID token."
),
)
"ID token."))
claims = self._verify_id_token(id_token)
# Prefer a freshly-issued RT, else keep the previous (some IDPs don't rotate).
return self._session(id_token, refresh_token_from(payload, previous_refresh_token), claims)
@@ -261,7 +233,8 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
advertised_issuer = field("issuer")
if advertised_issuer and advertised_issuer.rstrip("/") != self._issuer:
raise ProviderError(
f"OIDC discovery issuer mismatch: document advertises {advertised_issuer!r} but configured issuer is {self._issuer!r}"
f"OIDC discovery issuer mismatch: document advertises {advertised_issuer!r} "
f"but configured issuer is {self._issuer!r}"
)
for key, url in endpoints.items():
_require_https_or_loopback(url, field=key)
@@ -272,9 +245,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
**endpoints,
"revocation_endpoint": field("revocation_endpoint"),
"token_endpoint_auth_methods_supported": (
[str(m) for m in auth_methods_raw] if isinstance(auth_methods_raw, list) else []
),
}
[str(m) for m in auth_methods_raw] if isinstance(auth_methods_raw, list) else [])}
# ---- internals: JWT verification + mapping ----------------------------
@@ -287,8 +258,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
issuer = self._get_discovery()["issuer"]
return verify_jwt(
id_token, self._get_jwks_client(), algorithms=list(_ALLOWED_ID_TOKEN_ALGS),
audience=self._client_id, issuer=issuer, label="ID token",
)
audience=self._client_id, issuer=issuer, label="ID token")
def _session(self, id_token: str, refresh_token: str, claims: Dict[str, Any]) -> Session:
"""Map verified OIDC claims onto a Session. The verified ID token is stored in
@@ -304,8 +274,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
return session_from_claims(
self.name, claims, access_token=id_token, refresh_token=refresh_token, label="ID token", email=email,
display_name=str(claims.get("name") or claims.get("preferred_username") or claims.get("nickname") or email or ""),
org_id=str(org_id or ""),
)
org_id=str(org_id or ""))
# ---- Plugin entry point ----
@@ -330,14 +299,12 @@ def _settings() -> dict:
"HERMES_DASHBOARD_OIDC_CLIENT_ID) or under "
"dashboard.oauth.self_hosted.{issuer,client_id} in config.yaml — or pass "
"--insecure to skip the OAuth gate entirely. (issuer set: %s; client_id set: %s)"
% (bool(issuer), bool(client_id))
)
% (bool(issuer), bool(client_id)))
return {
"issuer": issuer, "client_id": client_id,
"scopes": setting("HERMES_DASHBOARD_OIDC_SCOPES", "scopes") or _DEFAULT_SCOPES,
# Credential: canonical home is the env var / ~/.hermes/.env. Empty ⇒ public client.
"client_secret": setting("HERMES_DASHBOARD_OIDC_CLIENT_SECRET", "client_secret"),
}
"client_secret": setting("HERMES_DASHBOARD_OIDC_CLIENT_SECRET", "client_secret")}
def register(ctx) -> None:

369
plugins/hermes-achievements/dashboard/plugin_api.py
View File

@@ -1,6 +1,8 @@
"""Hermes Achievements dashboard plugin backend.
"""Hermes Achievements dashboard plugin backend, mounted at /api/plugins/hermes-achievements/.
Mounted at /api/plugins/hermes-achievements/ by Hermes dashboard.
Scans the session history into per-session stats (checkpointed by fingerprint so warm
scans are cheap), aggregates them, and evaluates the tiered / multi-condition catalog.
Cold scans run on a background thread; ``/achievements`` serves the last snapshot.
"""
from __future__ import annotations
@@ -12,22 +14,9 @@ import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
try:
from hermes_constants import get_hermes_home
except ImportError:
import os as _os
def get_hermes_home() -> Path: # type: ignore[misc]
val = (_os.environ.get("HERMES_HOME") or "").strip()
return Path(val) if val else Path.home() / ".hermes"
from fastapi import APIRouter
try:
from fastapi import APIRouter
except Exception: # Allows local unit tests without dashboard dependencies.
class APIRouter: # type: ignore
def get(self, *_args, **_kwargs):
return lambda fn: fn
def post(self, *_args, **_kwargs):
return lambda fn: fn
from hermes_constants import get_hermes_home
router = APIRouter()
@@ -46,18 +35,13 @@ FILE_RE = re.compile(r"(?:/home/|~/?|\./|/mnt/)[\w./-]+\.(?:py|js|ts|tsx|jsx|css
TIER_NAMES = ["Copper", "Silver", "Gold", "Diamond", "Olympian"]
def _ach(
id: str, name: str, description: str, category: str, icon: str, *,
metric: Optional[str] = None, tiers: Optional[List[int]] = None,
requires: Optional[List[tuple]] = None, secret: bool = False,
) -> Dict[str, Any]:
"""Build one catalog entry.
``kind`` is derived: ``requires`` -> multi_condition; a ``max_*`` metric is a
per-session best (best_session); any other metric accumulates over the whole
history (lifetime).
"""
requires: Optional[List[tuple]] = None, secret: bool = False) -> Dict[str, Any]:
"""Build one catalog entry. ``kind`` is derived: ``requires`` -> multi_condition; a
``max_*`` metric is a per-session best (best_session); anything else accumulates over
the whole history (lifetime)."""
kind = "multi_condition" if requires is not None else ("best_session" if metric.startswith("max_") else "lifetime")
item: Dict[str, Any] = {"id": id, "name": name, "description": description, "category": category, "kind": kind, "icon": icon}
if secret:
@@ -151,14 +135,16 @@ ACHIEVEMENTS: List[Dict[str, Any]] = [
]
# ---- Durable state files ----
SNAPSHOT_FILE = "scan_snapshot.json"
CHECKPOINT_FILE = "scan_checkpoint.json"
def _data_dir() -> Path:
"""Durable data root (``<hermes home>/plugin-data/hermes-achievements/``).
Was the install tree (``plugins/hermes-achievements/``) before the plugin-data
convention existed — state parked there died on ``hermes plugins remove``/
``update``. Legacy files migrate on first read (see ``_data_file``).
"""
"""Durable data root (``<hermes home>/plugin-data/hermes-achievements/``). State used to
live in the install tree and died on ``hermes plugins remove``/``update``; legacy files
migrate on first read (see ``_data_file``)."""
try:
from plugins.plugin_storage import plugin_data_dir
return plugin_data_dir("hermes-achievements")
@@ -181,10 +167,6 @@ def _data_file(name: str) -> Path:
return path
SNAPSHOT_FILE = "scan_snapshot.json"
CHECKPOINT_FILE = "scan_checkpoint.json"
def _read_json(name: str) -> Any:
"""Parsed data file, or ``None`` when missing/unreadable."""
path = _data_file(name)
@@ -254,15 +236,15 @@ def _scan_status_payload(now: Optional[int] = None) -> Dict[str, Any]:
"ttl_seconds": SNAPSHOT_TTL_SECONDS,
"snapshot_generated_at": generated_at or None,
"snapshot_age_seconds": (current - generated_at) if generated_at else None,
"snapshot_stale": _is_snapshot_stale(snap, current),
}
"snapshot_stale": _is_snapshot_stale(snap, current)}
# ---- Per-session analysis ----
def _tool_name_from_call(call: Any) -> Optional[str]:
if not isinstance(call, dict):
return None
fn = call.get("function") or {}
return call.get("name") or fn.get("name")
return call.get("name") or (call.get("function") or {}).get("name")
def _content(msg: Dict[str, Any]) -> str:
@@ -282,13 +264,17 @@ def _count_tool(tool_names: List[str], *needles: str) -> int:
return sum(1 for name in lowered if any(needle in name for needle in needles))
_PROVIDER_MARKERS = ["openai", "anthropic", "google", "gemini", "mistral", "meta", "qwen", "deepseek", "xai", "nous", "ollama", "groq", "openrouter", "codex"]
_LOCAL_MARKERS = ["ollama", "llama.cpp", "localhost", "127.0.0.1", "local/", "local:", "gguf", "vllm-local"]
def model_provider(model_name: str) -> Optional[str]:
name = (model_name or "").strip().lower()
if not name or name == "none":
return None
if "/" in name:
return name.split("/", 1)[0]
for provider in ["openai", "anthropic", "google", "gemini", "mistral", "meta", "qwen", "deepseek", "xai", "nous", "ollama", "groq", "openrouter", "codex"]:
for provider in _PROVIDER_MARKERS:
if provider in name:
return "google" if provider == "gemini" else provider
return name.split(":", 1)[0].split("-", 1)[0]
@@ -296,10 +282,7 @@ def model_provider(model_name: str) -> Optional[str]:
def is_local_model_name(model_name: str) -> bool:
name = (model_name or "").strip().lower()
if not name or name == "none":
return False
local_markers = ["ollama", "llama.cpp", "localhost", "127.0.0.1", "local/", "local:", "gguf", "vllm-local"]
return any(marker in name for marker in local_markers)
return bool(name) and name != "none" and any(marker in name for marker in _LOCAL_MARKERS)
def analyze_messages(session_id: str, title: str, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
@@ -308,15 +291,14 @@ def analyze_messages(session_id: str, title: str, messages: List[Dict[str, Any]]
files_touched: Set[str] = set()
full_text_parts: List[str] = []
error_count = 0
for msg in messages:
text = _content(msg)
full_text_parts.append(text)
if msg.get("tool_name"):
name = str(msg["tool_name"])
tool_names.add(name)
# Tool result rows name the tool that already appeared in the assistant tool_calls.
# Keep it for distinct-tool detection, but do not double-count it as a new call.
# Tool result rows name the tool that already appeared in the assistant tool_calls:
# keep it for distinct-tool detection but don't double-count it as a new call.
if msg.get("role") != "tool":
tool_sequence.append(name)
for call in msg.get("tool_calls") or []:
@@ -398,32 +380,43 @@ def analyze_messages(session_id: str, title: str, messages: List[Dict[str, Any]]
"screenshot_events": hits(r"screenshot|playwright|vision_analyze|browser_vision|\.png|image data"),
"release_events": hits(r"\bgit\s+tag|release|version bump|changelog|publish|pushed? tag"),
"cache_events": hits(r"cache hit|prompt caching|cache_read"),
"model_names": set(),
}
"model_names": set()}
# ---- Evaluation ----
def _result(*, unlocked: bool, discovered: bool, state: str, tier, progress: int, next_tier, next_threshold: int, progress_pct: int) -> Dict[str, Any]:
"""Uniform evaluation result (key order is part of the wire shape)."""
return {"unlocked": unlocked, "discovered": discovered, "state": state, "tier": tier, "progress": progress, "next_tier": next_tier, "next_threshold": next_threshold, "progress_pct": progress_pct}
def _state(definition: Dict[str, Any], unlocked: bool, any_progress: bool) -> tuple[str, bool]:
"""``(state, discovered)``: secret badges stay hidden until the first matching signal."""
secret = bool(definition.get("secret"))
state = "unlocked" if unlocked else ("secret" if secret and not any_progress else "discovered")
return state, any_progress or not secret
def evaluate_tiered(definition: Dict[str, Any], aggregate: Dict[str, Any]) -> Dict[str, Any]:
metric = definition["threshold_metric"]
progress = int(aggregate.get(metric, 0) or 0)
progress = int(aggregate.get(definition["threshold_metric"], 0) or 0)
tiers_list = sorted(definition.get("tiers", []), key=lambda t: t["threshold"])
achieved = [t for t in tiers_list if progress >= t["threshold"]]
next_tiers = [t for t in tiers_list if progress < t["threshold"]]
tier = achieved[-1]["name"] if achieved else None
next_tier = next_tiers[0]["name"] if next_tiers else None
next_threshold = next_tiers[0]["threshold"] if next_tiers else (tiers_list[-1]["threshold"] if tiers_list else 1)
current_threshold = achieved[-1]["threshold"] if achieved else 0
denom = max(1, next_threshold - current_threshold)
pct = 100 if not next_tiers and achieved else max(0, min(99, math.floor(((progress - current_threshold) / denom) * 100)))
unlocked = bool(achieved)
discovered = bool(progress > 0)
state = "unlocked" if unlocked else ("secret" if definition.get("secret") and not discovered else "discovered")
return {"unlocked": unlocked, "discovered": discovered or not definition.get("secret"), "state": state, "tier": tier, "progress": progress, "next_tier": next_tier, "next_threshold": next_threshold, "progress_pct": pct}
state, discovered = _state(definition, bool(achieved), progress > 0)
return _result(
unlocked=bool(achieved), discovered=discovered, state=state, tier=achieved[-1]["name"] if achieved else None,
progress=progress, next_tier=next_tiers[0]["name"] if next_tiers else None, next_threshold=next_threshold, progress_pct=pct)
def evaluate_requirements(definition: Dict[str, Any], aggregate: Dict[str, Any]) -> Dict[str, Any]:
requirements = definition.get("requirements", [])
if not requirements:
return {"unlocked": False, "discovered": not definition.get("secret"), "state": "secret" if definition.get("secret") else "discovered", "tier": None, "progress": 0, "next_tier": None, "next_threshold": 1, "progress_pct": 0}
state, discovered = _state(definition, False, False)
return _result(unlocked=False, discovered=discovered, state=state, tier=None, progress=0, next_tier=None, next_threshold=1, progress_pct=0)
parts = []
any_progress = False
complete = True
@@ -434,14 +427,16 @@ def evaluate_requirements(definition: Dict[str, Any], aggregate: Dict[str, Any])
complete = complete and value >= threshold
parts.append(min(1.0, value / max(1, threshold)))
pct = math.floor((sum(parts) / len(parts)) * 100)
state = "unlocked" if complete else ("secret" if definition.get("secret") and not any_progress else "discovered")
return {"unlocked": complete, "discovered": any_progress or not definition.get("secret"), "state": state, "tier": None, "progress": pct, "next_tier": None, "next_threshold": 100, "progress_pct": 100 if complete else min(99, pct)}
state, discovered = _state(definition, complete, any_progress)
return _result(
unlocked=complete, discovered=discovered, state=state, tier=None, progress=pct, next_tier=None,
next_threshold=100, progress_pct=100 if complete else min(99, pct))
def evaluate_boolean(definition: Dict[str, Any], aggregate: Dict[str, Any]) -> Dict[str, Any]:
# Backward-compatible helper for old tests/definitions. New catalog avoids simple booleans.
unlocked = bool(aggregate.get(definition["metric"]))
return {"unlocked": unlocked, "discovered": True, "state": "unlocked" if unlocked else "discovered", "tier": None, "progress": 1 if unlocked else 0, "next_tier": None, "next_threshold": 1, "progress_pct": 100 if unlocked else 0}
def evaluate_definition(definition: Dict[str, Any], aggregate: Dict[str, Any]) -> Dict[str, Any]:
if "threshold_metric" in definition:
return evaluate_tiered(definition, aggregate)
return evaluate_requirements(definition, aggregate)
METRIC_LABELS = {
@@ -506,8 +501,7 @@ METRIC_LABELS = {
"release_events": "release, version, publish, or git tag events",
"session_count": "Hermes sessions",
"weekend_sessions": "sessions started on weekends",
"night_sessions": "sessions started late night or before dawn",
}
"night_sessions": "sessions started late night or before dawn"}
def metric_label(metric: str) -> str:
@@ -521,13 +515,11 @@ def criteria_for(definition: Dict[str, Any]) -> str:
tiers_list = sorted(definition.get("tiers", []), key=lambda t: t["threshold"])
if not tiers_list:
return "Requirement: use Hermes in the matching workflow."
metric = metric_label(definition["threshold_metric"])
ladder = ", ".join(f"{t['name']} {t['threshold']}" for t in tiers_list)
return f"Requirement: {metric}. Tier ladder: {ladder}."
return f"Requirement: {metric_label(definition['threshold_metric'])}. Tier ladder: {ladder}."
requirements = definition.get("requirements") or []
if requirements:
parts = [f"{metric_label(r['metric'])} ≥ {int(r.get('gte', 1))}" for r in requirements]
return "Requirement: " + "; ".join(parts) + "."
return "Requirement: " + "; ".join(f"{metric_label(r['metric'])} ≥ {int(r.get('gte', 1))}" for r in requirements) + "."
return "Requirement: complete the matching Hermes behavior."
@@ -539,36 +531,34 @@ def display_achievement(item: Dict[str, Any]) -> Dict[str, Any]:
return clean
def scan_sessions(
limit: Optional[int] = None,
progress_callback: Optional[Any] = None,
progress_every: int = 250,
) -> Dict[str, Any]:
# ---- Scanning + aggregation ----
def _scan_meta(mode: str, total: int, *, rescanned: int = 0, reused: int = 0, scanned_so_far: Optional[int] = None, expected_total: Optional[int] = None) -> Dict[str, Any]:
meta = {"mode": mode, "sessions_total": total, "sessions_rescanned": rescanned, "sessions_reused": reused}
if scanned_so_far is not None:
meta.update(sessions_scanned_so_far=scanned_so_far, sessions_expected_total=expected_total)
return meta
def scan_sessions(limit: Optional[int] = None, progress_callback: Optional[Any] = None, progress_every: int = 250) -> Dict[str, Any]:
"""Scan Hermes sessions and build per-session achievement stats.
``limit=None`` (the default) scans the ENTIRE history: SQLite's ``LIMIT -1``
means unlimited, so ``None``/non-positive map to ``-1``. A former cap of 200
silently shrank lifetime totals to ~2% on long-running installs.
Warm scans stay cheap: the checkpoint stores per-session stats keyed by
``(started_at, last_active)`` fingerprint and only re-analyzes changed
sessions. Cold scans over thousands of sessions take tens of seconds to
minutes, so ``evaluate_all`` runs them on a background thread.
``progress_callback(partial_sessions, scanned_so_far, total)`` fires every
``progress_every`` sessions so background scans can publish intermediate
snapshots and surface badges incrementally instead of all at the end.
``limit=None`` (default) scans the ENTIRE history (SQLite ``LIMIT -1``); a former cap
of 200 silently shrank lifetime totals on long-running installs. The checkpoint stores
per-session stats keyed by ``(started_at, last_active)`` fingerprint so warm scans only
re-analyze changed sessions. ``progress_callback(partial_sessions, scanned_so_far,
total)`` fires every ``progress_every`` sessions so background scans can publish
intermediate snapshots.
"""
try:
from hermes_state import SessionDB
except Exception as exc:
return {"sessions": [], "aggregate": {}, "error": f"Could not import SessionDB: {exc}", "scan_meta": {"mode": "failed", "sessions_total": 0, "sessions_rescanned": 0, "sessions_reused": 0}}
return {"sessions": [], "aggregate": {}, "error": f"Could not import SessionDB: {exc}", "scan_meta": _scan_meta("failed", 0)}
checkpoint = load_checkpoint()
previous_sessions = checkpoint.get("sessions") if isinstance(checkpoint.get("sessions"), dict) else {}
reused = rescanned = 0
db_limit = -1 if (limit is None or limit <= 0) else int(limit)
db = SessionDB()
try:
sessions_meta = db.list_sessions_rich(limit=db_limit, include_children=True, project_compression_tips=False)
@@ -582,45 +572,41 @@ def scan_sessions(
fp = session_fingerprint(meta)
cached = previous_sessions.get(sid)
cached = cached if isinstance(cached, dict) else {}
title = meta.get("title") or meta.get("preview")
if isinstance(cached.get("stats"), dict) and cached.get("fingerprint") == fp:
stats = dict(cached["stats"])
reused += 1
else:
messages = db.get_messages(sid)
stats = analyze_messages(sid, meta.get("title") or meta.get("preview") or "Untitled", messages)
stats = analyze_messages(sid, title or "Untitled", db.get_messages(sid))
rescanned += 1
stats.update(session_id=sid, title=meta.get("title") or meta.get("preview") or stats.get("title") or "Untitled", started_at=meta.get("started_at"), last_active=meta.get("last_active"), source=meta.get("source"))
stats.update(session_id=sid, title=title or stats.get("title") or "Untitled", started_at=meta.get("started_at"), last_active=meta.get("last_active"), source=meta.get("source"))
if meta.get("model"):
# Checkpoint round-trips turn the set into a list; handle both.
model = str(meta.get("model"))
names = stats.setdefault("model_names", set())
if isinstance(names, set):
names.add(model)
elif isinstance(names, list) and model not in names:
names.append(model)
elif not isinstance(names, list):
elif isinstance(names, list):
if model not in names:
names.append(model)
else:
stats["model_names"] = {model}
sessions.append(stats)
checkpoint_sessions[sid] = {"fingerprint": fp, "stats": _json_safe(stats)}
if progress_callback is not None and progress_every > 0 and (idx % progress_every == 0) and idx < total_sessions:
try:
progress_callback(list(sessions), idx, total_sessions)
except Exception:
pass # Advisory — a broken publisher must never abort the scan.
_write_json(CHECKPOINT_FILE, {"schema_version": 1, "generated_at": int(time.time()), "sessions": checkpoint_sessions})
finally:
close = getattr(db, "close", None)
if close:
close()
db.close()
return {
"sessions": sessions,
"aggregate": aggregate_stats(sessions),
"scan_meta": {"mode": "incremental" if reused > 0 else "full", "sessions_total": len(sessions), "sessions_rescanned": rescanned, "sessions_reused": reused, "sessions_scanned_so_far": len(sessions), "sessions_expected_total": total_sessions},
}
"scan_meta": _scan_meta(
"incremental" if reused > 0 else "full", len(sessions), rescanned=rescanned, reused=reused,
scanned_so_far=len(sessions), expected_total=total_sessions)}
# Per-session bests: aggregate metric -> session stat key (also drives evidence_for).
@@ -632,8 +618,7 @@ _SESSION_MAX_METRICS = {
"max_file_tool_calls_in_session": "file_tool_calls",
"max_web_calls_in_session": "web_calls",
"max_web_browser_calls_in_session": "web_browser_calls",
"max_files_touched_in_session": "files_touched_count",
}
"max_files_touched_in_session": "files_touched_count"}
# Lifetime sums: aggregate metric -> session stat key.
_SESSION_SUM_METRICS = {
"total_errors": "error_count",
@@ -648,8 +633,7 @@ _SESSION_SUM_METRICS = {
"total_cron_calls": "cron_calls",
"browser_calls": "browser_calls",
"image_vision_calls": "image_vision_calls",
"tts_calls": "tts_calls",
}
"tts_calls": "tts_calls"}
# ``*_events`` counters summed under their own name.
_SESSION_EVENT_KEYS = [
"traceback_events", "log_read_events", "port_conflict_events", "permission_denied_events", "install_error_events", "install_success_events", "restart_after_error_events", "env_var_error_events", "yaml_error_events", "docker_conflict_events", "frontend_activity_events", "css_activity_events", "git_events", "tiny_patch_after_errors_events", "skill_events", "skill_manage_events", "memory_events", "memory_write_events", "context_events", "gateway_events", "plugin_events", "rollback_events", "docs_activity_events", "model_events", "openrouter_events", "codex_events", "claude_events", "gemini_events", "local_model_events", "toolset_events", "config_events", "git_history_events", "test_events", "screenshot_events", "release_events", "cache_events",
@@ -661,7 +645,6 @@ def aggregate_stats(sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
agg: Dict[str, Any] = {"session_count": len(sessions)}
for key in (*_SESSION_MAX_METRICS, *_SESSION_SUM_METRICS, "distinct_model_count", "distinct_provider_count", "local_model_chat_sessions", "weekend_sessions", "night_sessions", *_SESSION_EVENT_KEYS):
agg[key] = 0
model_names: Set[str] = set()
provider_names: Set[str] = set()
for s in sessions:
@@ -673,11 +656,8 @@ def aggregate_stats(sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
agg[key] += s.get(key, 0)
session_models = s.get("model_names") or set()
model_names.update(session_models)
for model_name in session_models:
provider = model_provider(str(model_name))
if provider:
provider_names.add(provider)
if any(is_local_model_name(str(model_name)) for model_name in session_models):
provider_names.update(filter(None, (model_provider(str(m)) for m in session_models)))
if any(is_local_model_name(str(m)) for m in session_models):
agg["local_model_chat_sessions"] += 1
if s.get("started_at"):
try:
@@ -693,14 +673,6 @@ def aggregate_stats(sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
return agg
def evaluate_definition(definition: Dict[str, Any], aggregate: Dict[str, Any]) -> Dict[str, Any]:
if "threshold_metric" in definition:
return evaluate_tiered(definition, aggregate)
if "requirements" in definition:
return evaluate_requirements(definition, aggregate)
return evaluate_boolean(definition, aggregate)
def evidence_for(definition: Dict[str, Any], sessions: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
key = _SESSION_MAX_METRICS.get(definition.get("threshold_metric"))
if not sessions or key is None:
@@ -709,14 +681,28 @@ def evidence_for(definition: Dict[str, Any], sessions: List[Dict[str, Any]]) ->
return {"session_id": s.get("session_id"), "title": s.get("title"), "value": s.get(key, 0)}
def _compute_from_scan(scan: Dict[str, Any], *, is_partial: bool = False) -> Dict[str, Any]:
"""Evaluate every achievement definition against a scan result.
# ---- Snapshot assembly ----
Used by ``compute_all`` for finished scans AND by the background progress
callback for in-flight snapshots. ``is_partial=True`` skips persisting
``state.json`` unlocks — an "unlock time" from half a scan could be
invalidated by a later session.
"""
def _snapshot(evaluated: List[Dict[str, Any]], scan: Dict[str, Any], now: int) -> Dict[str, Any]:
"""Wire payload shared by finished, partial and pending snapshots."""
return {
"achievements": evaluated,
"sessions": scan.get("sessions", []),
"aggregate": scan.get("aggregate", {}),
"scan_meta": scan.get("scan_meta", {}),
"error": scan.get("error"),
"unlocked_count": sum(1 for a in evaluated if a["unlocked"]),
"discovered_count": sum(1 for a in evaluated if a.get("state") == "discovered"),
"secret_count": sum(1 for a in evaluated if a.get("state") == "secret"),
"total_count": len(evaluated),
"generated_at": now}
def _compute_from_scan(scan: Dict[str, Any], *, is_partial: bool = False) -> Dict[str, Any]:
"""Evaluate every achievement definition against a scan result. Used by ``compute_all``
for finished scans AND by the background progress callback for in-flight snapshots;
``is_partial=True`` skips persisting ``state.json`` unlocks — an "unlock time" from
half a scan could be invalidated by a later session."""
aggregate = scan.get("aggregate", {})
state = load_state() if not is_partial else {"unlocks": {}}
unlocks = state.setdefault("unlocks", {})
@@ -734,18 +720,7 @@ def _compute_from_scan(scan: Dict[str, Any], *, is_partial: bool = False) -> Dic
evaluated.append(display_achievement(item))
if not is_partial:
save_state(state)
return {
"achievements": evaluated,
"sessions": scan.get("sessions", []),
"aggregate": aggregate,
"scan_meta": scan.get("scan_meta", {}),
"error": scan.get("error"),
"unlocked_count": sum(1 for a in evaluated if a["unlocked"]),
"discovered_count": sum(1 for a in evaluated if a.get("state") == "discovered"),
"secret_count": sum(1 for a in evaluated if a.get("state") == "secret"),
"total_count": len(evaluated),
"generated_at": now,
}
return _snapshot(evaluated, scan, now)
def compute_all(progress_callback: Optional[Any] = None, progress_every: int = 250) -> Dict[str, Any]:
@@ -758,77 +733,59 @@ _BACKGROUND_SCAN_LOCK = threading.Lock()
def _build_pending_snapshot(now: int) -> Dict[str, Any]:
"""Structurally-complete placeholder served while the first-ever scan runs,
so the UI renders an empty list + spinner without special-casing "no data"."""
evaluated = [display_achievement({**d, **{"unlocked": False, "discovered": False, "state": "secret" if d.get("secret") else "discovered", "progress": 0, "progress_pct": 0, "next_tier": (d.get("tiers") or [{}])[0].get("name"), "next_threshold": (d.get("tiers") or [{}])[0].get("threshold", 1), "tier": None}}) for d in ACHIEVEMENTS]
return {
"achievements": evaluated,
"sessions": [],
"aggregate": {},
"scan_meta": {"mode": "pending", "sessions_total": 0, "sessions_rescanned": 0, "sessions_reused": 0},
"error": None,
"unlocked_count": 0,
"discovered_count": sum(1 for a in evaluated if a.get("state") == "discovered"),
"secret_count": sum(1 for a in evaluated if a.get("state") == "secret"),
"total_count": len(evaluated),
"generated_at": now,
}
"""Structurally-complete placeholder served while the first-ever scan runs, so the UI
renders an empty list + spinner without special-casing "no data"."""
evaluated = [
display_achievement({
**d, "unlocked": False, "discovered": False, "state": "secret" if d.get("secret") else "discovered", "progress": 0,
"progress_pct": 0, "next_tier": (d.get("tiers") or [{}])[0].get("name"),
"next_threshold": (d.get("tiers") or [{}])[0].get("threshold", 1), "tier": None})
for d in ACHIEVEMENTS]
return _snapshot(evaluated, {"scan_meta": _scan_meta("pending", 0), "error": None}, now)
def _set_cache(snapshot: Dict[str, Any], at: int) -> None:
global _SNAPSHOT_CACHE, _SNAPSHOT_CACHE_AT
_SNAPSHOT_CACHE = _json_safe(snapshot)
_SNAPSHOT_CACHE_AT = at
def _run_scan_and_update_cache(publish_partial_snapshots: bool = True) -> None:
"""Execute a scan + snapshot update. Called synchronously or from a thread.
With ``publish_partial_snapshots`` (background scans) the scanner periodically
publishes in-progress snapshots to ``_SNAPSHOT_CACHE`` so a long cold scan
unlocks badges incrementally instead of jumping from zero to final.
Synchronous /rescan callers pass ``False`` since they block on the result.
"""
global _SNAPSHOT_CACHE, _SNAPSHOT_CACHE_AT
"""Execute a scan + snapshot update (synchronously or from a thread). With
``publish_partial_snapshots`` (background scans) the scanner periodically publishes
in-progress snapshots to ``_SNAPSHOT_CACHE`` so a long cold scan unlocks badges
incrementally; synchronous /rescan callers pass ``False`` since they block on the result."""
with _SCAN_LOCK:
started = int(time.time())
_SCAN_STATUS["state"] = "running"
_SCAN_STATUS["started_at"] = started
_SCAN_STATUS["last_error"] = None
_SCAN_STATUS.update(state="running", started_at=started, last_error=None)
def _publish_partial(partial_sessions, scanned_so_far, total):
global _SNAPSHOT_CACHE, _SNAPSHOT_CACHE_AT
try:
partial_scan = {
"sessions": partial_sessions,
"aggregate": aggregate_stats(partial_sessions),
"scan_meta": {"mode": "in_progress", "sessions_total": scanned_so_far, "sessions_rescanned": 0, "sessions_reused": 0, "sessions_scanned_so_far": scanned_so_far, "sessions_expected_total": total},
"scan_meta": _scan_meta("in_progress", scanned_so_far, scanned_so_far=scanned_so_far, expected_total=total),
}
partial = _compute_from_scan(partial_scan, is_partial=True)
# _SNAPSHOT_CACHE_AT stays 0 so partials remain in the 'stale'
# regime: the UI keeps polling /scan-status and never mistakes
# an in-flight result for a finished one.
_SNAPSHOT_CACHE = _json_safe(partial)
_SNAPSHOT_CACHE_AT = 0
# _SNAPSHOT_CACHE_AT stays 0 so partials remain in the 'stale' regime: the UI
# keeps polling /scan-status and never mistakes an in-flight result for a finished one.
_set_cache(_compute_from_scan(partial_scan, is_partial=True), 0)
except Exception:
pass # Intermediate publication is best-effort; don't kill the scan.
callback = _publish_partial if publish_partial_snapshots else None
try:
computed = compute_all(progress_callback=callback)
_SNAPSHOT_CACHE = _json_safe(computed)
_SNAPSHOT_CACHE_AT = int(_SNAPSHOT_CACHE.get("generated_at") or int(time.time()))
computed = _json_safe(compute_all(progress_callback=_publish_partial if publish_partial_snapshots else None))
_set_cache(computed, int(computed.get("generated_at") or int(time.time())))
_write_json(SNAPSHOT_FILE, _SNAPSHOT_CACHE)
_SCAN_STATUS["state"] = "idle"
except Exception as exc:
_SCAN_STATUS["state"] = "failed"
_SCAN_STATUS["last_error"] = str(exc)
_SCAN_STATUS.update(state="failed", last_error=str(exc))
finally:
_SCAN_STATUS["finished_at"] = int(time.time())
_SCAN_STATUS["last_duration_ms"] = int((_SCAN_STATUS["finished_at"] - started) * 1000)
_SCAN_STATUS["run_count"] = int(_SCAN_STATUS.get("run_count", 0)) + 1
finished = int(time.time())
_SCAN_STATUS.update(finished_at=finished, last_duration_ms=int((finished - started) * 1000), run_count=int(_SCAN_STATUS.get("run_count", 0)) + 1)
def _start_background_scan() -> None:
"""Kick off a daemon-thread scan unless one is already running (idempotent).
The thread updates ``_SNAPSHOT_CACHE`` on completion and publishes partial
snapshots every ~250 sessions while running.
"""
"""Kick off a daemon-thread scan unless one is already running (idempotent)."""
global _BACKGROUND_SCAN_THREAD
with _BACKGROUND_SCAN_LOCK:
existing = _BACKGROUND_SCAN_THREAD
@@ -840,45 +797,33 @@ def _start_background_scan() -> None:
def evaluate_all(force: bool = False) -> Dict[str, Any]:
"""Return the current achievements payload.
* Fresh in-memory cache -> return it instantly.
* Stale on-disk snapshot -> load it, kick a background rescan, return the
stale data (UI decorates it with ``is_stale=True``).
* No snapshot yet (first-ever run) -> kick a background scan, return an
empty-but-valid "pending" payload so the UI can render a spinner.
* ``force=True`` (manual /rescan) -> run synchronously, replace the cache.
Cold scans on 8000+ session databases take minutes; the background thread
keeps them off the dashboard request path.
"""
"""Return the current achievements payload: a fresh in-memory cache is returned as is;
a stale on-disk snapshot is served while a background rescan runs (UI decorates it with
``is_stale=True``); with no snapshot yet an empty-but-valid "pending" payload is served
while the first scan runs; ``force=True`` (manual /rescan) scans synchronously. Cold
scans on 8000+ session databases take minutes, hence the background thread."""
global _SNAPSHOT_CACHE, _SNAPSHOT_CACHE_AT
now = int(time.time())
if not force and _cache_is_fresh(now):
return _SNAPSHOT_CACHE or {}
# Lazy-load the persisted snapshot so fresh process starts serve cached data.
if _SNAPSHOT_CACHE is None:
persisted = _read_json(SNAPSHOT_FILE)
if isinstance(persisted, dict):
_SNAPSHOT_CACHE = persisted
_SNAPSHOT_CACHE_AT = int(persisted.get("generated_at") or 0) or now
if force:
# No partial publishing: the caller is blocking on the final result.
_run_scan_and_update_cache(publish_partial_snapshots=False)
if _SNAPSHOT_CACHE is not None:
return _SNAPSHOT_CACHE
return _build_pending_snapshot(now) # Scan failed with no prior cache.
# Serve what we have (stale is fine) and refresh in the background; on a
# first-ever run the UI polls /scan-status and re-fetches when the scan completes.
if not _cache_is_fresh(now):
elif not _cache_is_fresh(now):
# Serve what we have (stale is fine) and refresh in the background; on a first-ever
# run the UI polls /scan-status and re-fetches when the scan completes.
_start_background_scan()
return _SNAPSHOT_CACHE if _SNAPSHOT_CACHE is not None else _build_pending_snapshot(now)
# ---- Routes ----
@router.get("/achievements")
async def achievements():
data = evaluate_all()

6
plugins/kanban/dashboard/plugin_api.py
View File

@@ -23,7 +23,8 @@ from dataclasses import asdict
from pathlib import Path
from typing import Any, Callable, Iterator, Optional
from fastapi import APIRouter, File, Form, HTTPException, Query, UploadFile, WebSocket, WebSocketDisconnect, status as http_status
from fastapi import (
APIRouter, File, Form, HTTPException, Query, UploadFile, WebSocket, WebSocketDisconnect, status as http_status)
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
@@ -1264,7 +1265,8 @@ def model_options():
load_picker_context(), explicit_only=True, canonical_order=True, probe_custom_providers=False)
return {
"providers": [
{"slug": row.get("slug", ""), "label": row.get("label") or row.get("slug", ""), "models": list(row.get("models") or [])}
{"slug": row.get("slug", ""), "label": row.get("label") or row.get("slug", ""),
"models": list(row.get("models") or [])}
for row in payload.get("providers", [])
if row.get("models")]}
except Exception: