From d20d37e5e0208bdc055b1704f832008c2f3f148f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:45:15 -0700 Subject: [PATCH 1/3] refactor(plugins/dashboard_auth): JwtOAuthProvider base (complete/refresh/verify + JWKS client) and NonInteractiveMixin in _shared; nous/self_hosted/basic/drain become thin subclasses --- plugins/dashboard_auth/_shared.py | 76 ++++++++++++++++++- plugins/dashboard_auth/basic/__init__.py | 22 ++---- plugins/dashboard_auth/drain/__init__.py | 16 ++-- plugins/dashboard_auth/nous/__init__.py | 65 +++++----------- .../dashboard_auth/self_hosted/__init__.py | 70 ++++++----------- 5 files changed, 133 insertions(+), 116 deletions(-) diff --git a/plugins/dashboard_auth/_shared.py b/plugins/dashboard_auth/_shared.py index 56f41cb0f0..877224bf93 100644 --- a/plugins/dashboard_auth/_shared.py +++ b/plugins/dashboard_auth/_shared.py @@ -17,7 +17,9 @@ from typing import Any, Callable, Dict, Optional import httpx -from hermes_cli.dashboard_auth import InvalidCodeError, LoginStart, ProviderError, Session, classify_jwks_lookup_error +from hermes_cli.dashboard_auth import ( + DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session, + classify_jwks_lookup_error) # JWKS Cache-Control max-age (nous contract C7); self-hosted mirrors it. JWKS_CACHE_SECONDS = 300 @@ -221,3 +223,75 @@ def verify_jwt( except Exception: pass raise ProviderError(f"{label} verification failed: {exc}{details}") from exc + + +# ---- Shared provider skeletons ---- + +class NonInteractiveMixin: + """OAuth-redirect stubs for providers without a browser login flow (password / service + credential). ``_NOT_INTERACTIVE`` is the operator-facing reason; ``_NO_START_LOGIN`` + optionally overrides the ``start_login`` message.""" + + _NOT_INTERACTIVE: str = "" + _NO_START_LOGIN: str = "" + + def start_login(self, *, redirect_uri: str) -> LoginStart: + raise NotImplementedError(self._NO_START_LOGIN or self._NOT_INTERACTIVE) + + def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session: + raise NotImplementedError(self._NOT_INTERACTIVE) + + +class JwtOAuthProvider(DashboardAuthProvider): + """Authorization-code + PKCE provider whose session token is a JWT we verify ourselves + (nous: Portal access token; self-hosted: OIDC ID token). Subclasses implement + ``_jwks_uri``, ``_claims_for``, ``_grant``, ``_refresh_request`` and ``_session``.""" + + _jwks_client: Any = None + + def _jwks_uri(self) -> str: + raise NotImplementedError + + def _claims_for(self, token: str) -> Dict[str, Any]: + raise NotImplementedError + + def _grant( + self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None, + previous_refresh_token: str = "", + ) -> Session: + raise NotImplementedError + + def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]: + """``(form_data, extra_headers)`` for the refresh grant.""" + raise NotImplementedError + + def _session(self, token: str, refresh_token: str, claims: Dict[str, Any]) -> Session: + raise NotImplementedError + + def _get_jwks_client(self) -> Any: + if self._jwks_client is None: + self._jwks_client = make_jwks_client(self._jwks_uri()) + return self._jwks_client + + 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._grant( + {"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, + "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: + raise RefreshExpiredError("no refresh token present in session") + data, headers = self._refresh_request(refresh_token) + return self._grant( + data, headers=headers, bad_request_exc=RefreshExpiredError, previous_refresh_token=refresh_token) + + def verify_session(self, *, access_token: str) -> Optional[Session]: + # None on expiry/invalidity (middleware then tries refresh); a ProviderError + # (JWKS unreachable) bubbles up so middleware emits 503. + try: + claims = self._claims_for(access_token) + except InvalidCodeError: + return None + return self._session(access_token, "", claims) diff --git a/plugins/dashboard_auth/basic/__init__.py b/plugins/dashboard_auth/basic/__init__.py index a09145322d..6abf5e7f18 100644 --- a/plugins/dashboard_auth/basic/__init__.py +++ b/plugins/dashboard_auth/basic/__init__.py @@ -20,9 +20,9 @@ import secrets import time from typing import Optional -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 +from hermes_cli.dashboard_auth import DashboardAuthProvider, InvalidCredentialsError, RefreshExpiredError, Session +from plugins.dashboard_auth._shared import ( + NonInteractiveMixin, SkipRegistration, load_config_section, register_provider, resolve_env_or_cfg) logger = logging.getLogger(__name__) _TAG = "dashboard-auth-basic" @@ -111,12 +111,16 @@ def _unsign(token: str, secret: bytes, kind: str) -> Optional[dict]: # ---- Provider ---- -class BasicAuthProvider(DashboardAuthProvider): +class BasicAuthProvider(NonInteractiveMixin, DashboardAuthProvider): """Username/password provider with stateless HMAC-signed sessions.""" name = "basic" display_name = "Username & Password" supports_password = True + _NOT_INTERACTIVE = "BasicAuthProvider is password-only; use complete_password_login." + _NO_START_LOGIN = ( + "BasicAuthProvider is password-only; there is no OAuth redirect flow. " + "The login page POSTs to /auth/password-login instead.") def __init__(self, *, username: str, password_hash: str, secret: bytes, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> None: if not username: @@ -130,16 +134,6 @@ class BasicAuthProvider(DashboardAuthProvider): self._secret = secret self._ttl = max(60, int(ttl_seconds)) - # ---- OAuth methods: not used (pure-password provider) ------------------ - - 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.") - - 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.") - # ---- password login ---------------------------------------------------- def complete_password_login(self, *, username: str, password: str) -> Session: diff --git a/plugins/dashboard_auth/drain/__init__.py b/plugins/dashboard_auth/drain/__init__.py index 3738e30fe1..de8447855a 100644 --- a/plugins/dashboard_auth/drain/__init__.py +++ b/plugins/dashboard_auth/drain/__init__.py @@ -16,8 +16,8 @@ import os from collections import Counter from typing import Optional -from hermes_cli.dashboard_auth import DashboardAuthProvider, LoginStart, Session, TokenPrincipal -from plugins.dashboard_auth._shared import SkipRegistration, load_config_section, register_provider +from hermes_cli.dashboard_auth import DashboardAuthProvider, Session, TokenPrincipal +from plugins.dashboard_auth._shared import NonInteractiveMixin, SkipRegistration, load_config_section, register_provider logger = logging.getLogger(__name__) _TAG = "dashboard-auth-drain" @@ -65,13 +65,15 @@ def assess_secret_strength(secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_ return None -class DrainSecretProvider(DashboardAuthProvider): +class DrainSecretProvider(NonInteractiveMixin, DashboardAuthProvider): """Non-interactive shared-bearer-secret provider for drain control.""" name = "drain-secret" display_name = "Drain Control (service credential)" supports_token = True supports_session = False + _NOT_INTERACTIVE = "DrainSecretProvider is a non-interactive service credential." + _NO_START_LOGIN = "DrainSecretProvider is a non-interactive service credential; there is no login flow." def __init__(self, *, secret: str, scope: str = "drain") -> None: # Defence in depth: construction enforces the entropy bar too, so a @@ -93,19 +95,13 @@ class DrainSecretProvider(DashboardAuthProvider): # ---- interactive methods: unsupported (service credential only) -------- - def start_login(self, *, redirect_uri: str) -> LoginStart: - raise NotImplementedError("DrainSecretProvider is a non-interactive service credential; there is no login flow.") - - def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session: - raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.") - def verify_session(self, *, access_token: str) -> Optional[Session]: # Never mints a Session, so never recognises a cookie. Return None (don't raise) # so it stacks harmlessly in the cookie-verify loop. return None def refresh_session(self, *, refresh_token: str) -> Session: - raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.") + raise NotImplementedError(self._NOT_INTERACTIVE) def revoke_session(self, *, refresh_token: str) -> None: return None diff --git a/plugins/dashboard_auth/nous/__init__.py b/plugins/dashboard_auth/nous/__init__.py index 43b69f6edd..3f56be7794 100644 --- a/plugins/dashboard_auth/nous/__init__.py +++ b/plugins/dashboard_auth/nous/__init__.py @@ -13,13 +13,12 @@ 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 LoginStart, ProviderError, Session from plugins.dashboard_auth._shared import ( + JwtOAuthProvider, SkipRegistration, exchange_token, load_config_section, - make_jwks_client, pkce_login_start, refresh_token_from, register_provider, @@ -38,7 +37,7 @@ _EXPECTED_CONTRACT_VERSION = 1 # contract C11 LAST_SKIP_REASON: str = "" # cleared on every register() so restarts don't leak stale reasons -class NousDashboardAuthProvider(DashboardAuthProvider): +class NousDashboardAuthProvider(JwtOAuthProvider): """Nous Portal OAuth via authorization-code + PKCE (S256).""" name = "nous" @@ -56,63 +55,41 @@ class NousDashboardAuthProvider(DashboardAuthProvider): self._token_url = f"{self._portal_url}/api/oauth/token" self._jwks_client: Any = None # lazily built (crypto import cost) - # ---- public API (DashboardAuthProvider) ------------------------------- - def start_login(self, *, redirect_uri: str) -> LoginStart: validate_redirect_uri(redirect_uri) return pkce_login_start(self._authorize_url, client_id=self._client_id, scope=_SCOPE, 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; Portal doesn't re-check it at the token endpoint. - 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) - - def refresh_session(self, *, refresh_token: str) -> Session: - if not refresh_token: - raise RefreshExpiredError("no refresh token present in session") - # The RT goes in BOTH the body (Portal's request schema requires it) and the - # ``x-nous-refresh-token`` header (Portal reconciles the two and keeps the value - # out of body access logs). Header-only → 400. - 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) - - def verify_session(self, *, access_token: str) -> Optional[Session]: - # None on expiry/invalidity (middleware then tries refresh); a ProviderError - # (JWKS unreachable) bubbles up so middleware emits 503. - try: - claims = self._verify_jwt(access_token) - except InvalidCodeError: - return None - return self._session(access_token, "", claims) - def revoke_session(self, *, refresh_token: str) -> None: # Portal exposes no token-endpoint revocation grant; logout is client-side cookie # clearing and the RT expires within its 24h TTL. return None - # ---- internals -------------------------------------------------------- + # ---- JwtOAuthProvider hooks ------------------------------------------- - def _token_grant( + def _jwks_uri(self) -> str: + return self._jwks_url + + def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]: + # The RT goes in BOTH the body (Portal's request schema requires it) and the + # ``x-nous-refresh-token`` header (Portal reconciles the two and keeps the value + # out of body access logs). Header-only → 400. + return ( + {"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token}, + {"x-nous-refresh-token": refresh_token}) + + def _grant( self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None, + previous_refresh_token: str = "", ) -> Session: 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") # Rotating RT the caller MUST persist back to the cookie. - return self._session(access_token, refresh_token_from(payload), self._verify_jwt(access_token)) + return self._session(access_token, refresh_token_from(payload), self._claims_for(access_token)) - def _get_jwks_client(self) -> Any: - if self._jwks_client is None: - self._jwks_client = make_jwks_client(self._jwks_url) - return self._jwks_client - def _verify_jwt(self, access_token: str) -> Dict[str, Any]: + def _claims_for(self, access_token: str) -> Dict[str, Any]: claims = verify_jwt( access_token, self._get_jwks_client(), algorithms=["RS256"], audience=self._client_id, # contract C2: bare client_id @@ -133,11 +110,11 @@ class NousDashboardAuthProvider(DashboardAuthProvider): 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: # Contract C4: no email / display_name in tokens. return session_from_claims( - self.name, claims, access_token=access_token, refresh_token=refresh_token, org_id=str(claims.get("org_id") or ""), - ) + self.name, claims, access_token=access_token, refresh_token=refresh_token, org_id=str(claims.get("org_id") or "")) # ---- Plugin entry point ---- diff --git a/plugins/dashboard_auth/self_hosted/__init__.py b/plugins/dashboard_auth/self_hosted/__init__.py index 87af9c31bf..64e2c092c6 100644 --- a/plugins/dashboard_auth/self_hosted/__init__.py +++ b/plugins/dashboard_auth/self_hosted/__init__.py @@ -19,15 +19,14 @@ 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 LoginStart, ProviderError, Session from plugins.dashboard_auth._shared import ( JSON_HEADERS, TOKEN_ENDPOINT_TIMEOUT_SEC as _TOKEN_ENDPOINT_TIMEOUT_SEC, + JwtOAuthProvider, SkipRegistration, exchange_token, load_config_section, - make_jwks_client, parse_json_body, pkce_login_start, refresh_token_from, @@ -65,7 +64,7 @@ def _require_https_or_loopback(url: str, *, field: str) -> str: raise ProviderError(f"OIDC {field} must be https:// (or http on localhost), got {url!r}") -class SelfHostedOIDCProvider(DashboardAuthProvider): +class SelfHostedOIDCProvider(JwtOAuthProvider): """Generic self-hosted OpenID Connect provider (authorization-code + PKCE).""" name = "self-hosted" @@ -92,8 +91,6 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): self._discovery_lock = threading.Lock() self._jwks_client: Any = None - # ---- public API (DashboardAuthProvider) ------------------------------- - def start_login(self, *, redirect_uri: str) -> LoginStart: # Validate the redirect before discovery so a bad redirect_uri surfaces even when the IDP is unreachable. validate_redirect_uri(redirect_uri) @@ -101,36 +98,6 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): return pkce_login_start( 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) - - def refresh_session(self, *, refresh_token: str) -> Session: - if not refresh_token: - raise RefreshExpiredError("no refresh token present in session") - return self._exchange( - { - "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}, - bad_request_exc=RefreshExpiredError, - 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) - # so this per-request check verifies a real JWT. None on expiry/invalidity; - # ProviderError if IDP/JWKS unreachable. - try: - claims = self._verify_id_token(access_token) - except InvalidCodeError: - return None - return self._session(access_token, "", claims) - def revoke_session(self, *, refresh_token: str) -> None: # Best-effort RFC 7009 revocation when the IDP advertises an endpoint. # Must never raise — logout is client-side cookie clearing regardless. @@ -152,7 +119,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): logger.debug("self-hosted OIDC: revoke failed (ignored): %s", exc) return None - # ---- internals: token exchange ---------------------------------------- + # ---- JwtOAuthProvider hooks: token exchange --------------------------- def _token_endpoint_auth(self, disco: Dict[str, Any]) -> tuple[Dict[str, str], Dict[str, str]]: """``(extra_data, extra_headers)`` for token-endpoint client auth. Public client → @@ -169,7 +136,18 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): userpass = f"{urllib.parse.quote(self._client_id, safe='')}:{urllib.parse.quote(self._client_secret, safe='')}" return {}, {"Authorization": f"Basic {base64.b64encode(userpass.encode('utf-8')).decode('ascii')}"} - def _exchange(self, data: Dict[str, str], *, bad_request_exc: type[Exception], previous_refresh_token: str = "") -> Session: + def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]: + # Re-request the same scopes so the rotated ID token keeps its identity claims + # (some IDPs narrow scope on refresh otherwise). + return ( + {"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token, + "scope": self._scopes}, + None) + + def _grant( + self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None, + previous_refresh_token: str = "", + ) -> Session: """POST the discovered token endpoint and turn the response into a Session. Confidential-client auth (body field or Basic header) is added for both grants — the IDP rejects an unauthenticated refresh with ``invalid_client``.""" @@ -234,8 +212,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): if advertised_issuer and advertised_issuer.rstrip("/") != self._issuer: raise ProviderError( f"OIDC discovery issuer mismatch: document advertises {advertised_issuer!r} " - f"but configured issuer is {self._issuer!r}" - ) + f"but configured issuer is {self._issuer!r}") for key, url in endpoints.items(): _require_https_or_loopback(url, field=key) # Absent/garbage auth-methods → [] → OIDC default (basic) applies. @@ -247,12 +224,10 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): "token_endpoint_auth_methods_supported": ( [str(m) for m in auth_methods_raw] if isinstance(auth_methods_raw, list) else [])} - # ---- internals: JWT verification + mapping ---------------------------- + # ---- JwtOAuthProvider hooks: verification + mapping ------------------- - def _get_jwks_client(self) -> Any: - if self._jwks_client is None: - self._jwks_client = make_jwks_client(self._get_discovery()["jwks_uri"]) - return self._jwks_client + def _jwks_uri(self) -> str: + return self._get_discovery()["jwks_uri"] def _verify_id_token(self, id_token: str) -> Dict[str, Any]: issuer = self._get_discovery()["issuer"] @@ -260,6 +235,8 @@ class SelfHostedOIDCProvider(DashboardAuthProvider): id_token, self._get_jwks_client(), algorithms=list(_ALLOWED_ID_TOKEN_ALGS), audience=self._client_id, issuer=issuer, label="ID token") + _claims_for = _verify_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 ``Session.access_token`` so the per-request ``verify_session`` re-verifies a real @@ -315,5 +292,4 @@ def register(ctx) -> None: if kw is not None: logger.info( "dashboard-auth-self-hosted: registered provider (issuer=%s, client_id=%s, scopes=%r, confidential=%s)", - kw["issuer"], kw["client_id"], kw["scopes"], bool(kw["client_secret"]), # never log the secret itself - ) + kw["issuer"], kw["client_id"], kw["scopes"], bool(kw["client_secret"])) # never log the secret itself From a1f0e6e657df8d58d625ce0097c113f050e442ff Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:54:41 -0700 Subject: [PATCH 2/3] refactor(plugins/kanban,dashboard_auth): compact docstrings/comments (WHY retained), document JwtOAuthProvider hook contract --- plugins/dashboard_auth/_shared.py | 28 ++---- plugins/kanban/dashboard/plugin_api.py | 134 ++++++++++--------------- 2 files changed, 59 insertions(+), 103 deletions(-) diff --git a/plugins/dashboard_auth/_shared.py b/plugins/dashboard_auth/_shared.py index 877224bf93..9d328eb970 100644 --- a/plugins/dashboard_auth/_shared.py +++ b/plugins/dashboard_auth/_shared.py @@ -244,29 +244,15 @@ class NonInteractiveMixin: class JwtOAuthProvider(DashboardAuthProvider): """Authorization-code + PKCE provider whose session token is a JWT we verify ourselves - (nous: Portal access token; self-hosted: OIDC ID token). Subclasses implement - ``_jwks_uri``, ``_claims_for``, ``_grant``, ``_refresh_request`` and ``_session``.""" + (nous: Portal access token; self-hosted: OIDC ID token). Subclasses set ``_client_id`` and + implement: ``_jwks_uri() -> str``; ``_claims_for(token) -> claims`` (raises + ``InvalidCodeError`` on expiry/foreign token, ``ProviderError`` otherwise); + ``_grant(data, *, bad_request_exc, headers=None, previous_refresh_token="") -> Session``; + ``_refresh_request(refresh_token) -> (form_data, extra_headers)``; + ``_session(token, refresh_token, claims) -> Session``.""" _jwks_client: Any = None - - def _jwks_uri(self) -> str: - raise NotImplementedError - - def _claims_for(self, token: str) -> Dict[str, Any]: - raise NotImplementedError - - def _grant( - self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None, - previous_refresh_token: str = "", - ) -> Session: - raise NotImplementedError - - def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]: - """``(form_data, extra_headers)`` for the refresh grant.""" - raise NotImplementedError - - def _session(self, token: str, refresh_token: str, claims: Dict[str, Any]) -> Session: - raise NotImplementedError + _client_id: str = "" def _get_jwks_client(self) -> Any: if self._jwks_client is None: diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py index a48b124ea6..2d28d5a44c 100644 --- a/plugins/kanban/dashboard/plugin_api.py +++ b/plugins/kanban/dashboard/plugin_api.py @@ -40,10 +40,9 @@ _BOARD_Q = Query(None, description="Kanban board slug (omit for current)") # --- Connection / board helpers --------------------------------------------- def _ws_upgrade_authorized(ws: "WebSocket") -> bool: - """Authorize a WebSocket upgrade via the dashboard's canonical WS gate - (``web_server._ws_auth_ok``: loopback ``?token=``, OAuth ``?ticket=``, internal ``?internal=``) - so this endpoint can never drift from core auth. Accepts when ``web_server`` isn't - importable (bare-FastAPI test harness).""" + """Authorize a WS upgrade via the dashboard's canonical gate (``web_server._ws_auth_ok``: + ``?token=`` / ``?ticket=`` / ``?internal=``) so this endpoint can never drift from core + auth; accepts when ``web_server`` isn't importable (bare-FastAPI test harness).""" try: from hermes_cli import web_server as _ws except Exception: @@ -78,9 +77,8 @@ def _existing_board_slug(slug: str) -> str: def _conn(board: Optional[str] = None): - """Open a kanban_db connection for the already-normalised ``board`` (``None`` = active). - ``init_db`` is idempotent; running it here lets a fresh install self-heal (no "no such - table" if POST /tasks arrives before GET /board).""" + """Connect to the already-normalised ``board`` (``None`` = active). ``init_db`` is + idempotent; running it here lets a fresh install self-heal if POST /tasks arrives first.""" try: kanban_db.init_db(board=board) except Exception as exc: @@ -97,9 +95,8 @@ def _board_conn(board: Optional[str]) -> Iterator[tuple[Optional[str], sqlite3.C def _with_board_pinned(board: Optional[str], fn: Callable[[], Any]) -> Any: - """Run ``fn`` with the board pinned context-locally rather than via the process-global - ``HERMES_KANBAN_BOARD`` env var: concurrent threadpool requests for different boards - would otherwise race on the shared env var and cross-write.""" + """Run ``fn`` with the board pinned context-locally, not via the process-global + ``HERMES_KANBAN_BOARD`` env var (concurrent requests for different boards would cross-write).""" with kanban_db.scoped_current_board(_resolve_board(board) or kanban_db.DEFAULT_BOARD): return fn() @@ -145,9 +142,8 @@ def _errors_to_500(prefix: str) -> Iterator[None]: # --- Serialization helpers -------------------------------------------------- -# Dashboard columns, left-to-right; "archived" is a filter toggle, not a column. -# Keep in sync with kanban_db.VALID_STATUSES — a status missing here gets -# mis-bucketed into ``todo`` by the board fallback. +# Dashboard columns, left-to-right ("archived" is a filter toggle, not a column). Keep in +# sync with kanban_db.VALID_STATUSES — a status missing here gets mis-bucketed into ``todo``. BOARD_COLUMNS: list[str] = ["triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done"] _CARD_SUMMARY_PREVIEW_CHARS = 200 @@ -177,10 +173,8 @@ def _placeholders(ids: list) -> str: def _compute_task_diagnostics(conn: sqlite3.Connection, task_ids: Optional[list[str]] = None) -> dict[str, list[dict]]: - """Run the diagnostic rule engine and return ``{task_id: [diagnostic_dict, ...]}`` - (tasks with no diagnostics omitted). Three aggregate queries (tasks, events, runs) - instead of N per-task lookups; slurps every event/run for the board — fine for the - dashboard's typical working set, paginate if profiling shows a hotspot.""" + """``{task_id: [diagnostic_dict, ...]}`` (tasks with none omitted) via three aggregate + queries (tasks, events, runs) — slurps the board; paginate if profiling shows a hotspot.""" from hermes_cli.config import load_config if task_ids is not None and not task_ids: @@ -255,8 +249,8 @@ def get_board( board: Optional[str] = _BOARD_Q, workflow_template_id: Optional[str] = Query(None, description="Restrict to tasks using this workflow template id"), current_step_key: Optional[str] = Query(None, description="Restrict to tasks at this workflow step key")): - """Full board grouped by status column. Omitting ``board`` falls through to the - active board (``HERMES_KANBAN_BOARD`` env → on-disk ``current`` pointer → ``default``).""" + """Full board grouped by status column; omitting ``board`` uses the active board + (``HERMES_KANBAN_BOARD`` env → on-disk ``current`` pointer → ``default``).""" with _board_conn(board) as (board, conn): tasks = kanban_db.list_tasks( conn, tenant=tenant, include_archived=include_archived, @@ -406,9 +400,8 @@ async def upload_task_attachment( file: UploadFile = File(...), board: Optional[str] = Query(None), uploaded_by: Optional[str] = Form(None)): - """Store an upload under ``attachments_root(board)//`` with a sanitised, - collision-resolved name and record its metadata. ``_safe_attachment_name`` raises - ``ValueError`` → 400.""" + """Store an upload under ``attachments_root(board)//`` (sanitised, + collision-resolved name; ``_safe_attachment_name`` ValueError → 400) and record it.""" with _board_conn(board) as (board, conn), _value_error_400(): _require_task(conn, task_id) safe_name = _safe_attachment_name(file.filename or "") @@ -514,10 +507,9 @@ _RUNNING_DIRECT_MSG = "Cannot set status to 'running' directly; use the dispatch def _drag_to(conn, task_id: str, s: str) -> bool: - """Drag-drop move into ready/todo/triage. blocked/scheduled -> ready re-opens via - ``unblock_task``; any task leaving ``review`` goes through ``reopen_review_task`` - (stale-run recovery, parent re-gate, ``review_reopened`` event) instead of a raw - status write. ``triage`` has no such transitions, so it skips the current-state query.""" + """Drag-drop into ready/todo/triage: blocked/scheduled -> ready re-opens via ``unblock_task``; + leaving ``review`` goes through ``reopen_review_task`` (stale-run recovery, parent re-gate, + ``review_reopened`` event) instead of a raw write; ``triage`` needs no current-state query.""" current = kanban_db.get_task(conn, task_id) if s != "triage" else None if s == "ready" and current and current.status in ("blocked", "scheduled"): return kanban_db.unblock_task(conn, task_id) @@ -526,10 +518,9 @@ def _drag_to(conn, task_id: str, s: str) -> bool: return _set_status_direct(conn, task_id, s) -# Status verb dispatch shared by PATCH /tasks/{id} and POST /tasks/bulk. Each handler is -# (conn, task_id, payload) -> ok. ``review`` routes through request_review (never a block, -# so it can't trip unblock-loop detection); ``force=True`` because a dashboard action is -# an explicit human override of a live worker claim. +# Status verb dispatch shared by PATCH /tasks/{id} and POST /tasks/bulk: (conn, task_id, +# payload) -> ok. ``review`` uses request_review (never a block, so it can't trip unblock-loop +# detection) with ``force=True``: a dashboard action is a human override of a live worker claim. _STATUS_HANDLERS: dict[str, Any] = { "done": lambda conn, tid, p: kanban_db.complete_task(conn, tid, result=p.result, summary=p.summary, metadata=p.metadata), "blocked": lambda conn, tid, p: kanban_db.block_task(conn, tid, reason=getattr(p, "block_reason", None)), @@ -558,8 +549,7 @@ def _set_priority(conn, task_id: str, priority: int, board: Optional[str]) -> No conn.execute( "INSERT INTO task_events (task_id, kind, payload, created_at) VALUES (?, 'reprioritized', ?, ?)", (task_id, json.dumps({"priority": int(priority)}), int(time.time()))) - # Mutation-boundary observer: this direct-SQL write bypasses every kanban_db mutator, - # so report it here — after the txn commits. + # Mutation-boundary observer (post-commit): this direct-SQL write bypasses every kanban_db mutator. kanban_db.notify_task_updated(conn, task_id, ("priority",), board=board) @@ -675,11 +665,9 @@ def _parents_blocking_ready(conn: sqlite3.Connection, task_id: str) -> list: def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str) -> bool: - """Direct status write for drag-drop moves not covered by the structured - complete/block/unblock/archive verbs (todo<->ready, running<->ready); appends a - ``status`` event. Leaving ``running`` closes the active run with outcome='reclaimed' - so attempt history isn't orphaned, and the worker is terminated only AFTER the txn - commits (events must be durable before the kill).""" + """Direct status write for drag-drop moves without a structured verb (todo<->ready, + running<->ready) + a ``status`` event. Leaving ``running`` closes the run as 'reclaimed' + so attempt history isn't orphaned; the worker is killed only AFTER the txn commits.""" terminations: list[tuple[Optional[int], Optional[str]]] = [] effective_status = new_status with kanban_db.write_txn(conn): @@ -773,9 +761,8 @@ def delete_link(parent_id: str = Query(...), child_id: str = Query(...), board: def _bulk_apply_one(conn, tid: str, payload: BulkTaskBody, board: Optional[str], entry: dict) -> None: - """Apply the bulk patch to one task, recording refusals in ``entry``. Unlike PATCH, - refusals don't abort the remaining ops — except a rejected status verb, which - short-circuits the whole entry (``_StatusRejected``).""" + """Apply the bulk patch to one task, recording refusals in ``entry`` without aborting the + remaining ops — except a rejected status verb (``_StatusRejected`` propagates).""" if payload.archive and not kanban_db.archive_task(conn, tid): entry.update(ok=False, error="archive refused") if payload.status is not None and not payload.archive: @@ -831,9 +818,8 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)): def list_diagnostics( board: Optional[str] = _BOARD_Q, severity: Optional[str] = Query(None, description="Filter by severity: warning|error|critical")): - """Return ``[{task_id, task_title, task_status, task_assignee, diagnostics}]`` for every - task with an active diagnostic, highest severity first then most recent. Also consumed - by ``hermes kanban diagnostics`` when the dashboard runs.""" + """Tasks with an active diagnostic, highest severity first then most recent; also + consumed by ``hermes kanban diagnostics`` when the dashboard runs.""" with _board_conn(board) as (board, conn): diags_by_task = _compute_task_diagnostics(conn, task_ids=None) if severity and diags_by_task: @@ -893,9 +879,8 @@ def get_run_endpoint(run_id: int, board: Optional[str] = _BOARD_Q): @router.get("/runs/{run_id}/inspect") def inspect_run_endpoint(run_id: int, board: Optional[str] = _BOARD_Q): - """Live psutil stats for a run's worker process. ``{alive: false, reason}`` when the run - ended, has no pid, the process is gone, or psutil is missing; ``access_denied`` style - errors are reported inline rather than as a 500.""" + """Live psutil stats for a run's worker; ``{alive: false, reason}`` when unavailable and + access-denied reported inline rather than as a 500.""" with _board_conn(board) as (board, conn): r = _require_run(conn, run_id) @@ -936,9 +921,8 @@ class TerminateRunBody(BaseModel): @router.post("/runs/{run_id}/terminate") def terminate_run_endpoint(run_id: int, payload: TerminateRunBody, board: Optional[str] = _BOARD_Q): - """Terminate the worker behind an in-flight run via ``reclaim_task`` so the - SIGTERM->SIGKILL flow, run bookkeeping, and event log match ``POST /tasks/{id}/reclaim``. - 404 unknown run; 409 already ended / not reclaimable.""" + """Terminate an in-flight run via ``reclaim_task`` (same SIGTERM->SIGKILL flow, bookkeeping + and events as ``POST /tasks/{id}/reclaim``); 409 if already ended / not reclaimable.""" with _board_conn(board) as (board, conn): r = _require_run(conn, run_id) if r.ended_at is not None: @@ -973,10 +957,8 @@ class SpecifyBody(BaseModel): @router.post("/tasks/{task_id}/specify") def specify_task_endpoint(task_id: str, payload: SpecifyBody, board: Optional[str] = Query(None)): - """Flesh out a triage task via the auxiliary LLM and promote it to ``todo`` - (``hermes kanban specify``). Returns ``{ok, task_id, reason, new_title}``; a non-OK - outcome is NOT an HTTP error — the UI renders the reason inline. Sync ``def`` so the - slow LLM call runs in FastAPI's threadpool.""" + """Flesh out a triage task via the auxiliary LLM (``hermes kanban specify``). Non-OK is NOT + an HTTP error — the UI renders the reason inline. Sync ``def`` → runs in the threadpool.""" def _run(): from hermes_cli import kanban_specify # lazy: missing aux client must not break plugin load return kanban_specify.specify_task(task_id, author=(payload.author or None)) @@ -1005,8 +987,7 @@ def reassign_task_endpoint(task_id: str, payload: ReassignBody, board: Optional[ return {"ok": True, "task_id": task_id, "assignee": payload.profile or None} -# Estimate: rough token/complexity read via the auxiliary model. NOT a dollar -# cost — providers don't report cost reliably. +# Estimate: rough token/complexity read via the auxiliary model. NOT a dollar cost. _ESTIMATE_SYSTEM_PROMPT = ( "You estimate how much work an autonomous coding agent will spend on a " "kanban task. Given the task title and description, respond with STRICT " @@ -1106,9 +1087,8 @@ def get_config(): # --- Home-channel subscriptions (per-task, per-platform toggles) ------------- -# Each gateway platform has at most one "home" (chat_id, thread_id, name). A toggle-on -# writes exactly the notify_subs row ``/kanban create`` would, so the existing gateway -# notifier delivers completed/blocked/gave_up with no extra plumbing. +# Each gateway platform has at most one "home" (chat_id, thread_id, name); a toggle-on writes +# exactly the notify_subs row ``/kanban create`` would, so the gateway notifier needs no plumbing. def _configured_home_channels() -> list[dict]: """Every platform with a home_channel, from the live GatewayConfig (so env overlays @@ -1228,9 +1208,8 @@ def dispatch(dry_run: bool = Query(False), max_n: int = Query(8, alias="max"), b @router.get("/model-options") def model_options(): - """Authenticated providers + curated models for the model-override dropdown, via - ``inventory.build_models_payload`` (same substrate as the Models page / TUI picker) so - the dropdown can't offer a pair the rest of Hermes rejects. Skips pricing enrichment + """Providers + curated models for the override dropdown via ``inventory.build_models_payload`` + (same substrate as the Models page) so it can't offer a pair Hermes rejects. Skips pricing and custom-provider probes: a slow/offline local endpoint must not hang the drawer.""" try: from hermes_cli.inventory import build_models_payload, load_picker_context @@ -1272,9 +1251,8 @@ class RenameBoardBody(BaseModel): project_id: Optional[str] = None -# Board transfer exchanges filesystem PATHS, not bytes (same contract as profile -# export/import): the desktop/dashboard clients run the native save/open dialog -# on the machine hosting the backend, so a path is all either side needs. +# Board transfer exchanges filesystem PATHS, not bytes (same contract as profile export/import): +# clients run the native save/open dialog on the machine hosting the backend. class ExportBoardBody(BaseModel): output: str = "" # empty → staging path under the kanban root @@ -1536,9 +1514,8 @@ def update_profile_description(profile_name: str, payload: DescribeBody): @router.post("/profiles/{profile_name}/describe-auto") def auto_describe_profile(profile_name: str, payload: DescribeAutoBody): - """Generate a description via ``auxiliary.profile_describer`` and persist it with - ``description_auto: true`` (``hermes profile describe --auto``). Non-OK - outcomes are NOT HTTP errors — the UI renders the reason inline.""" + """``hermes profile describe --auto``: persist with ``description_auto: true``. + Non-OK outcomes are NOT HTTP errors — the UI renders the reason inline.""" with _errors_to_500("describer crashed"): from hermes_cli import profile_describer outcome = profile_describer.describe_profile(profile_name, overwrite=bool(payload.overwrite)) @@ -1553,10 +1530,8 @@ class DecomposeBody(BaseModel): @router.post("/tasks/{task_id}/decompose") def decompose_task_endpoint(task_id: str, payload: DecomposeBody, board: Optional[str] = Query(None)): - """Fan a triage task out into child tasks via the auxiliary LLM, routed to specialist - profiles by description (``hermes kanban decompose``). Returns ``{ok, task_id, reason, - fanout, child_ids, new_title}``; non-OK is NOT an HTTP error. Sync ``def`` so the slow - LLM call runs in the threadpool.""" + """Fan a triage task out into child tasks via the auxiliary LLM (``hermes kanban decompose``). + Non-OK is NOT an HTTP error. Sync ``def`` → runs in the threadpool.""" def _run(): from hermes_cli import kanban_decompose return kanban_decompose.decompose_task(task_id, author=(payload.author or None)) @@ -1644,8 +1619,7 @@ def set_orchestration_settings(payload: OrchestrationSettingsBody): # --- WebSocket: /events?since=&board= ------------------------ -# Poll interval for the event tail loop. SQLite WAL + 300 ms polling is the -# simplest robust approach: negligible CPU, no shared state across workers. +# Event tail poll interval: WAL + 300 ms polling is the simplest robust approach (negligible CPU). _EVENT_POLL_SECONDS = 0.3 @@ -1670,9 +1644,8 @@ async def stream_events(ws: WebSocket): return await ws.accept() - # One SQLite connection per socket, used and closed only on a dedicated worker - # thread (sqlite connections are thread-affine); reusing it avoids churning - # WAL/SHM sidecars while an idle dashboard polls. + # One SQLite connection per socket, used/closed only on a dedicated worker thread + # (connections are thread-affine); reuse avoids churning WAL/SHM sidecars while idle. event_conn: Optional[sqlite3.Connection] = None event_executor: Optional[ThreadPoolExecutor] = None @@ -1706,9 +1679,8 @@ async def stream_events(ws: WebSocket): return (rows[-1]["id"] if rows else cursor_val), out while True: - # Race receive() against the poll interval so a client disconnect is detected - # even when no events are flowing; otherwise an idle board leaks zombie poll - # tasks until the next send_json() fails. Other client messages (pong, text) are ignored. + # Race receive() against the poll interval so a disconnect is detected even when no + # events flow (else idle boards leak poll tasks). Other client messages are ignored. try: msg = await asyncio.wait_for(ws.receive(), timeout=_EVENT_POLL_SECONDS) if msg["type"] == "websocket.disconnect": @@ -1723,9 +1695,7 @@ async def stream_events(ws: WebSocket): except WebSocketDisconnect: return except asyncio.CancelledError: - # Normal shutdown (Ctrl-C cancels the task mid-poll). CancelledError is a BaseException, - # so the Exception handler below wouldn't quiet it and Uvicorn would print a traceback. - return + return # normal shutdown; CancelledError is a BaseException the handler below wouldn't quiet except Exception as exc: # never crash the dashboard worker log.warning("Kanban event stream error: %s", exc) try: From ff6df69bb10248701be36781077ca37c050d65f7 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:08:57 -0700 Subject: [PATCH 3/3] refactor(plugins/kanban): _run_aux helper for specify/decompose auxiliary-LLM endpoints --- plugins/kanban/dashboard/plugin_api.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py index 2d28d5a44c..0aba6056bd 100644 --- a/plugins/kanban/dashboard/plugin_api.py +++ b/plugins/kanban/dashboard/plugin_api.py @@ -10,6 +10,7 @@ dispatcher's write txns); it carries its credential in the query string (browser from __future__ import annotations import asyncio +import importlib import json import logging import re @@ -108,6 +109,14 @@ def _require(getter: Callable, conn: sqlite3.Connection, ident, label: str): return obj +def _run_aux(board: Optional[str], module: str, fn: str, task_id: str, author: Optional[str]) -> Any: + """Run a slow auxiliary-LLM task helper (``hermes_cli..``) with the board pinned; + the module is imported lazily so a missing aux client can't break plugin load.""" + def _run(): + return getattr(importlib.import_module(f"hermes_cli.{module}"), fn)(task_id, author=(author or None)) + return _with_board_pinned(board, _run) + + def _require_task(conn: sqlite3.Connection, task_id: str) -> kanban_db.Task: return _require(kanban_db.get_task, conn, task_id, "task") @@ -959,11 +968,7 @@ class SpecifyBody(BaseModel): def specify_task_endpoint(task_id: str, payload: SpecifyBody, board: Optional[str] = Query(None)): """Flesh out a triage task via the auxiliary LLM (``hermes kanban specify``). Non-OK is NOT an HTTP error — the UI renders the reason inline. Sync ``def`` → runs in the threadpool.""" - def _run(): - from hermes_cli import kanban_specify # lazy: missing aux client must not break plugin load - return kanban_specify.specify_task(task_id, author=(payload.author or None)) - - outcome = _with_board_pinned(board, _run) + outcome = _run_aux(board, "kanban_specify", "specify_task", task_id, payload.author) return {"ok": bool(outcome.ok), "task_id": outcome.task_id, "reason": outcome.reason, "new_title": outcome.new_title} @@ -1532,11 +1537,7 @@ class DecomposeBody(BaseModel): def decompose_task_endpoint(task_id: str, payload: DecomposeBody, board: Optional[str] = Query(None)): """Fan a triage task out into child tasks via the auxiliary LLM (``hermes kanban decompose``). Non-OK is NOT an HTTP error. Sync ``def`` → runs in the threadpool.""" - def _run(): - from hermes_cli import kanban_decompose - return kanban_decompose.decompose_task(task_id, author=(payload.author or None)) - - outcome = _with_board_pinned(board, _run) + outcome = _run_aux(board, "kanban_decompose", "decompose_task", task_id, payload.author) return { "ok": bool(outcome.ok), "task_id": outcome.task_id, "reason": outcome.reason, "fanout": bool(outcome.fanout), "child_ids": outcome.child_ids or [], "new_title": outcome.new_title}