diff --git a/plugins/dashboard_auth/_shared.py b/plugins/dashboard_auth/_shared.py
index 72c12b296c..6535cb6210 100644
--- a/plugins/dashboard_auth/_shared.py
+++ b/plugins/dashboard_auth/_shared.py
@@ -1,9 +1,8 @@
 """Helpers shared by the bundled dashboard-auth providers.
 
-Each provider module keeps its own ``logger`` / ``LAST_SKIP_REASON`` (the
-gate reads those by module) and its ``register(ctx)``; the config/env
-resolution, PKCE login start, token-endpoint exchange and JWT verification
-boilerplate lives here.
+Each provider module keeps its own ``logger`` / ``LAST_SKIP_REASON`` (the gate reads
+those by module) and its ``register(ctx)``; the config/env resolution, skip/register
+bookkeeping, PKCE login start, token-endpoint exchange and JWT verification live here.
 """
 
 from __future__ import annotations
@@ -14,17 +13,11 @@ import logging
 import os
 import secrets
 import urllib.parse
-from typing import Any, Dict, Optional
+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 InvalidCodeError, LoginStart, ProviderError, Session, classify_jwks_lookup_error
 
 # JWKS Cache-Control max-age (nous contract C7); self-hosted mirrors it.
 JWKS_CACHE_SECONDS = 300
@@ -35,11 +28,8 @@ JSON_HEADERS = {"Accept": "application/json"}
 # ---- Config / env resolution ----
 
 def load_config_section(logger: logging.Logger, tag: str, *path: str) -> dict:
-    """Return the ``config.yaml`` block at ``path`` as a dict, or ``{}``.
-
-    Robust to load_config() raising (fresh install, malformed YAML), keys
-    being absent, or the value not being a dict — every shape yields ``{}``.
-    """
+    """The ``config.yaml`` block at ``path`` as a dict, or ``{}`` — robust to load_config()
+    raising (fresh install, malformed YAML), absent keys, or a non-dict value."""
     try:
         from hermes_cli.config import cfg_get, load_config
 
@@ -57,6 +47,40 @@ def resolve_env_or_cfg(env_name: str, cfg_value: Any) -> str:
     return os.environ.get(env_name, "").strip() or str(cfg_value or "").strip()
 
 
+# ---- register() bookkeeping ----
+
+class SkipRegistration(Exception):
+    """Raised by a provider's settings resolver to decline registration; ``reason`` is
+    the operator-facing text stored in the module's ``LAST_SKIP_REASON``."""
+
+    def __init__(self, reason: str, level: str = "debug") -> None:
+        super().__init__(reason)
+        self.reason, self.level = reason, level
+
+
+def register_provider(
+    ctx, logger: logging.Logger, tag: str, provider_cls: type, settings: Callable[[], dict],
+) -> tuple[Optional[dict], str]:
+    """Build ``provider_cls(**settings())`` and register it on ``ctx``.
+
+    Returns ``(kwargs, "")`` on success, ``(None, skip_reason)`` when ``settings`` raised
+    ``SkipRegistration`` (logged at its level) or construction raised ``ValueError`` /
+    ``ProviderError`` (logged as a warning). Callers store the reason in ``LAST_SKIP_REASON``.
+    """
+    try:
+        kwargs = settings()
+        provider = provider_cls(**kwargs)
+    except SkipRegistration as skip:
+        getattr(logger, skip.level)("%s: %s", tag, skip.reason)
+        return None, skip.reason
+    except (ValueError, ProviderError) as exc:
+        reason = f"{provider_cls.__name__} construction failed: {exc}"
+        logger.warning("%s: %s", tag, reason)
+        return None, reason
+    ctx.register_dashboard_auth_provider(provider)
+    return kwargs, ""
+
+
 # ---- OAuth / PKCE ----
 
 def b64url_no_pad(raw: bytes) -> str:
@@ -65,11 +89,9 @@ def b64url_no_pad(raw: bytes) -> str:
 
 
 def validate_redirect_uri(redirect_uri: str) -> None:
-    """Fast-fail obviously-broken redirect_uris before bouncing to the IDP.
-
-    The IDP's allowlist is authoritative. Any ``http://`` host is allowed so
-    dashboards behind TLS-terminating proxies / on LAN IPs aren't rejected.
-    """
+    """Fast-fail obviously-broken redirect_uris before bouncing to the IDP (whose allowlist
+    is authoritative). Any ``http://`` host is allowed so dashboards behind TLS-terminating
+    proxies / on LAN IPs aren't rejected."""
     parsed = urllib.parse.urlparse(redirect_uri)
     if parsed.scheme not in ("https", "http"):
         raise ProviderError(f"redirect_uri must be http(s), got {redirect_uri!r}")
@@ -77,23 +99,15 @@ def validate_redirect_uri(redirect_uri: str) -> None:
         raise ProviderError(f"redirect_uri path must end with '/auth/callback', got {redirect_uri!r}")
 
 
-def pkce_login_start(
-    authorize_url: str, *, client_id: str, scope: str, redirect_uri: str
-) -> LoginStart:
-    """Build the authorization-code + PKCE (S256) redirect and cookie payload.
-
-    Callers validate ``redirect_uri`` first. The auth-route layer expects
-    ``cookie_payload["hermes_session_pkce"]`` as a flat ``state=…;verifier=…``
-    string (it prepends ``provider=``).
-    """
+def pkce_login_start(authorize_url: str, *, client_id: str, scope: str, redirect_uri: str) -> LoginStart:
+    """Build the authorization-code + PKCE (S256) redirect and cookie payload. Callers
+    validate ``redirect_uri`` first. The auth-route layer expects
+    ``cookie_payload["hermes_session_pkce"]`` as a flat ``state=…;verifier=…`` string
+    (it prepends ``provider=``)."""
     code_verifier = b64url_no_pad(secrets.token_bytes(64))  # ~86 chars
     state = b64url_no_pad(secrets.token_bytes(32))
     params = {
-        "response_type": "code",
-        "client_id": client_id,
-        "redirect_uri": redirect_uri,
-        "scope": scope,
-        "state": state,
+        "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",
     }
@@ -115,38 +129,26 @@ 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,
+    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]]:
     """POST a token grant and return ``(token, payload)``.
 
-    A 400 (OAuth-shaped error envelope) raises ``bad_request_exc`` —
-    ``InvalidCodeError`` for the auth-code path, ``RefreshExpiredError`` for
-    refresh — so the middleware's distinct handling is preserved. Any other
-    non-200, transport failure, missing ``token_key`` or non-bearer
-    ``token_type`` raises ``ProviderError``. Redirects are deliberately NOT
-    followed: the body carries an auth code / refresh token.
+    A 400 (OAuth-shaped error envelope) raises ``bad_request_exc`` — ``InvalidCodeError``
+    for the auth-code path, ``RefreshExpiredError`` for refresh — so the middleware's
+    distinct handling is preserved. Any other non-200, transport failure, missing
+    ``token_key`` or non-bearer ``token_type`` raises ``ProviderError``. Redirects are
+    deliberately NOT followed: the body carries an auth code / refresh token.
     """
     try:
-        response = httpx.post(
-            url, data=data, headers={**JSON_HEADERS, **(headers or {})}, timeout=TOKEN_ENDPOINT_TIMEOUT_SEC
-        )
+        response = httpx.post(url, data=data, headers={**JSON_HEADERS, **(headers or {})}, timeout=TOKEN_ENDPOINT_TIMEOUT_SEC)
     except httpx.RequestError as exc:
         raise ProviderError(f"{endpoint} unreachable: {exc}") from exc
-
     if response.status_code == 400:
         error_code = parse_json_body(response).get("error", "invalid_request")
         raise bad_request_exc(f"{idp} rejected token request: {error_code}")
     if response.status_code != 200:
         raise ProviderError(f"{endpoint} returned {response.status_code}: {response.text[:200]!r}")
-
     payload = parse_json_body(response)
     token = payload.get(token_key)
     if not token or not isinstance(token, str):
@@ -165,37 +167,24 @@ 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 = "",
+    provider: str, claims: Dict[str, Any], *, access_token: str, refresh_token: str,
+    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,
+        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,
     )
 
 
 # ---- JWT verification ----
 
 def make_jwks_client(jwks_url: str) -> Any:
-    """PyJWKClient with explicit Accept/User-Agent (some WAFs block the
-    library default). Imported lazily so plugin discovery stays cheap."""
+    """PyJWKClient with explicit Accept/User-Agent (some WAFs block the library default).
+    Imported lazily so plugin discovery stays cheap."""
     from jwt import PyJWKClient
 
     return PyJWKClient(
@@ -205,22 +194,15 @@ def make_jwks_client(jwks_url: str) -> Any:
 
 
 def verify_jwt(
-    token: str,
-    jwks_client: Any,
-    *,
-    algorithms: list[str],
-    audience: str,
-    issuer: str,
-    label: str,
+    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 (opaque peer key, foreign kid) → ``InvalidCodeError`` (None /
-    next provider); folding both into 503 broke peer-key bearers. Expiry raises
-    ``InvalidCodeError`` (verify_session maps it to None); any other claim
-    failure raises ``ProviderError`` with the unverified iss/aud appended so
-    operators can spot config drift.
+    Unreachable JWKS → ``ProviderError`` (503); a bearer that is not one of our JWTs
+    (opaque peer key, foreign kid) → ``InvalidCodeError`` (None / next provider); folding
+    both into 503 broke peer-key bearers. Expiry raises ``InvalidCodeError`` (verify_session
+    maps it to None); any other claim failure raises ``ProviderError`` with the unverified
+    iss/aud appended so operators can spot config drift.
     """
     import jwt  # lazy — keeps startup fast for the ungated path
 
@@ -228,7 +210,6 @@ def verify_jwt(
         signing_key = jwks_client.get_signing_key_from_jwt(token)
     except Exception as exc:
         raise classify_jwks_lookup_error(exc) from exc
-
     try:
         return jwt.decode(
             token, signing_key.key, algorithms=algorithms, audience=audience, issuer=issuer,
@@ -237,8 +218,8 @@ def verify_jwt(
     except jwt.ExpiredSignatureError as exc:
         raise InvalidCodeError(f"{label} expired: {exc}") from exc
     except jwt.InvalidTokenError as exc:
-        # Decoding without verification is safe here: verification already
-        # failed and these values are surfaced for diagnostics only, never trusted.
+        # Decoding without verification is safe here: verification already failed and
+        # these values are surfaced for diagnostics only, never trusted.
         details = ""
         try:
             unverified = jwt.decode(token, options={"verify_signature": False, "verify_exp": False})
diff --git a/plugins/dashboard_auth/basic/__init__.py b/plugins/dashboard_auth/basic/__init__.py
index 5b0c361857..85c6805b29 100644
--- a/plugins/dashboard_auth/basic/__init__.py
+++ b/plugins/dashboard_auth/basic/__init__.py
@@ -1,10 +1,12 @@
 """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``);
-everything downstream (cookies, verify, refresh, ws-tickets, logout) is shared.
-No IDP and no database: sessions are stateless HMAC-signed tokens this provider
-mints and verifies — zero infrastructure for a single-box dashboard.
+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)::
 
@@ -16,21 +18,11 @@ Configuration (env wins over config.yaml when set non-empty)::
         secret: "<32+ random bytes, base64 or hex>"  # optional signing key
         session_ttl_seconds: 43200    # optional access-token lifetime (12h)
 
-    HERMES_DASHBOARD_BASIC_AUTH_USERNAME
-    HERMES_DASHBOARD_BASIC_AUTH_PASSWORD_HASH   # preferred
-    HERMES_DASHBOARD_BASIC_AUTH_PASSWORD        # plaintext fallback
-    HERMES_DASHBOARD_BASIC_AUTH_SECRET
-    HERMES_DASHBOARD_BASIC_AUTH_TTL_SECONDS
+    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.
-
-Passwords use stdlib :func:`hashlib.scrypt` (no third-party dependency).
-``complete_password_login`` is constant-time and 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.
 """
 
 from __future__ import annotations
@@ -45,16 +37,11 @@ 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 load_config_section, resolve_env_or_cfg
+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__)
+_TAG = "dashboard-auth-basic"
 
 # The middleware transparently refreshes via the 30-day refresh token when the
 # access token lapses, so the TTL controls refresh frequency, not login length.
@@ -78,17 +65,11 @@ LAST_SKIP_REASON: str = ""
 # ---- Password hashing (stdlib scrypt) ----
 
 def hash_password(password: str) -> str:
-    """Return a ``scrypt$n$r$p$<salt_b64>$<dk_b64>`` hash string.
-
-    Public so operators can 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'))"``.
-    """
+    """Return a ``scrypt$n$r$p$<salt_b64>$<dk_b64>`` hash string. Public so operators can
+    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,
-    )
+    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()}"
 
 
@@ -104,10 +85,7 @@ def _verify_password(password: str, encoded: str) -> bool:
     except (ValueError, TypeError):
         return False
     try:
-        actual = hashlib.scrypt(
-            password.encode("utf-8"), salt=salt, n=n, r=r, p=p,
-            dklen=len(expected), maxmem=0,
-        )
+        actual = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=n, r=r, p=p, dklen=len(expected), maxmem=0)
     except (ValueError, MemoryError):
         return False
     return hmac.compare_digest(actual, expected)
@@ -154,14 +132,7 @@ class BasicAuthProvider(DashboardAuthProvider):
     display_name = "Username & Password"
     supports_password = True
 
