Merge branch 'simp/r3-26-F' into simp/integration3

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