-    def __init__(
-        self,
-        *,
-        username: str,
-        password_hash: str,
-        secret: bytes,
-        ttl_seconds: int = _DEFAULT_TTL_SECONDS,
-    ) -> None:
+    def __init__(self, *, username: str, password_hash: str, secret: bytes, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> None:
         if not username:
             raise ValueError("username must be non-empty")
         if not password_hash:
@@ -181,22 +152,17 @@ class BasicAuthProvider(DashboardAuthProvider):
             "The login page POSTs to /auth/password-login instead."
         )
 
-    def complete_login(
-        self, *, code: str, state: str, code_verifier: str, redirect_uri: str
-    ) -> Session:
+    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:
-        # Always run a scrypt verify (real hash if the username matches, else
-        # the dummy) and compare the username with compare_digest too, so
-        # neither the username nor its length leaks via timing.
-        username_ok = hmac.compare_digest(
-            username.encode("utf-8"), self._username.encode("utf-8")
-        )
-        target_hash = self._password_hash if username_ok else _DUMMY_HASH
-        password_ok = _verify_password(password, target_hash)
+        # Always run a scrypt verify (real hash if the username matches, else the dummy)
+        # and compare the username with compare_digest too, so neither the username nor
+        # its length leaks via timing.
+        username_ok = hmac.compare_digest(username.encode("utf-8"), self._username.encode("utf-8"))
+        password_ok = _verify_password(password, self._password_hash if username_ok else _DUMMY_HASH)
         if not (username_ok and password_ok):
             raise InvalidCredentialsError("invalid username or password")
         return self._mint_session(self._username)
@@ -207,8 +173,7 @@ class BasicAuthProvider(DashboardAuthProvider):
         payload = _unsign(access_token, self._secret, "access")
         if payload is None:
             return None
-        user_id = str(payload.get("sub", ""))
-        return self._session(user_id, int(payload["exp"]), access_token, "")
+        return self._session(str(payload.get("sub", "")), int(payload["exp"]), access_token, "")
 
     def refresh_session(self, *, refresh_token: str) -> Session:
         if not refresh_token:
@@ -219,8 +184,7 @@ class BasicAuthProvider(DashboardAuthProvider):
         return self._mint_session(str(payload.get("sub", self._username)))
 
     def revoke_session(self, *, refresh_token: str) -> None:
-        # Stateless tokens — nothing to revoke server-side; the session
-        # expires within its TTL. Must not raise.
+        # Stateless tokens — nothing to revoke server-side; the session expires within its TTL. Must not raise.
         return None
 
     # ---- internals ---------------------------------------------------------
@@ -244,15 +208,13 @@ class BasicAuthProvider(DashboardAuthProvider):
 # ---- Plugin entry point ----
 
 def _load_config_basic_auth_section() -> dict:
-    return load_config_section(logger, "dashboard-auth-basic", "dashboard", "basic_auth")
+    return load_config_section(logger, _TAG, "dashboard", "basic_auth")
 
 
 def _resolve_secret(cfg_section: dict) -> bytes:
-    """Resolve the token-signing secret (base64, hex, or raw text).
-
-    When unset, generates a random per-process secret (sessions then don't
-    survive a restart or span multiple workers — logged at INFO).
-    """
+    """Resolve the token-signing secret (base64, hex, or raw text). When unset, generates
+    a random per-process secret (sessions then don't survive a restart or span multiple
+    workers — logged at INFO)."""
     raw = resolve_env_or_cfg("HERMES_DASHBOARD_BASIC_AUTH_SECRET", cfg_section.get("secret"))
     if not raw:
         logger.info(
@@ -272,12 +234,8 @@ def _resolve_secret(cfg_section: dict) -> bytes:
     return raw.encode("utf-8")
 
 
-def register(ctx) -> None:
-    """Register ``BasicAuthProvider`` when username + (password or
-    password_hash) are configured; a no-op for OAuth / ``--insecure`` setups."""
-    global LAST_SKIP_REASON
-    LAST_SKIP_REASON = ""
-
+def _settings() -> dict:
+    """Resolve BasicAuthProvider kwargs from env/config; raises ``SkipRegistration``."""
     section = _load_config_basic_auth_section()
 
     def setting(env_name: str, cfg_key: str) -> str:
@@ -287,36 +245,27 @@ def register(ctx) -> None:
     password_hash = setting("HERMES_DASHBOARD_BASIC_AUTH_PASSWORD_HASH", "password_hash")
     plaintext = setting("HERMES_DASHBOARD_BASIC_AUTH_PASSWORD", "password")
     ttl_raw = setting("HERMES_DASHBOARD_BASIC_AUTH_TTL_SECONDS", "session_ttl_seconds")
-
     if not username:
-        LAST_SKIP_REASON = (
+        raise SkipRegistration(
             "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."
         )
-        logger.debug("dashboard-auth-basic: %s", LAST_SKIP_REASON)
-        return
-
     if not password_hash and not plaintext:
-        LAST_SKIP_REASON = (
+        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)."
+            "with plugins.dashboard_auth.basic.hash_password).",
+            level="warning",
         )
-        logger.warning("dashboard-auth-basic: %s", LAST_SKIP_REASON)
-        return
-
-    # 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).
+    # 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).
     plaintext_from_env = os.environ.get("HERMES_DASHBOARD_BASIC_AUTH_PASSWORD", "").strip()
     if plaintext_from_env:
         password_hash = hash_password(plaintext_from_env)
-        logger.info(
-            "dashboard-auth-basic: hashed env-supplied password in-memory "
-            "(overrides any config password_hash)."
-        )
+        logger.info("dashboard-auth-basic: hashed env-supplied password in-memory (overrides any config password_hash).")
     elif not password_hash:
         password_hash = hash_password(plaintext)
         logger.info(
@@ -324,21 +273,18 @@ def register(ctx) -> None:
             "For production, precompute dashboard.basic_auth.password_hash "
             "and remove the plaintext password from config."
         )
-
     try:
         ttl = int(ttl_raw) if ttl_raw else _DEFAULT_TTL_SECONDS
     except ValueError:
         ttl = _DEFAULT_TTL_SECONDS
+    return {"username": username, "password_hash": password_hash, "secret": _resolve_secret(section), "ttl_seconds": ttl}
 
-    try:
-        provider = BasicAuthProvider(
-            username=username, password_hash=password_hash,
-            secret=_resolve_secret(section), ttl_seconds=ttl,
-        )
-    except ValueError as exc:
-        LAST_SKIP_REASON = f"BasicAuthProvider construction failed: {exc}"
-        logger.warning("dashboard-auth-basic: %s", LAST_SKIP_REASON)
-        return
 
-    ctx.register_dashboard_auth_provider(provider)
-    logger.info("dashboard-auth-basic: registered password provider (username=%s)", username)
+def register(ctx) -> None:
+    """Register ``BasicAuthProvider`` when username + (password or
+    password_hash) are configured; a no-op for OAuth / ``--insecure`` setups."""
+    global LAST_SKIP_REASON
+    LAST_SKIP_REASON = ""
+    kwargs, LAST_SKIP_REASON = register_provider(ctx, logger, _TAG, BasicAuthProvider, _settings)
+    if kwargs is not None:
+        logger.info("dashboard-auth-basic: registered password provider (username=%s)", kwargs["username"])
diff --git a/plugins/dashboard_auth/drain/__init__.py b/plugins/dashboard_auth/drain/__init__.py
index 132ebf5853..e1420508b8 100644
--- a/plugins/dashboard_auth/drain/__init__.py
+++ b/plugins/dashboard_auth/drain/__init__.py
@@ -1,33 +1,16 @@
 """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 deployed agent's environment; this provider
-verifies an inbound ``Authorization`` bearer against it with a constant-time
-compare and vouches for the caller as the ``drain-control`` principal. No
-login, cookie, session or refresh — the interactive ABC methods raise
-``NotImplementedError``. A real auth plugin (not an ad-hoc header check on the
-route) so the credential lives inside the dashboard auth framework.
-
-Security properties:
-* Per-agent unique secret — a leak's blast radius is one agent.
-* 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).
-* ``hmac.compare_digest`` on the request path — not a timing oracle.
-
-Configuration — the secret is a CREDENTIAL, so it is env-only (provisioned by
-NAS at deploy time); behavioural knobs live in config.yaml::
-
-    HERMES_DASHBOARD_DRAIN_SECRET   # per-agent shared secret (>=43 url-safe-b64 chars)
-
-    dashboard:
-      drain_auth:
-        scope: drain            # capability label attached to the principal
-        min_secret_chars: 43    # entropy bar (default 43 ~= 256 bits)
-
-When the env var is unset the plugin is a no-op (records a skip reason).
+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).
 """
 from __future__ import annotations
 
@@ -38,15 +21,11 @@ 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 load_config_section
+from hermes_cli.dashboard_auth import DashboardAuthProvider, LoginStart, Session, TokenPrincipal
+from plugins.dashboard_auth._shared import SkipRegistration, load_config_section, register_provider
 
 logger = logging.getLogger(__name__)
+_TAG = "dashboard-auth-drain"
 
 # token_urlsafe(32) produces exactly 43 chars, so a correctly-provisioned
 # secret clears the default bar exactly.
@@ -71,14 +50,10 @@ def _shannon_bits(value: str) -> float:
     return per_char * n
 
 
-def assess_secret_strength(
-    secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_CHARS
-) -> Optional[str]:
-    """Return a human-readable rejection reason if ``secret`` is too weak, else ``None``.
-
-    Checks, in order: length >= ``min_chars``, distinct chars >=
-    ``_MIN_DISTINCT_CHARS``, Shannon entropy >= ``_MIN_SHANNON_BITS``.
-    """
+def assess_secret_strength(secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_CHARS) -> Optional[str]:
+    """Human-readable rejection reason if ``secret`` is too weak, else ``None``. Checks, in
+    order: length >= ``min_chars``, distinct chars >= ``_MIN_DISTINCT_CHARS``, Shannon
+    entropy >= ``_MIN_SHANNON_BITS``."""
     if not secret:
         return "secret is empty"
     if len(secret) < min_chars:
@@ -89,16 +64,10 @@ def assess_secret_strength(
         )
     distinct = len(set(secret))
     if distinct < _MIN_DISTINCT_CHARS:
-        return (
-            f"secret has only {distinct} distinct characters (need >= "
-            f"{_MIN_DISTINCT_CHARS}); looks structured/low-entropy"
-        )
+        return f"secret has only {distinct} distinct characters (need >= {_MIN_DISTINCT_CHARS}); looks structured/low-entropy"
     bits = _shannon_bits(secret)
     if bits < _MIN_SHANNON_BITS:
-        return (
-            f"secret entropy too low: {bits:.0f} bits (need >= "
-            f"{_MIN_SHANNON_BITS:.0f}); looks structured/repeated"
-        )
+        return f"secret entropy too low: {bits:.0f} bits (need >= {_MIN_SHANNON_BITS:.0f}); looks structured/repeated"
     return None
 
 
@@ -131,18 +100,14 @@ 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."
-        )
+        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:
+    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.
+        # 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:
@@ -155,64 +120,51 @@ class DrainSecretProvider(DashboardAuthProvider):
 # ---- Plugin entry point ----
 
 def _load_config_drain_auth_section() -> dict:
-    return load_config_section(logger, "dashboard-auth-drain", "dashboard", "drain_auth")
+    return load_config_section(logger, _TAG, "dashboard", "drain_auth")
 
 
-def register(ctx) -> None:
-    """Register ``DrainSecretProvider`` when a strong secret is set.
-
-    No-op (records a skip reason) when ``HERMES_DASHBOARD_DRAIN_SECRET`` is
-    unset or fails the entropy gate. On success also registers the drain
-    route as token-authable via the generic seam.
-    """
-    global LAST_SKIP_REASON
-    LAST_SKIP_REASON = ""
-
+def _settings() -> dict:
+    """Resolve DrainSecretProvider kwargs from env/config; raises ``SkipRegistration``."""
     secret = os.environ.get("HERMES_DASHBOARD_DRAIN_SECRET", "").strip()
     if not secret:
-        LAST_SKIP_REASON = (
+        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."
         )
-        logger.debug("dashboard-auth-drain: %s", LAST_SKIP_REASON)
-        return
-
     section = _load_config_drain_auth_section()
     scope = str(section.get("scope", "drain") or "drain").strip() or "drain"
     try:
         min_chars = int(section.get("min_secret_chars", _DEFAULT_MIN_SECRET_CHARS))
     except (TypeError, ValueError):
         min_chars = _DEFAULT_MIN_SECRET_CHARS
-
     reason = assess_secret_strength(secret, min_chars=min_chars)
     if reason is not None:
-        LAST_SKIP_REASON = (
-            f"HERMES_DASHBOARD_DRAIN_SECRET rejected — {reason}. "
-            "The drain endpoint stays disabled (fail-closed)."
+        raise SkipRegistration(
+            f"HERMES_DASHBOARD_DRAIN_SECRET rejected — {reason}. The drain endpoint stays disabled (fail-closed).",
+            level="warning",
         )
-        logger.warning("dashboard-auth-drain: %s", LAST_SKIP_REASON)
+    return {"secret": secret, "scope": scope}
+
+
+def register(ctx) -> None:
+    """Register ``DrainSecretProvider`` when a strong secret is set; no-op (records a skip
+    reason) when ``HERMES_DASHBOARD_DRAIN_SECRET`` is unset or fails the entropy gate. On
+    success also registers the drain route as token-authable via the generic seam."""
+    global LAST_SKIP_REASON
+    LAST_SKIP_REASON = ""
+    kwargs, LAST_SKIP_REASON = register_provider(ctx, logger, _TAG, DrainSecretProvider, _settings)
+    if kwargs is None:
         return
-
-    try:
-        provider = DrainSecretProvider(secret=secret, scope=scope)
-    except ValueError as exc:
-        LAST_SKIP_REASON = f"DrainSecretProvider construction failed: {exc}"
-        logger.warning("dashboard-auth-drain: %s", LAST_SKIP_REASON)
-        return
-
-    ctx.register_dashboard_auth_provider(provider)
-
-    # Opt the drain endpoint into the token-auth seam so the interactive
-    # cookie gate doesn't bounce NAS's bearer call.
+    # Opt the drain endpoint into the token-auth seam so the interactive cookie gate
+    # doesn't bounce NAS's bearer call.
     try:
         from hermes_cli.dashboard_auth.token_auth import register_token_route
 
         register_token_route(DRAIN_ROUTE_PATH)
     except Exception as exc:  # noqa: BLE001 — seam import must not crash plugin load
         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)",
-        scope, DRAIN_ROUTE_PATH,
+        kwargs["scope"], DRAIN_ROUTE_PATH,
     )
diff --git a/plugins/dashboard_auth/nous/__init__.py b/plugins/dashboard_auth/nous/__init__.py
index 1c0400b50b..861abbec3e 100644
--- a/plugins/dashboard_auth/nous/__init__.py
+++ b/plugins/dashboard_auth/nous/__init__.py
@@ -1,36 +1,18 @@
 """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, so
-loopback / ``--insecure`` operators are unaffected.
+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.
 
-Configuration (env wins over config.yaml when set non-empty; an empty env value
-is treated as unset so a provisioned-but-blank Fly secret can't shadow config)::
-
-    dashboard:
-      oauth:
-        client_id: agent:{agent_instance_id}   # required
-        portal_url: https://portal.example     # optional
-
-    HERMES_DASHBOARD_OAUTH_CLIENT_ID  — shape ``agent:{agent_instance_id}``
-    HERMES_DASHBOARD_PORTAL_URL       — default https://portal.nousresearch.com
-
-Contract points encoded here:
-  - client_id is per-instance; its suffix is cross-checked against the token's
-    ``agent_instance_id`` claim (defense-in-depth).
-  - scope is ``agent_dashboard:access`` only; audience is the bare client_id.
-  - access tokens are RS256 JWTs verified against ``/.well-known/jwks.json``
-    (JWKS cached 5 minutes).
-  - 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) trips reuse-detection, revoking the whole session. A 400 on
-    refresh (expired / revoked / reuse-detected) → ``RefreshExpiredError`` →
-    re-login.
-  - ``oauth_contract_version``: missing → warn + proceed; ``!= 1`` → refuse.
-
-``LAST_SKIP_REASON`` is read by the gate's fail-closed branch to surface a
-specific operator message when the plugin loads but declines to register.
+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.
 """
 
 from __future__ import annotations
@@ -38,20 +20,15 @@ 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,
     load_config_section,
     make_jwks_client,
     pkce_login_start,
     refresh_token_from,
+    register_provider,
     resolve_env_or_cfg,
     session_from_claims,
     validate_redirect_uri,
@@ -59,6 +36,7 @@ from plugins.dashboard_auth._shared import (
 )
 
 logger = logging.getLogger(__name__)
+_TAG = "dashboard-auth-nous"
 
 _DEFAULT_PORTAL_URL = "https://portal.nousresearch.com"
 _SCOPE = "agent_dashboard:access"  # contract C3
@@ -74,8 +52,7 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
     display_name = "Nous Research"
 
     def __init__(self, *, client_id: str, portal_url: str) -> None:
-        # Defense-in-depth: register() filters too, but a malformed id must
-        # never construct a provider.
+        # Defense-in-depth: register() filters too, but a malformed id must never construct a provider.
         if not client_id.startswith("agent:"):
             raise ValueError(f"client_id must match contract shape 'agent:{{instance_id}}', got {client_id!r}")
         self._client_id = client_id
@@ -90,22 +67,14 @@ class NousDashboardAuthProvider(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
-        )
+        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.
+    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,
+                "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
+                "client_id": self._client_id, "code_verifier": code_verifier,
             },
             bad_request_exc=InvalidCodeError,
         )
@@ -113,22 +82,18 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
     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.
+        # 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,
-            },
+            {"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.
+        # 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:
@@ -136,15 +101,14 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
         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.
+        # Portal exposes no token-endpoint revocation grant; logout is client-side cookie
+        # clearing and the RT expires within its 24h TTL.
         return None
 
     # ---- internals --------------------------------------------------------
 
     def _token_grant(
-        self, data: Dict[str, str], *, bad_request_exc: type[Exception],
-        headers: Optional[Dict[str, str]] = None,
+        self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None,
     ) -> Session:
         access_token, payload = exchange_token(
             self._token_url, data, headers=headers, bad_request_exc=bad_request_exc,
@@ -165,60 +129,45 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
             audience=self._client_id,  # contract C2: bare client_id
             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).
+        # 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} "
-                f"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.",
+                "Nous Portal token missing oauth_contract_version claim (contract says it should be %d); proceeding anyway.",
                 _EXPECTED_CONTRACT_VERSION,
             )
         elif contract_version != _EXPECTED_CONTRACT_VERSION:
             raise ProviderError(
-                f"unsupported oauth_contract_version={contract_version!r}, "
-                f"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:
         # 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 ----
 
 def _load_config_oauth_section() -> dict:
-    return load_config_section(logger, "dashboard-auth-nous", "dashboard", "oauth")
+    return load_config_section(logger, _TAG, "dashboard", "oauth")
 
 
-def register(ctx) -> None:
-    """Register ``NousDashboardAuthProvider`` when a client_id is configured.
-
-    On skip, writes a reason naming BOTH configuration surfaces to
-    :data:`LAST_SKIP_REASON` for the gate's fail-closed error message.
-    """
-    global LAST_SKIP_REASON
-    LAST_SKIP_REASON = ""
-
+def _settings() -> dict:
+    """Resolve NousDashboardAuthProvider kwargs; the skip reason names BOTH configuration surfaces."""
     section = _load_config_oauth_section()
     client_id = resolve_env_or_cfg("HERMES_DASHBOARD_OAUTH_CLIENT_ID", section.get("client_id", ""))
-    portal_url = (
-        resolve_env_or_cfg("HERMES_DASHBOARD_PORTAL_URL", section.get("portal_url", ""))
-        or _DEFAULT_PORTAL_URL
-    )
-
+    portal_url = resolve_env_or_cfg("HERMES_DASHBOARD_PORTAL_URL", section.get("portal_url", "")) or _DEFAULT_PORTAL_URL
     if not client_id:
-        LAST_SKIP_REASON = (
+        raise SkipRegistration(
             "HERMES_DASHBOARD_OAUTH_CLIENT_ID is not set (and dashboard.oauth.client_id "
             "in config.yaml is empty). The Nous Portal provisions this env var (shape "
             "'agent:{instance_id}') when it deploys a Hermes Agent instance — set it to "
@@ -226,24 +175,20 @@ def register(ctx) -> None:
             "dashboard.oauth.client_id in config.yaml), or pass --insecure to skip the "
             "OAuth gate entirely."
         )
-        logger.debug("dashboard-auth-nous: %s", LAST_SKIP_REASON)
-        return
-
     if not client_id.startswith("agent:"):
-        LAST_SKIP_REASON = (
+        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."
+            f"time; check your Fly app's secrets or override with the value from the Portal admin UI.",
+            level="warning",
         )
-        logger.warning("dashboard-auth-nous: %s", LAST_SKIP_REASON)
-        return
+    return {"client_id": client_id, "portal_url": portal_url}
 
-    try:
-        provider = NousDashboardAuthProvider(client_id=client_id, portal_url=portal_url)
-    except ValueError as exc:
-        LAST_SKIP_REASON = f"NousDashboardAuthProvider construction failed: {exc}"
-        logger.warning("dashboard-auth-nous: %s", LAST_SKIP_REASON)
-        return
 
-    ctx.register_dashboard_auth_provider(provider)
-    logger.info("dashboard-auth-nous: registered provider (client_id=%s, portal=%s)", client_id, portal_url)
+def register(ctx) -> None:
+    """Register ``NousDashboardAuthProvider`` when a client_id is configured."""
+    global LAST_SKIP_REASON
+    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"])
diff --git a/plugins/dashboard_auth/self_hosted/__init__.py b/plugins/dashboard_auth/self_hosted/__init__.py
index decdd7078a..d6c9631576 100644
--- a/plugins/dashboard_auth/self_hosted/__init__.py
+++ b/plugins/dashboard_auth/self_hosted/__init__.py
@@ -1,28 +1,16 @@
 """SelfHostedOIDCProvider — generic self-hosted OpenID Connect dashboard auth.
 
-A standards-compliant OIDC Relying Party for the ``hermes dashboard`` gate.
-Unlike the ``nous`` provider (Nous Portal's bespoke contract), this speaks
-plain OIDC so it works against 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 only:
-
-  1. discovers endpoints from ``{issuer}/.well-known/openid-configuration``,
-  2. builds the ``/authorize`` URL with PKCE (S256),
-  3. exchanges the code at the discovered ``token_endpoint``,
-  4. verifies the **ID token** against the discovered ``jwks_uri`` with
-     ``iss``/``aud`` pinned, mapping ``sub``/``email``/``name`` onto a Session.
-
-Why the ID token, not the access token: OIDC guarantees the ID token is a
-signed JWT carrying identity claims; the access token's format is opaque per
-spec (many IDPs issue random strings). The ``nous`` provider verifies its
-access token only because Portal mints a custom JWT there.
-
-Public (PKCE-only) and confidential (PKCE + ``client_secret``) clients are
-both supported. With a secret, the client additionally authenticates at the
-token endpoint via ``client_secret_basic`` or ``client_secret_post`` chosen
-from ``token_endpoint_auth_methods_supported``. PKCE is sent in both modes —
-the secret is layered on top, never a replacement (OAuth 2.1 / RFC 9700).
+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)::
 
@@ -35,13 +23,10 @@ Configuration (env wins over config.yaml when set non-empty)::
           scopes: "openid profile email"                           # optional
           # client_secret: confidential clients only — prefer the env var.
 
-    HERMES_DASHBOARD_OIDC_ISSUER
-    HERMES_DASHBOARD_OIDC_CLIENT_ID
-    HERMES_DASHBOARD_OIDC_SCOPES         # optional
+    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) the module-level :data:`LAST_SKIP_REASON`
-carries a human-readable reason for the gate's fail-closed error.
+On skip (missing issuer / client_id) ``LAST_SKIP_REASON`` carries the reason for the gate.
 """
 
 from __future__ import annotations
@@ -55,23 +40,18 @@ 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,
+    SkipRegistration,
     exchange_token,
     load_config_section,
     make_jwks_client,
     parse_json_body,
     pkce_login_start,
     refresh_token_from,
+    register_provider,
     resolve_env_or_cfg,
     session_from_claims,
     validate_redirect_uri,
@@ -79,14 +59,14 @@ from plugins.dashboard_auth._shared import (
 )
 
 logger = logging.getLogger(__name__)
+_TAG = "dashboard-auth-self-hosted"
 
-# ``openid`` is mandatory (no ID token without it); profile/email populate
-# display_name/email.
+# ``openid`` is mandatory (no ID token without it); profile/email populate display_name/email.
 _DEFAULT_SCOPES = "openid profile email"
 
-# RS256 is the OIDC default; ES256 is common on modern IDPs (Zitadel, newer
-# Keycloak). HS256 is deliberately excluded: it implies a shared secret we
-# don't hold in the public-client model and is a JWT algorithm-confusion footgun.
+# RS256 is the OIDC default; ES256 is common on modern IDPs (Zitadel, newer Keycloak).
+# HS256 is deliberately excluded: it implies a shared secret we don't hold in the
+# public-client model and is a JWT algorithm-confusion footgun.
 _ALLOWED_ID_TOKEN_ALGS = ("RS256", "ES256", "RS384", "RS512", "ES384", "ES512")
 
 _DISCOVERY_TIMEOUT_SEC = 10.0
@@ -101,14 +81,9 @@ def _require_https_or_loopback(url: str, *, field: str) -> str:
     """Reject non-HTTPS endpoint URLs (loopback http allowed) so a misconfigured
     issuer can't ship auth codes / refresh tokens in cleartext."""
     parsed = urllib.parse.urlparse(url)
-    if parsed.scheme == "https" or (
-        parsed.scheme == "http"
-        and (parsed.hostname or "") in ("localhost", "127.0.0.1", "::1")
-    ):
+    if parsed.scheme == "https" or (parsed.scheme == "http" and (parsed.hostname or "") in ("localhost", "127.0.0.1", "::1")):
         return url
-    raise ProviderError(
-        f"OIDC {field} must be https:// (or http on localhost), got {url!r}"
-    )
+    raise ProviderError(f"OIDC {field} must be https:// (or http on localhost), got {url!r}")
 
 
 class SelfHostedOIDCProvider(DashboardAuthProvider):
@@ -117,30 +92,22 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
     name = "self-hosted"
     display_name = "Self-Hosted OIDC"
 
-    def __init__(
-        self,
-        *,
-        issuer: str,
-        client_id: str,
-        scopes: str = _DEFAULT_SCOPES,
-        client_secret: str = "",
-    ) -> None:
+    def __init__(self, *, issuer: str, client_id: str, scopes: str = _DEFAULT_SCOPES, client_secret: str = "") -> None:
         if not issuer:
             raise ValueError("issuer is required")
         if not client_id:
             raise ValueError("client_id is required")
-        # Trailing slash normalised for stable compares; ``iss`` is pinned
-        # against the *discovered* issuer so a config/IDP slash mismatch is tolerated.
+        # Trailing slash normalised for stable compares; ``iss`` is pinned against the
+        # *discovered* issuer so a config/IDP slash mismatch is tolerated.
         self._issuer = issuer.rstrip("/")
         _require_https_or_loopback(self._issuer, field="issuer")
         self._client_id = client_id
         self._scopes = scopes.strip() or _DEFAULT_SCOPES
-        # Empty/whitespace secret ⇒ public client, so a provisioned-but-blank
-        # secret can't flip us into a broken confidential mode.
+        # Empty/whitespace secret ⇒ public client, so a provisioned-but-blank secret
+        # can't flip us into a broken confidential mode.
         self._client_secret = (client_secret or "").strip()
-
-        # Discovery + JWKS resolve lazily so registration never hits the
-        # network (the IDP may be down at boot; fail per-request instead).
+        # Discovery + JWKS resolve lazily so registration never hits the network
+        # (the IDP may be down at boot; fail per-request instead).
         self._discovery: Dict[str, Any] | None = None
         self._discovery_fetched_at: float = 0.0
         self._discovery_lock = threading.Lock()
@@ -149,26 +116,19 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
     # ---- 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 the redirect before discovery so a bad redirect_uri surfaces even when the IDP is unreachable.
         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:
+    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,
+                "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
+                "client_id": self._client_id, "code_verifier": code_verifier,
             },
             bad_request_exc=InvalidCodeError,
         )
@@ -178,11 +138,9 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
             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).
+                "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,
@@ -190,9 +148,9 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
         )
 
     def verify_session(self, *, access_token: str) -> Optional[Session]:
-        # The session cookie carries the ID token in the access-token slot
-        # (see _session_from_tokens) so this per-request check verifies a real
-        # JWT. None on expiry/invalidity; ProviderError if IDP/JWKS unreachable.
+        # 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:
@@ -223,46 +181,29 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
     # ---- internals: token exchange ----------------------------------------
 
     def _token_endpoint_auth(self, disco: Dict[str, Any]) -> tuple[Dict[str, str], Dict[str, str]]:
-        """Return ``(extra_data, extra_headers)`` for token-endpoint client auth.
-
-        Public client → ``({}, {})`` (PKCE alone). Confidential client →
-        ``client_secret_post`` when the IDP advertises it *without*
-        ``client_secret_basic``, else HTTP Basic (the OIDC default and the
-        fallback when nothing is advertised). RFC 6749 §2.3.1.
-        """
+        """``(extra_data, extra_headers)`` for token-endpoint client auth. Public client →
+        ``({}, {})`` (PKCE alone). Confidential client → ``client_secret_post`` when the IDP
+        advertises it *without* ``client_secret_basic``, else HTTP Basic (the OIDC default
+        and the fallback when nothing is advertised). RFC 6749 §2.3.1."""
         if not self._client_secret:
             return {}, {}
         methods = disco.get("token_endpoint_auth_methods_supported") or []
         if "client_secret_post" in methods and "client_secret_basic" not in methods:
             return {"client_secret": self._client_secret}, {}
-        # Both halves must be form-url-encoded *before* base64 (RFC 6749
-        # §2.3.1) or a secret containing ':' / reserved chars corrupts the header.
+        # Both halves must be form-url-encoded *before* base64 (RFC 6749 §2.3.1) or a
+        # secret containing ':' / reserved chars corrupts the header.
         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 _exchange(self, data: Dict[str, str], *, bad_request_exc: type[Exception], 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 here
-        for both grants — the IDP rejects an unauthenticated refresh with
-        ``invalid_client``. For a public client the request is PKCE-only.
-        """
+        Confidential-client auth (body field or Basic header) is added for both grants —
+        the IDP rejects an unauthenticated refresh with ``invalid_client``."""
         disco = self._get_discovery()
         extra_data, extra_headers = self._token_endpoint_auth(disco)
         id_token, payload = exchange_token(
-            disco["token_endpoint"],
-            {**data, **extra_data},
-            headers=extra_headers,
-            bad_request_exc=bad_request_exc,
-            idp="IDP",
-            endpoint="OIDC token endpoint",
-            token_key="id_token",
+            disco["token_endpoint"], {**data, **extra_data}, headers=extra_headers, bad_request_exc=bad_request_exc,
+            idp="IDP", endpoint="OIDC token endpoint", token_key="id_token",
             missing_msg=(
                 "OIDC token response missing id_token — ensure the 'openid' "
                 "scope is configured and the client is allowed to receive an "
@@ -289,20 +230,17 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
                 if disco is None:
                     disco = self._discovery = self._fetch_discovery()
                     self._discovery_fetched_at = time.time()
-                    # New issuer/keys → rebind the JWKS client to the fresh jwks_uri.
-                    self._jwks_client = None
+                    self._jwks_client = None  # new issuer/keys → rebind the JWKS client to the fresh jwks_uri
         return disco
 
     def _fetch_discovery(self) -> Dict[str, Any]:
         url = f"{self._issuer}/.well-known/openid-configuration"
         try:
-            # follow_redirects=True: many IDPs answer discovery with a 3xx
-            # (Authentik canonicalises .well-known; proxies upgrade http→https)
-            # and httpx defaults to not following. Safe because the issuer pin
-            # and HTTPS checks below validate the *resolved* document, so a
-            # redirect to a hostile location can't smuggle in a bad issuer or a
-            # cleartext endpoint. The token/revocation POSTs deliberately do
-            # NOT follow redirects (they carry an auth code / refresh token).
+            # follow_redirects=True: many IDPs answer discovery with a 3xx (Authentik
+            # canonicalises .well-known; proxies upgrade http→https) and httpx defaults to
+            # not following. Safe because the issuer pin and HTTPS checks below validate the
+            # *resolved* document. The token/revocation POSTs deliberately do NOT follow
+            # redirects (they carry an auth code / refresh token).
             response = httpx.get(url, headers=JSON_HEADERS, timeout=_DISCOVERY_TIMEOUT_SEC, follow_redirects=True)
         except httpx.RequestError as exc:
             raise ProviderError(f"OIDC discovery unreachable: {exc}") from exc
@@ -317,23 +255,16 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
 
         endpoints = {k: field(k) for k in ("authorization_endpoint", "token_endpoint", "jwks_uri")}
         if not all(endpoints.values()):
-            raise ProviderError(
-                "OIDC discovery missing one of authorization_endpoint / "
-                "token_endpoint / jwks_uri"
-            )
-
+            raise ProviderError("OIDC discovery missing one of authorization_endpoint / token_endpoint / jwks_uri")
         # Issuer pin: a mismatch means the document came from the wrong place
         # (proxy/MITM/misconfig). Only a trailing-slash difference is tolerated.
         advertised_issuer = field("issuer")
         if advertised_issuer and advertised_issuer.rstrip("/") != self._issuer:
             raise ProviderError(
-                f"OIDC discovery issuer mismatch: document advertises "
-                f"{advertised_issuer!r} but configured issuer is "
-                f"{self._issuer!r}"
+                f"OIDC discovery issuer mismatch: document advertises {advertised_issuer!r} 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.
         auth_methods_raw = payload.get("token_endpoint_auth_methods_supported")
         return {
@@ -360,26 +291,19 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
         )
 
     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 JWT; the opaque OAuth
-        access token is not kept — the dashboard only needs identity.
-        """
+        """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
+        JWT; the opaque OAuth access token is not kept — the dashboard only needs identity."""
         email = str(claims.get("email", "") or "")
-        # Org/tenant is non-standard: accept common spellings, else join
-        # ``groups`` so multi-tenant IDPs surface *something* (free-form string).
+        # Org/tenant is non-standard: accept common spellings, else join ``groups`` so
+        # multi-tenant IDPs surface *something* (free-form string).
         org_id = claims.get("org_id") or claims.get("organization") or ""
         groups = claims.get("groups")
         if not org_id and isinstance(groups, list) and groups:
             org_id = ",".join(str(g) for g in groups)
         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 ""
-            ),
+            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 ""),
         )
 
@@ -387,18 +311,11 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
 # ---- Plugin entry point ----
 
 def _load_config_oauth_section() -> dict:
-    return load_config_section(logger, "dashboard-auth-self-hosted", "dashboard", "oauth", "self_hosted")
+    return load_config_section(logger, _TAG, "dashboard", "oauth", "self_hosted")
 
 
-def register(ctx) -> None:
-    """Register :class:`SelfHostedOIDCProvider` when issuer + client_id are set.
-
-    On skip, :data:`LAST_SKIP_REASON` names BOTH configuration surfaces so
-    operators don't guess wrong about which to set.
-    """
-    global LAST_SKIP_REASON
-    LAST_SKIP_REASON = ""
-
+def _settings() -> dict:
+    """Resolve SelfHostedOIDCProvider kwargs; the skip reason names BOTH configuration surfaces."""
     oidc_cfg = _load_config_oauth_section()
 
     def setting(env_name: str, cfg_key: str) -> str:
@@ -406,12 +323,8 @@ def register(ctx) -> None:
 
     issuer = setting("HERMES_DASHBOARD_OIDC_ISSUER", "issuer")
     client_id = setting("HERMES_DASHBOARD_OIDC_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")
-
     if not issuer or not client_id:
-        LAST_SKIP_REASON = (
+        raise SkipRegistration(
             "Self-hosted OIDC dashboard auth is not configured. Set both an issuer and "
             "a client_id — either as env vars (HERMES_DASHBOARD_OIDC_ISSUER + "
             "HERMES_DASHBOARD_OIDC_CLIENT_ID) or under "
@@ -419,21 +332,21 @@ def register(ctx) -> None:
             "--insecure to skip the OAuth gate entirely. (issuer set: %s; client_id set: %s)"
             % (bool(issuer), bool(client_id))
         )
-        logger.debug("dashboard-auth-self-hosted: %s", LAST_SKIP_REASON)
-        return
+    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"),
+    }
 
-    try:
-        provider = SelfHostedOIDCProvider(
-            issuer=issuer, client_id=client_id, scopes=scopes, client_secret=client_secret
+
+def register(ctx) -> None:
+    """Register :class:`SelfHostedOIDCProvider` when issuer + client_id are set."""
+    global LAST_SKIP_REASON
+    LAST_SKIP_REASON = ""
+    kw, LAST_SKIP_REASON = register_provider(ctx, logger, _TAG, SelfHostedOIDCProvider, _settings)
+    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
         )
-    except (ValueError, ProviderError) as exc:
-        LAST_SKIP_REASON = f"SelfHostedOIDCProvider construction failed: {exc}"
-        logger.warning("dashboard-auth-self-hosted: %s", LAST_SKIP_REASON)
-        return
-
-    ctx.register_dashboard_auth_provider(provider)
-    logger.info(
-        "dashboard-auth-self-hosted: registered provider "
-        "(issuer=%s, client_id=%s, scopes=%r, confidential=%s)",
-        issuer, client_id, scopes, bool(client_secret),  # never log the secret itself
-    )
diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py
index 3c4d16c70c..42562d3c50 100644
--- a/plugins/kanban/dashboard/plugin_api.py
+++ b/plugins/kanban/dashboard/plugin_api.py
@@ -1,17 +1,12 @@
-"""Kanban dashboard plugin — backend API routes.
+"""Kanban dashboard plugin — backend API routes, mounted at /api/plugins/kanban/.
 
-Mounted at /api/plugins/kanban/ by the dashboard plugin system. Every handler
-is a thin wrapper around ``hermes_cli.kanban_db`` (the same code paths the CLI
-and gateway ``/kanban`` command use, so the three surfaces cannot drift).
-
-Live updates arrive via the ``/events`` WebSocket, which tails the append-only
-``task_events`` table on a short poll (WAL mode lets reads run alongside the
-dispatcher's IMMEDIATE write transactions).
-
-Security: plugin HTTP routes sit behind the dashboard's session-token auth
-middleware like core API routes. The ``/events`` WebSocket carries its
-credential in the query string (browsers cannot set ``Authorization`` on an
-upgrade request) and is gated by the dashboard's canonical WS auth check.
+Every handler is a thin wrapper around ``hermes_cli.kanban_db`` (the same code
+paths the CLI and gateway ``/kanban`` command use, so the surfaces cannot drift).
+Live updates: the ``/events`` WebSocket tails the append-only ``task_events``
+table on a short poll (WAL reads run alongside the dispatcher's write txns).
+HTTP routes sit behind the dashboard's session-token middleware; the WebSocket
+carries its credential in the query string (browsers can't set ``Authorization``
+on an upgrade) and is gated by the dashboard's canonical WS auth check.
 """
 
 from __future__ import annotations
@@ -23,7 +18,7 @@ import re
 import sqlite3
 import time
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 from dataclasses import asdict
 from pathlib import Path
 from typing import Any, Callable, Iterator, Optional
@@ -34,24 +29,22 @@ from pydantic import BaseModel, Field
 
 from hermes_cli import kanban_db
 from hermes_cli import kanban_diagnostics as kd
-from hermes_cli.kanban_db import (
-    KANBAN_ATTACHMENT_MAX_BYTES,
-    _collision_free_path,
-    _safe_attachment_name,
-)
+from hermes_cli.kanban_db import KANBAN_ATTACHMENT_MAX_BYTES, _collision_free_path, _safe_attachment_name
 
 log = logging.getLogger(__name__)
 
 router = APIRouter()
 
+_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)."""
+    (``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)."""
     try:
         from hermes_cli import web_server as _ws
     except Exception:
@@ -86,9 +79,9 @@ 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 board). ``init_db`` is idempotent; running it here means a fresh install
-    self-heals (no "no such table" if POST /tasks arrives before GET /board)."""
+    """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)."""
     try:
         kanban_db.init_db(board=board)
     except Exception as exc:
@@ -100,17 +93,14 @@ def _conn(board: Optional[str] = None):
 def _board_conn(board: Optional[str]) -> Iterator[tuple[Optional[str], sqlite3.Connection]]:
     """Resolve the ``board`` query param, open a connection, close it on exit."""
     board = _resolve_board(board)
-    conn = _conn(board=board)
-    try:
+    with closing(_conn(board=board)) as conn:
         yield board, conn
-    finally:
-        conn.close()
 
 
 def _with_board_pinned(board: Optional[str], fn: Callable[[], Any]) -> Any:
-    """Run ``fn`` with the board pinned context-locally (not 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 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."""
     with kanban_db.scoped_current_board(_resolve_board(board) or kanban_db.DEFAULT_BOARD):
         return fn()
 
@@ -152,11 +142,9 @@ 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 (e.g.
-# ``scheduled``) gets mis-bucketed into ``todo`` by the board fallback.
-BOARD_COLUMNS: list[str] = [
-    "triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done",
-]
+# Keep in sync with kanban_db.VALID_STATUSES — a status missing here gets
+# mis-bucketed into ``todo`` by the board fallback.
+BOARD_COLUMNS: list[str] = ["triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done"]
 
 _CARD_SUMMARY_PREVIEW_CHARS = 200
 
@@ -168,8 +156,7 @@ def _task_dict(task: kanban_db.Task, *, latest_summary: Optional[str] = None) ->
         d["age"] = kanban_db.task_age(task)
     except Exception:
         d["age"] = {"created_age_seconds": None, "started_age_seconds": None, "time_to_complete_seconds": None}
-    # Latest non-null run summary (workers hand off via ``task_runs.summary``,
-    # not ``tasks.result``); None until a run has produced one.
+    # Latest non-null run summary (workers hand off via ``task_runs.summary``, not ``tasks.result``).
     d["latest_summary"] = latest_summary
     return d
 
@@ -177,44 +164,38 @@ def _task_dict(task: kanban_db.Task, *, latest_summary: Optional[str] = None) ->
 def _attachment_dict(a: kanban_db.Attachment) -> dict[str, Any]:
     """``stored_path`` is the absolute on-disk path workers read; UI downloads by ``id``."""
     return {
-        "id": a.id, "task_id": a.task_id, "filename": a.filename,
-        "content_type": a.content_type, "size": a.size, "uploaded_by": a.uploaded_by,
-        "stored_path": a.stored_path, "created_at": a.created_at,
+        "id": a.id, "task_id": a.task_id, "filename": a.filename, "content_type": a.content_type,
+        "size": a.size, "uploaded_by": a.uploaded_by, "stored_path": a.stored_path, "created_at": a.created_at,
     }
 
 
-def _compute_task_diagnostics(
-    conn: sqlite3.Connection,
-    task_ids: Optional[list[str]] = None,
-) -> dict[str, list[dict]]:
-    """Run the diagnostic rule engine (``kanban_diagnostics``) and return
-    ``{task_id: [diagnostic_dict, ...]}``; tasks with no diagnostics are omitted.
+def _placeholders(ids: list) -> str:
+    return ",".join(["?"] * len(ids))
 
-    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 (hundreds of tasks); paginate if profiling shows a hotspot.
-    """
+
+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."""
     from hermes_cli.config import load_config
 
     diag_config = kd.config_from_runtime_config(load_config())
-
     if task_ids is not None:
         if not task_ids:
             return {}
-        placeholders = ",".join(["?"] * len(task_ids))
-        rows = conn.execute(f"SELECT * FROM tasks WHERE id IN ({placeholders})", tuple(task_ids)).fetchall()
+        rows = conn.execute(f"SELECT * FROM tasks WHERE id IN ({_placeholders(task_ids)})", tuple(task_ids)).fetchall()
     else:
         rows = conn.execute("SELECT * FROM tasks WHERE status != 'archived'").fetchall()
     if not rows:
         return {}
 
     row_ids = [r["id"] for r in rows]
-    placeholders = ",".join(["?"] * len(row_ids))
 
     def _rows_by_task(table: str) -> dict[str, list]:
         by_task: dict[str, list] = {tid: [] for tid in row_ids}
         for row in conn.execute(
-            f"SELECT * FROM {table} WHERE task_id IN ({placeholders}) ORDER BY id", tuple(row_ids),
+            f"SELECT * FROM {table} WHERE task_id IN ({_placeholders(row_ids)}) ORDER BY id", tuple(row_ids),
         ):
             by_task.setdefault(row["task_id"], []).append(row)
         return by_task
@@ -226,8 +207,7 @@ def _compute_task_diagnostics(
     for r in rows:
         tid = r["id"]
         diags = kd.compute_task_diagnostics(
-            r, events_by_task.get(tid, []), runs_by_task.get(tid, []),
-            config=diag_config, graph=graph_by_task.get(tid),
+            r, events_by_task.get(tid, []), runs_by_task.get(tid, []), config=diag_config, graph=graph_by_task.get(tid),
         )
         if diags:
             out[tid] = [d.to_dict() for d in diags]
@@ -235,8 +215,7 @@ def _compute_task_diagnostics(
 
 
 def _warnings_summary_from_diagnostics(diagnostics: list[dict]) -> Optional[dict]:
-    """Compact card badge summary ``{count, kinds, latest_at, highest_severity}``;
-    None when ``diagnostics`` is empty."""
+    """Compact card badge summary ``{count, kinds, latest_at, highest_severity}``; None when empty."""
     if not diagnostics:
         return None
     kinds: dict[str, int] = {}
@@ -254,8 +233,7 @@ def _warnings_summary_from_diagnostics(diagnostics: list[dict]) -> Optional[dict
 
 
 def _attach_diagnostics(task_d: dict, diags: Optional[list[dict]]) -> None:
-    """Full list goes in the payload (drawer renders without a second round-trip);
-    the card badge only needs the summary."""
+    """Full list in the payload (drawer renders without a second round-trip); card badge gets the summary."""
     if diags:
         task_d["diagnostics"] = diags
         task_d["warnings"] = _warnings_summary_from_diagnostics(diags)
@@ -264,11 +242,7 @@ def _attach_diagnostics(task_d: dict, diags: Optional[list[dict]]) -> None:
 def _links_for(conn: sqlite3.Connection, task_id: str) -> dict[str, list[str]]:
     """Return {'parents': [...], 'children': [...]} for a task."""
     def _ids(col: str, other: str) -> list[str]:
-        return [
-            r[col] for r in conn.execute(
-                f"SELECT {col} FROM task_links WHERE {other} = ? ORDER BY {col}", (task_id,),
-            )
-        ]
+        return [r[col] for r in conn.execute(f"SELECT {col} FROM task_links WHERE {other} = ? ORDER BY {col}", (task_id,))]
     return {"parents": _ids("parent_id", "child_id"), "children": _ids("child_id", "parent_id")}
 
 
@@ -278,10 +252,8 @@ def _links_for(conn: sqlite3.Connection, task_id: str) -> dict[str, list[str]]:
 def get_board(
     tenant: Optional[str] = Query(None, description="Filter to a single tenant"),
     include_archived: bool = Query(False),
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
-    workflow_template_id: Optional[str] = Query(
-        None, description="Restrict to tasks using this workflow template id",
-    ),
+    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
@@ -291,23 +263,18 @@ def get_board(
             conn, tenant=tenant, include_archived=include_archived,
             workflow_template_id=workflow_template_id, current_step_key=current_step_key,
         )
-        # Link / comment / progress rollups are each one aggregate query rather
-        # than N per-task lookups.
+        # Link / comment / progress rollups are each one aggregate query rather than N per-task lookups.
         link_counts: dict[str, dict[str, int]] = {}
         for row in conn.execute("SELECT parent_id, child_id FROM task_links").fetchall():
             link_counts.setdefault(row["parent_id"], {"parents": 0, "children": 0})["children"] += 1
             link_counts.setdefault(row["child_id"], {"parents": 0, "children": 0})["parents"] += 1
-
         comment_counts: dict[str, int] = {
-            r["task_id"]: r["n"]
-            for r in conn.execute("SELECT task_id, COUNT(*) AS n FROM task_comments GROUP BY task_id")
+            r["task_id"]: r["n"] for r in conn.execute("SELECT task_id, COUNT(*) AS n FROM task_comments GROUP BY task_id")
         }
-
         # Per parent: children done / total, rendered as "N/M".
         progress: dict[str, dict[str, int]] = {}
         for row in conn.execute(
-            "SELECT l.parent_id AS pid, t.status AS cstatus "
-            "FROM task_links l JOIN tasks t ON t.id = l.child_id"
+            "SELECT l.parent_id AS pid, t.status AS cstatus FROM task_links l JOIN tasks t ON t.id = l.child_id"
         ).fetchall():
             p = progress.setdefault(row["pid"], {"done": 0, "total": 0})
             p["total"] += 1
@@ -316,15 +283,12 @@ def get_board(
 
         diagnostics_per_task = _compute_task_diagnostics(conn, task_ids=None)
         latest_event_id = conn.execute("SELECT COALESCE(MAX(id), 0) AS m FROM task_events").fetchone()["m"]
-
         columns: dict[str, list[dict]] = {c: [] for c in BOARD_COLUMNS}
         if include_archived:
             columns["archived"] = []
-
-        # One window-function query for latest summaries (avoids N+1); cards get
-        # a truncated preview, the full text comes from /tasks/:id.
+        # One window-function query for latest summaries (avoids N+1); cards get a
+        # truncated preview, the full text comes from /tasks/:id.
         summary_map = kanban_db.latest_summaries(conn, [t.id for t in tasks])
-
         for t in tasks:
             full = summary_map.get(t.id)
             d = _task_dict(t, latest_summary=(full[:_CARD_SUMMARY_PREVIEW_CHARS] if full else None))
@@ -335,18 +299,12 @@ def get_board(
             columns[t.status if t.status in columns else "todo"].append(d)
 
         # Per-column ordering (priority DESC, created_at ASC) comes from list_tasks.
-        tenants = [
-            r["tenant"]
-            for r in conn.execute("SELECT DISTINCT tenant FROM tasks WHERE tenant IS NOT NULL ORDER BY tenant")
-        ]
+        tenants = [r["tenant"] for r in conn.execute("SELECT DISTINCT tenant FROM tasks WHERE tenant IS NOT NULL ORDER BY tenant")]
         assignees = [
-            r["assignee"]
-            for r in conn.execute(
-                "SELECT DISTINCT assignee FROM tasks WHERE assignee IS NOT NULL "
-                "AND status != 'archived' ORDER BY assignee"
+            r["assignee"] for r in conn.execute(
+                "SELECT DISTINCT assignee FROM tasks WHERE assignee IS NOT NULL AND status != 'archived' ORDER BY assignee"
             )
         ]
-
         return {
             "columns": [{"name": name, "tasks": columns[name]} for name in columns],
             "tenants": tenants, "assignees": assignees,
@@ -360,18 +318,12 @@ def get_board(
 def get_task(
     task_id: str,
     board: Optional[str] = Query(None),
-    run_state_type: Optional[str] = Query(
-        None, description="With run_state_name: filter runs by column 'status' or 'outcome'",
-    ),
-    run_state_name: Optional[str] = Query(
-        None, description="With run_state_type: exact value for that run column",
-    ),
+    run_state_type: Optional[str] = Query(None, description="With run_state_name: filter runs by column 'status' or 'outcome'"),
+    run_state_name: Optional[str] = Query(None, description="With run_state_type: exact value for that run column"),
 ):
     with _board_conn(board) as (board, conn):
         if (run_state_type is None) ^ (run_state_name is None):
-            raise HTTPException(
-                status_code=400, detail="run_state_type and run_state_name must be passed together or omitted",
-            )
+            raise HTTPException(status_code=400, detail="run_state_type and run_state_name must be passed together or omitted")
         if run_state_type is not None and run_state_type not in ("status", "outcome"):
             raise HTTPException(status_code=400, detail="run_state_type must be 'status' or 'outcome'")
         task = _require_task(conn, task_id)
@@ -379,27 +331,20 @@ def get_task(
         task_d = _task_dict(task, latest_summary=kanban_db.latest_summary(conn, task_id))
         links = _links_for(conn, task_id)
         child_summaries = kanban_db.latest_summaries(conn, links["children"])
-        child_results = []
-        for child_id in links["children"]:
-            child = kanban_db.get_task(conn, child_id)
-            if child is None:
-                continue
-            child_results.append({
-                "id": child.id, "title": child.title, "status": child.status,
-                "latest_summary": child_summaries.get(child.id), "result": child.result,
-            })
-        diags = _compute_task_diagnostics(conn, task_ids=[task_id])
-        _attach_diagnostics(task_d, diags.get(task_id) or [])
+        children = filter(None, (kanban_db.get_task(conn, cid) for cid in links["children"]))
+        _attach_diagnostics(task_d, _compute_task_diagnostics(conn, task_ids=[task_id]).get(task_id) or [])
         return {
             "task": task_d,
             "comments": [asdict(c) for c in kanban_db.list_comments(conn, task_id)],
             "events": [asdict(e) for e in kanban_db.list_events(conn, task_id)],
             "attachments": [_attachment_dict(a) for a in kanban_db.list_attachments(conn, task_id)],
             "links": links,
-            "child_results": child_results,
+            "child_results": [
+                {"id": c.id, "title": c.title, "status": c.status, "latest_summary": child_summaries.get(c.id), "result": c.result}
+                for c in children
+            ],
             "runs": [
-                asdict(r)
-                for r in kanban_db.list_runs(conn, task_id, state_type=run_state_type, state_name=run_state_name)
+                asdict(r) for r in kanban_db.list_runs(conn, task_id, state_type=run_state_type, state_name=run_state_name)
             ],
         }
 
@@ -436,16 +381,16 @@ def create_task(payload: CreateTaskBody, board: Optional[str] = Query(None)):
         task_id = kanban_db.create_task(conn, created_by="dashboard", board=board, **payload.model_dump())
         task = kanban_db.get_task(conn, task_id)
         body: dict[str, Any] = {"task": _task_dict(task) if task else None}
-        # Dispatcher-presence warning so the UI can banner a ready+assigned task
-        # that would otherwise sit idle (no gateway / dispatch_in_gateway=false).
-        # triage/todo are expected to wait; unassigned tasks can't dispatch anyway.
+        # Dispatcher-presence warning so the UI can banner a ready+assigned task that would
+        # otherwise sit idle (no gateway / dispatch_in_gateway=false). triage/todo are
+        # expected to wait; unassigned tasks can't dispatch anyway.
         if task and task.status == "ready" and task.assignee:
             try:
                 from hermes_cli.kanban import _check_dispatcher_presence
                 from hermes_constants import get_hermes_home
 
-                # Probe the request's active home: the dashboard backend may run
-                # under a different HERMES_HOME than the board's profile.
+                # Probe the request's active home: the dashboard backend may run under a
+                # different HERMES_HOME than the board's profile.
                 running, message = _check_dispatcher_presence(hermes_home=get_hermes_home())
                 if not running and message:
                     body["warning"] = message
@@ -455,9 +400,8 @@ def create_task(payload: CreateTaskBody, board: Optional[str] = Query(None)):
 
 
 # --- Attachments — upload / list / download / delete ------------------------
-# Size cap, filename sanitiser, and collision resolver live in ``kanban_db`` so
-# the dashboard, agent toolset, and CLI share one implementation.
-# ``_safe_attachment_name`` raises ``ValueError`` → mapped to 400 below.
+# Size cap, filename sanitiser, and collision resolver live in ``kanban_db`` so the
+# dashboard, agent toolset, and CLI share one implementation.
 
 @router.get("/tasks/{task_id}/attachments")
 def list_task_attachments(task_id: str, board: Optional[str] = Query(None)):
@@ -473,12 +417,12 @@ async def upload_task_attachment(
     board: Optional[str] = Query(None),
     uploaded_by: Optional[str] = Form(None),
 ):
-    """Store an upload under ``attachments_root(board)/<task_id>/`` with a
-    sanitised, collision-resolved name and record its metadata."""
+    """Store an upload under ``attachments_root(board)/<task_id>/`` with a sanitised,
+    collision-resolved name and record its metadata. ``_safe_attachment_name`` raises
+    ``ValueError`` → 400."""
     with _board_conn(board) as (board, conn), _value_error_400():
         _require_task(conn, task_id)
         safe_name = _safe_attachment_name(file.filename or "")
-
         dest_dir = kanban_db.task_attachments_dir(task_id, board=board)
         dest_dir.mkdir(parents=True, exist_ok=True)
         dest_path = _collision_free_path(dest_dir, safe_name)  # foo.pdf → foo (1).pdf …
@@ -487,10 +431,7 @@ async def upload_task_attachment(
         total = 0
         try:
             with open(dest_path, "wb") as out:
-                while True:
-                    chunk = await file.read(1024 * 1024)
-                    if not chunk:
-                        break
+                while chunk := await file.read(1024 * 1024):
                     total += len(chunk)
                     if total > KANBAN_ATTACHMENT_MAX_BYTES:
                         out.close()
@@ -519,8 +460,7 @@ def download_attachment(attachment_id: int, board: Optional[str] = Query(None)):
         att = kanban_db.get_attachment(conn, attachment_id)
         if att is None:
             raise HTTPException(status_code=404, detail="attachment not found")
-        # Defense in depth against a tampered DB row: the blob must still live
-        # under the board's attachments root.
+        # Defense in depth against a tampered DB row: the blob must still live under the board's attachments root.
         root = kanban_db.attachments_root(board=board).resolve()
         try:
             stored = Path(att.stored_path).resolve()
@@ -529,9 +469,7 @@ def download_attachment(attachment_id: int, board: Optional[str] = Query(None)):
             raise HTTPException(status_code=404, detail="attachment file unavailable")
         if not stored.is_file():
             raise HTTPException(status_code=404, detail="attachment file missing on disk")
-        return FileResponse(
-            path=str(stored), filename=att.filename, media_type=att.content_type or "application/octet-stream",
-        )
+        return FileResponse(path=str(stored), filename=att.filename, media_type=att.content_type or "application/octet-stream")
 
 
 @router.delete("/attachments/{attachment_id}")
@@ -593,13 +531,10 @@ _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 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."""
     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)
@@ -608,24 +543,16 @@ 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. 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_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),
-    ),
-    "scheduled": lambda conn, tid, p: kanban_db.schedule_task(
-        conn, tid, reason=getattr(p, "block_reason", None),
-    ),
+    "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)),
+    "scheduled": lambda conn, tid, p: kanban_db.schedule_task(conn, tid, reason=getattr(p, "block_reason", None)),
     "review": lambda conn, tid, p: kanban_db.request_review(
-        conn, tid, summary=p.summary, metadata=p.metadata,
-        reviewer=(p.assignee or None), force=True,
+        conn, tid, summary=p.summary, metadata=p.metadata, reviewer=(p.assignee or None), force=True,
     ),
     "ready": lambda conn, tid, p: _drag_to(conn, tid, "ready"),
     "todo": lambda conn, tid, p: _drag_to(conn, tid, "todo"),
@@ -648,43 +575,84 @@ def _set_priority(conn, task_id: str, priority: int, board: Optional[str]) -> No
     with kanban_db.write_txn(conn):
         conn.execute("UPDATE tasks SET priority = ? WHERE id = ?", (int(priority), task_id))
         conn.execute(
-            "INSERT INTO task_events (task_id, kind, payload, created_at) "
-            "VALUES (?, 'reprioritized', ?, ?)",
+            "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: this direct-SQL write bypasses every kanban_db mutator,
+    # so report it here — after the txn commits.
     kanban_db.notify_task_updated(conn, task_id, ("priority",), board=board)
 
 
 def _apply_model_override(conn, task_id: str, p) -> bool:
     """Raises ValueError/RuntimeError from kanban_db for the caller to map."""
-    new_model = (None if p.clear_model_override else (p.model_override or "").strip() or None)
+    new_model = None if p.clear_model_override else (p.model_override or "").strip() or None
     return kanban_db.set_model_override(conn, task_id, new_model, provider=p.provider_override)
 
 
 def _apply_reasoning_effort(conn, task_id: str, p) -> bool:
-    new_effort = None if p.clear_reasoning_effort else p.reasoning_effort
-    return kanban_db.set_reasoning_effort(conn, task_id, new_effort)
+    return kanban_db.set_reasoning_effort(conn, task_id, None if p.clear_reasoning_effort else p.reasoning_effort)
 
 
 # Override knobs shared by PATCH and bulk: (payload wants it?, apply, bulk refusal message).
 _OVERRIDE_OPS = (
-    (lambda p: p.clear_model_override or p.model_override is not None,
-     _apply_model_override, "model override refused"),
-    (lambda p: p.clear_reasoning_effort or p.reasoning_effort is not None,
-     _apply_reasoning_effort, "reasoning override refused"),
+    (lambda p: p.clear_model_override or p.model_override is not None, _apply_model_override, "model override refused"),
+    (lambda p: p.clear_reasoning_effort or p.reasoning_effort is not None, _apply_reasoning_effort, "reasoning override refused"),
 )
 
 
+def _patch_status(conn, task_id: str, payload: UpdateTaskBody, review_assignee_deferred: bool) -> None:
+    """PATCH status phase: 400 on a rejected verb, 409 when the transition is refused
+    (naming the blocking parent(s) for ``ready`` so the UI renders an actionable toast)."""
+    s = payload.status
+    if s == "archived":
+        ok = kanban_db.archive_task(conn, task_id)
+    else:
+        try:
+            ok = _apply_status(conn, task_id, s, payload, f"unknown status: {s}")
+        except _StatusRejected as e:
+            raise HTTPException(status_code=400, detail=str(e))
+        if s == "review" and ok and review_assignee_deferred and not payload.assignee:
+            ok = kanban_db.assign_task(conn, task_id, None)
+    if ok:
+        return
+    blockers = _parents_blocking_ready(conn, task_id) if s == "ready" else []
+    if blockers:
+        names = ", ".join(f"{p['title']!r} ({p['id']}, status={p['status']})" for p in blockers)
+        raise HTTPException(status_code=409, detail=f"Cannot move to 'ready': blocked by parent(s) not done — {names}")
+    raise HTTPException(status_code=409, detail=f"status transition to {s!r} not valid from current state")
+
+
+def _patch_title_body(conn, task_id: str, payload: UpdateTaskBody, board: Optional[str]) -> None:
+    """PATCH title/body phase: one UPDATE + ``edited`` event, then the post-commit observer
+    (field names only — values never leave the DB via this payload)."""
+    with kanban_db.write_txn(conn):
+        sets, vals = [], []
+        if payload.title is not None:
+            if not payload.title.strip():
+                raise HTTPException(status_code=400, detail="title cannot be empty")
+            sets.append("title = ?")
+            vals.append(payload.title.strip())
+        if payload.body is not None:
+            sets.append("body = ?")
+            vals.append(payload.body)
+        vals.append(task_id)
+        conn.execute(f"UPDATE tasks SET {', '.join(sets)} WHERE id = ?", vals)
+        conn.execute(
+            "INSERT INTO task_events (task_id, kind, payload, created_at) VALUES (?, 'edited', NULL, ?)",
+            (task_id, int(time.time())),
+        )
+    kanban_db.notify_task_updated(
+        conn, task_id, [f for f in ("title", "body") if getattr(payload, f) is not None], board=board,
+    )
+
+
 @router.patch("/tasks/{task_id}")
 def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Query(None)):
     with _board_conn(board) as (board, conn):
         _require_task(conn, task_id)
-
         # For a combined assignee+review patch, request_review must capture the
         # current implementer before the task is routed to the reviewer.
-        review_assignee_deferred = (payload.status == "review" and payload.assignee is not None)
+        review_assignee_deferred = payload.status == "review" and payload.assignee is not None
         if payload.assignee is not None and not review_assignee_deferred:
             try:
                 ok = kanban_db.assign_task(conn, task_id, payload.assignee or None)
@@ -692,30 +660,8 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
                 raise HTTPException(status_code=409, detail=str(e))
             if not ok:
                 raise HTTPException(status_code=404, detail="task not found")
-
         if payload.status is not None:
-            s = payload.status
-            if s == "archived":
-                ok = kanban_db.archive_task(conn, task_id)
-            else:
-                try:
-                    ok = _apply_status(conn, task_id, s, payload, f"unknown status: {s}")
-                except _StatusRejected as e:
-                    raise HTTPException(status_code=400, detail=str(e))
-                if s == "review" and ok and review_assignee_deferred and not payload.assignee:
-                    ok = kanban_db.assign_task(conn, task_id, None)
-            if not ok:
-                # For ``ready``, name the blocking parent(s) so the dashboard can
-                # render an actionable toast instead of a silent no-op.
-                blockers = _parents_blocking_ready(conn, task_id) if s == "ready" else []
-                if blockers:
-                    names = ", ".join(f"{p['title']!r} ({p['id']}, status={p['status']})" for p in blockers)
-                    raise HTTPException(
-                        status_code=409,
-                        detail=f"Cannot move to 'ready': blocked by parent(s) not done — {names}",
-                    )
-                raise HTTPException(status_code=409, detail=f"status transition to {s!r} not valid from current state")
-
+            _patch_status(conn, task_id, payload, review_assignee_deferred)
         for wanted, apply, _refused in _OVERRIDE_OPS:
             if wanted(payload):
                 try:
@@ -724,36 +670,10 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
                     raise HTTPException(status_code=400, detail=str(e))
                 if not ok:
                     raise HTTPException(status_code=404, detail="task not found")
-
         if payload.priority is not None:
             _set_priority(conn, task_id, payload.priority, board)
-
         if payload.title is not None or payload.body is not None:
-            with kanban_db.write_txn(conn):
-                sets, vals = [], []
-                if payload.title is not None:
-                    if not payload.title.strip():
-                        raise HTTPException(status_code=400, detail="title cannot be empty")
-                    sets.append("title = ?")
-                    vals.append(payload.title.strip())
-                if payload.body is not None:
-                    sets.append("body = ?")
-                    vals.append(payload.body)
-                vals.append(task_id)
-                conn.execute(f"UPDATE tasks SET {', '.join(sets)} WHERE id = ?", vals)
-                conn.execute(
-                    "INSERT INTO task_events (task_id, kind, payload, created_at) "
-                    "VALUES (?, 'edited', NULL, ?)",
-                    (task_id, int(time.time())),
-                )
-            # Post-commit mutation observer; field names only — values never
-            # leave the DB via this payload.
-            kanban_db.notify_task_updated(
-                conn, task_id,
-                [f for f in ("title", "body") if getattr(payload, f) is not None],
-                board=board,
-            )
-
+            _patch_title_body(conn, task_id, payload, board)
         updated = kanban_db.get_task(conn, task_id)
         return {"task": _task_dict(updated) if updated else None}
 
@@ -779,25 +699,22 @@ 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)."""
+    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)."""
     terminations: list[tuple[Optional[int], Optional[str]]] = []
     effective_status = new_status
     with kanban_db.write_txn(conn):
         prev = conn.execute(
-            "SELECT status, current_run_id, worker_pid, claim_lock FROM tasks WHERE id = ?",
-            (task_id,),
+            "SELECT status, current_run_id, worker_pid, claim_lock FROM tasks WHERE id = ?", (task_id,),
         ).fetchone()
         if prev is None:
             return False
-
         if prev["status"] == "running" and new_status == "ready":
             resume_status = kanban_db._retry_status_for_run(conn, task_id, prev["current_run_id"])
             if resume_status == "review":
                 effective_status = "review" if kanban_db._parents_satisfied(conn, task_id) else "todo"
-
         # Never promote to 'ready' unless all parents are done — otherwise the
         # dispatcher spawns a child whose upstream work hasn't completed.
         if effective_status == "ready":
@@ -814,7 +731,6 @@ def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str)
         reopening_satisfied_parent = (
             prev["status"] in {"done", "archived"} and effective_status not in {"done", "archived"}
         )
-
         cur = conn.execute(
             "UPDATE tasks SET status = ?, "
             "  claim_lock = CASE WHEN ? = 'running' THEN claim_lock ELSE NULL END, "
@@ -828,19 +744,13 @@ def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str)
         run_id = None
         if was_running and effective_status != "running" and prev["current_run_id"]:
             run_id = kanban_db._end_run(
-                conn, task_id,
-                outcome="reclaimed", status="reclaimed",
+                conn, task_id, outcome="reclaimed", status="reclaimed",
                 summary=f"status changed to {effective_status} (dashboard/direct)",
             )
             terminations.append((prev["worker_pid"], prev["claim_lock"]))
         conn.execute(
-            "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) "
-            "VALUES (?, ?, 'status', ?, ?)",
-            (
-                task_id, run_id,
-                json.dumps({"status": effective_status, "requested_status": new_status}),
-                int(time.time()),
-            ),
+            "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) VALUES (?, ?, 'status', ?, ?)",
+            (task_id, run_id, json.dumps({"status": effective_status, "requested_status": new_status}), int(time.time())),
         )
         if reopening_satisfied_parent:
             # Domain-layer invalidation composes via a savepoint inside our txn
@@ -885,15 +795,43 @@ def add_link(payload: LinkBody, board: Optional[str] = Query(None)):
 
 
 @router.delete("/links")
-def delete_link(
-    parent_id: str = Query(...),
-    child_id: str = Query(...),
-    board: Optional[str] = Query(None),
-):
+def delete_link(parent_id: str = Query(...), child_id: str = Query(...), board: Optional[str] = Query(None)):
     with _board_conn(board) as (board, conn):
         return {"ok": bool(kanban_db.unlink_tasks(conn, parent_id, child_id))}
 
 
+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``)."""
+    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:
+        s = payload.status
+        if not _apply_status(conn, tid, s, payload, f"unknown status {s!r}"):
+            entry.update(ok=False, error=f"transition to {s!r} refused")
+    if payload.assignee is not None:
+        try:
+            ok = (
+                kanban_db.reassign_task(conn, tid, payload.assignee or None, reclaim_first=True)
+                if payload.reclaim_first
+                else kanban_db.assign_task(conn, tid, payload.assignee or None)
+            )
+            if not ok:
+                entry.update(ok=False, error="assign refused")
+        except RuntimeError as e:
+            entry.update(ok=False, error=str(e))
+    if payload.priority is not None:
+        _set_priority(conn, tid, payload.priority, board)
+    for wanted, apply, refused in _OVERRIDE_OPS:
+        if wanted(payload):
+            try:
+                if not apply(conn, tid, payload):
+                    entry.update(ok=False, error=refused)
+            except (ValueError, RuntimeError) as e:
+                entry.update(ok=False, error=str(e))
+
+
 @router.post("/tasks/bulk")
 def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
     """Apply the same patch to every id. Independent iteration — per-task
@@ -908,41 +846,9 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
             try:
                 if kanban_db.get_task(conn, tid) is None:
                     entry.update(ok=False, error="not found")
-                    results.append(entry)
-                    continue
-                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:
-                    s = payload.status
-                    try:
-                        ok = _apply_status(conn, tid, s, payload, f"unknown status {s!r}")
-                    except _StatusRejected as e:
-                        entry.update(ok=False, error=str(e))
-                        results.append(entry)
-                        continue
-                    if not ok:
-                        entry.update(ok=False, error=f"transition to {s!r} refused")
-                if payload.assignee is not None:
-                    try:
-                        ok = (
-                            kanban_db.reassign_task(conn, tid, payload.assignee or None, reclaim_first=True)
-                            if payload.reclaim_first
-                            else kanban_db.assign_task(conn, tid, payload.assignee or None)
-                        )
-                        if not ok:
-                            entry.update(ok=False, error="assign refused")
-                    except RuntimeError as e:
-                        entry.update(ok=False, error=str(e))
-                if payload.priority is not None:
-                    _set_priority(conn, tid, payload.priority, board)
-                for wanted, apply, refused in _OVERRIDE_OPS:
-                    if wanted(payload):
-                        try:
-                            if not apply(conn, tid, payload):
-                                entry.update(ok=False, error=refused)
-                        except (ValueError, RuntimeError) as e:
-                            entry.update(ok=False, error=str(e))
-            except Exception as e:  # one bad id shouldn't kill the batch
+                else:
+                    _bulk_apply_one(conn, tid, payload, board, entry)
+            except Exception as e:  # one bad id shouldn't kill the batch (incl. _StatusRejected)
                 entry.update(ok=False, error=str(e))
             results.append(entry)
         return {"results": results}
@@ -952,12 +858,12 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
 
 @router.get("/diagnostics")
 def list_diagnostics(
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
+    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."""
+    """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."""
     with _board_conn(board) as (board, conn):
         diags_by_task = _compute_task_diagnostics(conn, task_ids=None)
         if severity and diags_by_task:
@@ -968,24 +874,18 @@ def list_diagnostics(
             }
         if not diags_by_task:
             return {"diagnostics": [], "count": 0}
-
         ids = list(diags_by_task.keys())
-        placeholders = ",".join(["?"] * len(ids))
         rows = {
-            r["id"]: r
-            for r in conn.execute(
-                f"SELECT id, title, status, assignee FROM tasks WHERE id IN ({placeholders})", tuple(ids),
+            r["id"]: r for r in conn.execute(
+                f"SELECT id, title, status, assignee FROM tasks WHERE id IN ({_placeholders(ids)})", tuple(ids),
             ).fetchall()
         }
         out = []
         for tid, dl in diags_by_task.items():
             r = rows.get(tid)
             out.append({
-                "task_id": tid,
-                "task_title": r["title"] if r else None,
-                "task_status": r["status"] if r else None,
-                "task_assignee": r["assignee"] if r else None,
-                "diagnostics": dl,
+                "task_id": tid, "task_title": r["title"] if r else None, "task_status": r["status"] if r else None,
+                "task_assignee": r["assignee"] if r else None, "diagnostics": dl,
             })
         sev_idx = {s: i for i, s in enumerate(kd.SEVERITY_ORDER)}
 
@@ -993,7 +893,6 @@ def list_diagnostics(
             top = row["diagnostics"][0]
             return (-sev_idx.get(top.get("severity"), -1), -(top.get("last_seen_at") or 0))
         out.sort(key=_sort_key)
-
         return {"diagnostics": out, "count": sum(len(d["diagnostics"]) for d in out)}
 
 
@@ -1006,9 +905,7 @@ except ImportError:
 
 
 @router.get("/workers/active")
-def list_active_workers(
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
-):
+def list_active_workers(board: Optional[str] = _BOARD_Q):
     """Every running worker: an open ``task_runs`` row with a ``worker_pid`` whose
     task is ``running``. Returns ``{workers, count, checked_at}``."""
     with _board_conn(board) as (board, conn):
@@ -1025,23 +922,17 @@ def list_active_workers(
 
 
 @router.get("/runs/{run_id}")
-def get_run_endpoint(
-    run_id: int,
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
-):
+def get_run_endpoint(run_id: int, board: Optional[str] = _BOARD_Q):
     """``{run: {...}}`` with the same serialisation as ``GET /tasks/{id}``; 404 if unknown."""
     with _board_conn(board) as (board, conn):
         return {"run": asdict(_require_run(conn, run_id))}
 
 
 @router.get("/runs/{run_id}/inspect")
-def inspect_run_endpoint(
-    run_id: int,
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
-):
-    """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."""
+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."""
     with _board_conn(board) as (board, conn):
         r = _require_run(conn, run_id)
 
@@ -1055,7 +946,6 @@ def inspect_run_endpoint(
     pid = r.worker_pid
     if _psutil is None:
         return _dead("psutil not available", pid=pid)
-
     try:
         proc = _psutil.Process(pid)
         info = proc.as_dict(attrs=["cpu_percent", "memory_info", "num_threads", "status", "create_time", "cmdline"])
@@ -1083,14 +973,10 @@ class TerminateRunBody(BaseModel):
 
 
 @router.post("/runs/{run_id}/terminate")
-def terminate_run_endpoint(
-    run_id: int,
-    payload: TerminateRunBody,
-    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
-):
+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."""
+    SIGTERM->SIGKILL flow, run bookkeeping, and event log match ``POST /tasks/{id}/reclaim``.
+    404 unknown run; 409 already ended / not reclaimable."""
     with _board_conn(board) as (board, conn):
         r = _require_run(conn, run_id)
         if r.ended_at is not None:
@@ -1132,9 +1018,9 @@ 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."""
+    (``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."""
     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))
@@ -1202,36 +1088,28 @@ def estimate_task_endpoint(task_id: str, board: Optional[str] = Query(None)):
     return _run_estimate(task.title, task.body)
 
 
+def _cap(s: Optional[str], n: int) -> str:
+    s = (s or "").strip()
+    return s if len(s) <= n else s[:n] + "…"
+
+
 def _run_estimate(title: str, body: Optional[str]) -> dict:
-    """Never raises — config/parse/API errors become ``{"ok": False, "reason"}``
-    so the UI renders them inline."""
+    """Never raises — config/parse/API errors become ``{"ok": False, "reason"}`` so the UI renders them inline."""
     if not (title or "").strip():
         return {"ok": False, "reason": "a title is required to estimate"}
-
     try:
         from agent.auxiliary_client import call_llm
     except Exception:
         return {"ok": False, "reason": "auxiliary client unavailable"}
-
-    def _cap(s: Optional[str], n: int) -> str:
-        s = (s or "").strip()
-        return s if len(s) <= n else s[:n] + "…"
-
-    user_msg = (f"Title: {_cap(title, 400)}\n\n" f"Description:\n{_cap(body, 4000) or '(none)'}")
+    user_msg = f"Title: {_cap(title, 400)}\n\nDescription:\n{_cap(body, 4000) or '(none)'}"
     try:
         resp = call_llm(
             task="kanban_estimator",
-            messages=[
-                {"role": "system", "content": _ESTIMATE_SYSTEM_PROMPT},
-                {"role": "user", "content": user_msg},
-            ],
-            temperature=0.0,
-            max_tokens=300,
-            timeout=60,
+            messages=[{"role": "system", "content": _ESTIMATE_SYSTEM_PROMPT}, {"role": "user", "content": user_msg}],
+            temperature=0.0, max_tokens=300, timeout=60,
         )
     except Exception as exc:
         return {"ok": False, "reason": f"LLM error: {type(exc).__name__}"}
-
     try:
         raw = (resp.choices[0].message.content or "").strip()
         model = getattr(resp, "model", None)
@@ -1239,7 +1117,6 @@ def _run_estimate(title: str, body: Optional[str]) -> dict:
         raw, model = "", None
 
     # Same tolerant JSON-blob extraction the specifier uses.
-    parsed: Optional[dict] = None
     try:
         blob = raw
         if not blob.lstrip().startswith("{"):
@@ -1251,7 +1128,6 @@ def _run_estimate(title: str, body: Optional[str]) -> dict:
         parsed = None
     if not parsed:
         return {"ok": False, "reason": "could not parse an estimate from the model"}
-
     try:
         est_tokens = int(parsed.get("est_tokens") or 0)
     except (TypeError, ValueError):
@@ -1260,7 +1136,6 @@ def _run_estimate(title: str, body: Optional[str]) -> dict:
     if complexity not in {"S", "M", "L"}:
         complexity = None
     rationale = str(parsed.get("rationale") or "").strip() or None
-
     return {"ok": True, "est_tokens": est_tokens, "complexity": complexity, "rationale": rationale, "model": model}
 
 
@@ -1287,13 +1162,13 @@ 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 existing gateway
+# notifier delivers completed/blocked/gave_up with no extra plumbing.
 
 def _configured_home_channels() -> list[dict]:
-    """Every platform with a home_channel, from the live GatewayConfig (so env
-    overlays like ``TELEGRAM_HOME_CHANNEL`` are honored), sorted by platform."""
+    """Every platform with a home_channel, from the live GatewayConfig (so env overlays
+    like ``TELEGRAM_HOME_CHANNEL`` are honored), sorted by platform."""
     try:
         from gateway.config import load_gateway_config
         gw_cfg = load_gateway_config()
@@ -1339,8 +1214,7 @@ def get_home_channels(task_id: Optional[str] = Query(None), board: Optional[str]
         with _board_conn(board) as (board, conn):
             subs = kanban_db.list_notify_subs(conn, task_id)
         subscribed_homes = {
-            (str(sub.get("platform") or ""), str(sub.get("chat_id") or ""), str(sub.get("thread_id") or ""))
-            for sub in subs
+            (str(sub.get("platform") or ""), str(sub.get("chat_id") or ""), str(sub.get("thread_id") or "")) for sub in subs
         }
     return {
         "home_channels": [
@@ -1398,11 +1272,7 @@ def get_assignees(board: Optional[str] = Query(None)):
 
 
 @router.get("/tasks/{task_id}/log")
-def get_task_log(
-    task_id: str,
-    tail: Optional[int] = Query(None, ge=1, le=2_000_000),
-    board: Optional[str] = Query(None),
-):
+def get_task_log(task_id: str, tail: Optional[int] = Query(None, ge=1, le=2_000_000), board: Optional[str] = Query(None)):
     """Worker stdout/stderr log. ``tail`` caps the response bytes; 404 if the
     task never spawned. On-disk log rotates at 2 MiB with one ``.log.1`` kept."""
     with _board_conn(board) as (board, conn):
@@ -1417,11 +1287,7 @@ def get_task_log(
 
 
 @router.post("/dispatch")
-def dispatch(
-    dry_run: bool = Query(False),
-    max_n: int = Query(8, alias="max"),
-    board: Optional[str] = Query(None),
-):
+def dispatch(dry_run: bool = Query(False), max_n: int = Query(8, alias="max"), board: Optional[str] = Query(None)):
     """Dispatch nudge so the UI doesn't wait out the 60 s dispatcher tick."""
     with _board_conn(board) as (board, conn):
         result = kanban_db.dispatch_once(conn, dry_run=dry_run, max_spawn=max_n, board=board)
@@ -1433,11 +1299,10 @@ def dispatch(
 
 @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 and custom-provider probes: a slow/offline local endpoint
-    must not hang the drawer."""
+    """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
+    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
 
@@ -1446,19 +1311,14 @@ def model_options():
         )
         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:
         log.exception("kanban model-options failed")
-        # Empty catalog → the UI falls back to a free-text input.
-        return {"providers": []}
+        return {"providers": []}  # empty catalog → the UI falls back to a free-text input
 
 
 # --- Boards CRUD (multi-project support) --------------------------------------
@@ -1536,12 +1396,9 @@ def _board_counts(slug: str) -> dict[str, int]:
     try:
         if not kanban_db.kanban_db_path(board=slug).exists():
             return {}
-        conn = kanban_db.connect(board=slug)
-        try:
+        with closing(kanban_db.connect(board=slug)) as conn:
             rows = conn.execute("SELECT status, COUNT(*) AS n FROM tasks GROUP BY status").fetchall()
             return {r["status"]: int(r["n"]) for r in rows}
-        finally:
-            conn.close()
     except Exception:
         return {}
 
@@ -1590,8 +1447,8 @@ def list_boards(include_archived: bool = Query(False)):
     for b in boards:
         b["is_current"] = (b["slug"] == current)
         b["counts"] = _board_counts(b["slug"])
-        # Live cards only — archived tasks are hidden from every default board
-        # view, so counting them in the switcher badge would visibly disagree.
+        # Live cards only — archived tasks are hidden from every default board view,
+        # so counting them in the switcher badge would visibly disagree.
         b["total"] = sum(n for status, n in b["counts"].items() if status != "archived")
         b["default_workspace_kind"] = _default_workspace_kind(b)
         pid = b.get("project_id") or None
@@ -1615,17 +1472,13 @@ def _validate_workdir(raw: str) -> str:
 def create_board_endpoint(payload: CreateBoardBody):
     """Create a board. Idempotent — ``slug`` collision returns the existing one."""
     default_workdir = _validate_workdir(payload.default_workdir) if payload.default_workdir else None
-    # A chosen project's primary repo becomes the default workdir unless one was
-    # passed explicitly.
+    # A chosen project's primary repo becomes the default workdir unless one was passed explicitly.
     project_id, _pname, primary_path = _resolve_project(payload.project_id)
     if primary_path and not default_workdir:
         default_workdir = primary_path
     with _value_error_400():
         meta = kanban_db.create_board(
-            payload.slug,
-            default_workdir=default_workdir,
-            project_id=project_id,
-            **_board_display_kwargs(payload),
+            payload.slug, default_workdir=default_workdir, project_id=project_id, **_board_display_kwargs(payload),
         )
     if payload.switch:
         with _value_error_400():
@@ -1642,8 +1495,7 @@ def rename_board(slug: str, payload: RenameBoardBody):
     if payload.default_workdir is not None:
         raw = payload.default_workdir.strip()
         default_workdir = _validate_workdir(raw) if raw else ""
-    # A resolved project mirrors its repo into default_workdir unless the caller
-    # set default_workdir explicitly.
+    # A resolved project mirrors its repo into default_workdir unless the caller set it explicitly.
     project_id: Optional[str] = None
     if payload.project_id is not None:
         if payload.project_id.strip():
@@ -1653,10 +1505,7 @@ def rename_board(slug: str, payload: RenameBoardBody):
         else:
             project_id = ""  # clear the scope
     meta = kanban_db.write_board_metadata(
-        normed,
-        default_workdir=default_workdir,
-        project_id=project_id,
-        **_board_display_kwargs(payload),
+        normed, default_workdir=default_workdir, project_id=project_id, **_board_display_kwargs(payload),
     )
     return {"board": _annotate_board_meta(meta)}
 
@@ -1696,11 +1545,8 @@ async def export_board_endpoint(slug: str, body: ExportBoardBody):
         except OSError as exc:
             raise HTTPException(status_code=500, detail=f"Could not create export directory: {exc}")
         output = str(staging / f"{slug}-{time.strftime('%Y%m%d-%H%M%S')}.tar.gz")
-
     return await _run_transfer(
-        lambda: kanban_transfer.export_board(
-            slug, output, include_attachments=body.attachments, include_logs=body.logs,
-        ),
+        lambda: kanban_transfer.export_board(slug, output, include_attachments=body.attachments, include_logs=body.logs),
         f"POST /boards/{slug}/export",
     )
 
@@ -1713,7 +1559,6 @@ async def import_board_endpoint(body: ImportBoardBody):
     archive = (body.archive or "").strip()
     if not archive:
         raise HTTPException(status_code=400, detail="archive path is required")
-
     result = await _run_transfer(
         lambda: kanban_transfer.import_board(archive, (body.slug or "").strip() or None, activate=body.switch),
         "POST /boards/import",
@@ -1781,16 +1626,13 @@ 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 <name> --auto``).
-    Non-OK outcomes are NOT HTTP errors — the UI renders the reason inline."""
+    """Generate a description via ``auxiliary.profile_describer`` and persist it with
+    ``description_auto: true`` (``hermes profile describe <name> --auto``). 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))
-    return {
-        "ok": bool(outcome.ok), "profile": outcome.profile_name,
-        "reason": outcome.reason, "description": outcome.description,
-    }
+    return {"ok": bool(outcome.ok), "profile": outcome.profile_name, "reason": outcome.reason, "description": outcome.description}
 
 
 # --- Decompose (built-in decomposer fan-out) ----------------------------------
@@ -1800,15 +1642,11 @@ 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."""
+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."""
     def _run():
         from hermes_cli import kanban_decompose
         return kanban_decompose.decompose_task(task_id, author=(payload.author or None))
@@ -1830,6 +1668,9 @@ class OrchestrationSettingsBody(BaseModel):
     auto_promote_children: Optional[bool] = None
 
 
+_PROFILE_SETTINGS = ("orchestrator_profile", "default_assignee")
+
+
 @router.get("/orchestration")
 def get_orchestration_settings():
     """Current orchestration knobs from config.yaml plus the resolved effective
@@ -1838,7 +1679,6 @@ def get_orchestration_settings():
     kanban_cfg = (cfg.get("kanban") or {}) if isinstance(cfg, dict) else {}
     explicit_orch = (kanban_cfg.get("orchestrator_profile") or "").strip()
     explicit_default = (kanban_cfg.get("default_assignee") or "").strip()
-
     resolved_orch = explicit_orch
     resolved_default = explicit_default
     try:
@@ -1852,7 +1692,6 @@ def get_orchestration_settings():
         active_default = "default"
         resolved_orch = resolved_orch or active_default
         resolved_default = resolved_default or active_default
-
     return {
         "orchestrator_profile": explicit_orch,
         "default_assignee": explicit_default,
@@ -1865,8 +1704,7 @@ def get_orchestration_settings():
 
 
 def _validated_profile_name(raw: Optional[str], profiles_mod) -> str:
-    """Strip a profile name; 400 if non-empty and unknown. Fails open when the
-    lookup itself errors."""
+    """Strip a profile name; 400 if non-empty and unknown. Fails open when the lookup itself errors."""
     name = (raw or "").strip()
     if name and profiles_mod is not None:
         try:
@@ -1886,29 +1724,18 @@ def set_orchestration_settings(payload: OrchestrationSettingsBody):
     with _errors_to_500("failed to load config"):
         from hermes_cli.config import load_config, save_config
         cfg = load_config() or {}
-
     kanban_section = cfg.setdefault("kanban", {})
     if not isinstance(kanban_section, dict):
-        kanban_section = {}
-        cfg["kanban"] = kanban_section
-
+        kanban_section = cfg["kanban"] = {}
     try:
         from hermes_cli import profiles as profiles_mod
     except Exception:
         profiles_mod = None  # type: ignore
-
-    if payload.orchestrator_profile is not None:
-        kanban_section["orchestrator_profile"] = _validated_profile_name(payload.orchestrator_profile, profiles_mod)
-    if payload.default_assignee is not None:
-        kanban_section["default_assignee"] = _validated_profile_name(payload.default_assignee, profiles_mod)
-    if payload.auto_decompose is not None:
-        kanban_section["auto_decompose"] = bool(payload.auto_decompose)
-    if payload.auto_promote_children is not None:
-        kanban_section["auto_promote_children"] = bool(payload.auto_promote_children)
-
+    # Field order == write order (profiles validated first, then the booleans).
+    for key, value in payload.model_dump(exclude_none=True).items():
+        kanban_section[key] = _validated_profile_name(value, profiles_mod) if key in _PROFILE_SETTINGS else bool(value)
     with _errors_to_500("failed to save config"):
         save_config(cfg)
-
     return get_orchestration_settings()  # callers re-render from the resolved state
 
 
@@ -1919,6 +1746,13 @@ def set_orchestration_settings(payload: OrchestrationSettingsBody):
 _EVENT_POLL_SECONDS = 0.3
 
 
+def _int_param(ws: WebSocket, name: str) -> int:
+    try:
+        return int(ws.query_params.get(name, "0"))
+    except ValueError:
+        return 0
+
+
 @router.websocket("/events")
 async def stream_events(ws: WebSocket):
     if not _ws_upgrade_authorized(ws):
@@ -1939,11 +1773,7 @@ async def stream_events(ws: WebSocket):
             event_conn = None
 
     try:
-        try:
-            cursor = int(ws.query_params.get("since", "0"))
-        except ValueError:
-            cursor = 0
-
+        cursor = _int_param(ws, "since")
         # Board is pinned at the handshake; the UI opens a new WS on board change
         # rather than reconciling two cursors mid-stream.
         ws_board_raw = ws.query_params.get("board")
@@ -1962,28 +1792,24 @@ async def stream_events(ws: WebSocket):
                 (cursor_val,),
             ).fetchall()
             out: list[dict] = []
-            new_cursor = cursor_val
             for r in rows:
                 try:
                     payload = json.loads(r["payload"]) if r["payload"] else None
                 except Exception:
                     payload = None
                 out.append({**dict(r), "payload": payload})
-                new_cursor = r["id"]
-            return new_cursor, out
+            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.
+            # 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.
             try:
                 msg = await asyncio.wait_for(ws.receive(), timeout=_EVENT_POLL_SECONDS)
                 if msg["type"] == "websocket.disconnect":
                     return
-                # Other client messages (pong, text) are ignored.
             except asyncio.TimeoutError:
                 pass  # no client message — poll the DB
-
             if event_executor is None:
                 event_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="kanban-events")
             cursor, events = await asyncio.get_running_loop().run_in_executor(event_executor, _fetch_new, cursor)
@@ -1992,9 +1818,8 @@ 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 an application traceback.
+        # 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
     except Exception as exc:  # never crash the dashboard worker
         log.warning("Kanban event stream error: %s", exc)