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

refactor(agent/error_surface,i18n): fold layer resolution into helper, collapse guard ladders (strings untouched)

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:34:22 -07:00
parent d4781581ee
commit 282688e37e
2 changed files with 40 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
agent/error_surface.py
View File

@@ -2,16 +2,11 @@
Maps the internal failure taxonomy (``FailoverReason`` values carried in turn
results as ``failure_reason``, or raw exceptions from the turn dispatcher)
onto a small, stable wire descriptor::
{"layer": <ui layer>, "code": <specific code>, "retryable": <bool>}
Layers (wire values): provider (model API rejected/failed the call), endpoint
(user-configured custom/local endpoint transport failure), streaming (SSE
dropped mid-turn), auth, billing (fallback signal; clients usually have a
richer ``billing_block``), gateway (local runtime errored), disk (disk full /
persistence failure).
onto a small, stable wire descriptor ``{"layer", "code", "retryable"}``.
Layers: provider (model API rejected/failed), endpoint (user-configured
custom/local endpoint transport failure), streaming (SSE dropped mid-turn),
auth, billing (fallback signal; clients usually have a richer
``billing_block``), gateway (local runtime errored), disk (disk full).
Dependency-light and NEVER raises: surfacing diagnostics must not break the
error path it describes. Descriptors are advisory — clients fall back to
string sniffing when absent or partial.
@@ -35,10 +30,7 @@ LAYER_DISK = "disk"
# failure_reason → UI layer. Unlisted reasons fall back to LAYER_PROVIDER:
# every FailoverReason comes from classifying a provider call.
_REASON_TO_LAYER = {
"auth": LAYER_AUTH,
"auth_permanent": LAYER_AUTH,
"billing": LAYER_BILLING,
"billing_unverified": LAYER_BILLING,
"auth": LAYER_AUTH, "auth_permanent": LAYER_AUTH, "billing": LAYER_BILLING, "billing_unverified": LAYER_BILLING,
}
# Failures between us and the base_url (not a provider verdict); on a
@@ -102,6 +94,15 @@ def _disk_full(candidate: Any) -> bool:
return False
def _result_layer(reason: str, error_text: str, provider: str) -> str:
layer = _REASON_TO_LAYER.get(reason)
if layer is not None:
return layer
if reason in _TRANSPORT_REASONS and _is_custom_endpoint(provider):
return LAYER_ENDPOINT
return LAYER_STREAMING if _looks_like_stream_drop(error_text) else LAYER_PROVIDER
def build_error_surface_from_result(result: Any, provider: str = "", model: str = "") -> Optional[dict]:
"""Descriptor for a returned-error turn result (``failed=True`` dicts).
@@ -115,35 +116,23 @@ def build_error_surface_from_result(result: Any, provider: str = "", model: str
reason = str(result.get("failure_reason") or "").strip()
if not error_text and not reason:
return None
# Disk-full wins outright: the fix (free space) is unrelated to the
# provider stack; hermes_state owns the pattern list.
if error_text and _disk_full(error_text):
return _surface(LAYER_DISK, "disk_full", False, provider, model)
if result.get("billing_block") or reason in ("billing", "billing_unverified"):
return _surface(LAYER_BILLING, reason or "billing", False, provider, model)
if not reason:
# Failed result without a classified reason (legacy paths).
if _looks_like_stream_drop(error_text):
return _surface(LAYER_STREAMING, "stream_drop", True, provider, model)
return _surface(LAYER_PROVIDER, "unknown", True, provider, model)
layer = _REASON_TO_LAYER.get(reason)
if layer is None:
if reason in _TRANSPORT_REASONS and _is_custom_endpoint(provider):
layer = LAYER_ENDPOINT
elif _looks_like_stream_drop(error_text):
layer = LAYER_STREAMING
else:
layer = LAYER_PROVIDER
# Prefer the classifier's own verdict (``failure_retryable``); the
# reason-set fallback covers older results.
retryable = result.get("failure_retryable")
if not isinstance(retryable, bool):
retryable = reason not in _NON_RETRYABLE_REASONS
return _surface(layer, reason, retryable, provider, model)
return _surface(_result_layer(reason, error_text, provider), reason, retryable, provider, model)
except Exception: # pragma: no cover — never break the error path
logger.debug("error_surface: result classification failed", exc_info=True)
return None
@@ -157,13 +146,9 @@ def build_error_surface_from_exception(exc: BaseException, provider: str = "", m
"""
try:
message = str(exc) or type(exc).__name__
if _disk_full(exc):
return _surface(LAYER_DISK, "disk_full", False, provider, model)
exc_module = type(exc).__module__ or ""
api_like = exc_module.split(".")[0] in _API_EXC_MODULE_PREFIXES or hasattr(exc, "status_code")
api_like = (type(exc).__module__ or "").split(".")[0] in _API_EXC_MODULE_PREFIXES or hasattr(exc, "status_code")
if not api_like or not isinstance(exc, Exception):
return _surface(LAYER_GATEWAY, type(exc).__name__, True, provider, model)

70
agent/i18n.py
View File

@@ -66,27 +66,20 @@ def _locales_dir() -> Path:
return candidate
logger.warning(
"HERMES_BUNDLED_LOCALES points to a non-directory path (%s); "
"falling back to bundled/source locale resolution",
override,
"falling back to bundled/source locale resolution", override,
)
return Path(__file__).resolve().parent.parent / "locales"
def _normalize_lang(value: Any) -> str:
"""Map a user-supplied value (code, alias, or regional tag like ``zh-CN``) to a supported code, else default."""
if not isinstance(value, str):
return DEFAULT_LANGUAGE
key = value.strip().lower()
if not key:
return DEFAULT_LANGUAGE
key = value.strip().lower() if isinstance(value, str) else ""
if key in SUPPORTED_LANGUAGES:
return key
if key in _LANGUAGE_ALIASES:
return _LANGUAGE_ALIASES[key]
base = key.split("-", 1)[0] # strip region suffix
if base in SUPPORTED_LANGUAGES:
return base
return DEFAULT_LANGUAGE
return base if base in SUPPORTED_LANGUAGES else DEFAULT_LANGUAGE
def _cache_catalog(lang: str, flat: dict[str, str]) -> dict[str, str]:
@@ -103,20 +96,18 @@ def _load_catalog(lang: str) -> dict[str, str]:
return cached
path = _locales_dir() / f"{lang}.yaml"
flat: dict[str, str] = {}
if not path.is_file():
logger.debug("i18n catalog missing for %s at %s", lang, path)
return _cache_catalog(lang, {})
try:
import yaml
with path.open("r", encoding="utf-8") as f:
raw = yaml.safe_load(f) or {}
except Exception as exc:
logger.warning("Failed to load i18n catalog %s: %s", path, exc)
return _cache_catalog(lang, {})
flat: dict[str, str] = {}
_flatten_into(raw, "", flat)
else:
try:
import yaml
with path.open("r", encoding="utf-8") as f:
raw = yaml.safe_load(f) or {}
except Exception as exc:
logger.warning("Failed to load i18n catalog %s: %s", path, exc)
else:
_flatten_into(raw, "", flat)
return _cache_catalog(lang, flat)
@@ -124,8 +115,7 @@ def _flatten_into(node: Any, prefix: str, out: dict[str, str]) -> None:
# Non-string, non-dict leaves are ignored -- catalogs are text-only.
if isinstance(node, dict):
for key, value in node.items():
child_key = f"{prefix}.{key}" if prefix else str(key)
_flatten_into(value, child_key, out)
_flatten_into(value, f"{prefix}.{key}" if prefix else str(key), out)
elif isinstance(node, str):
out[prefix] = node
@@ -154,9 +144,7 @@ def reset_language_cache() -> None:
def get_language() -> str:
"""Resolve the active language using env > config > default order."""
env_lang = os.environ.get("HERMES_LANGUAGE")
if env_lang:
return _normalize_lang(env_lang)
return _config_language_cached() or DEFAULT_LANGUAGE
return _normalize_lang(env_lang) if env_lang else _config_language_cached() or DEFAULT_LANGUAGE
def t(key: str, lang: str | None = None, **format_kwargs: Any) -> str:
@@ -167,30 +155,18 @@ def t(key: str, lang: str | None = None, **format_kwargs: Any) -> str:
"""
target = _normalize_lang(lang) if lang else get_language()
value = _load_catalog(target).get(key)
if value is None and target != DEFAULT_LANGUAGE:
value = _load_catalog(DEFAULT_LANGUAGE).get(key)
if value is None:
logger.debug("i18n miss: key=%r lang=%r", key, target)
value = key
if format_kwargs:
try:
return value.format(**format_kwargs)
except (KeyError, IndexError, ValueError) as exc:
logger.warning(
"i18n format failed for key=%r lang=%r kwargs=%r: %s",
key, target, format_kwargs, exc,
)
return value
return value
if not format_kwargs:
return value
try:
return value.format(**format_kwargs)
except (KeyError, IndexError, ValueError) as exc:
logger.warning("i18n format failed for key=%r lang=%r kwargs=%r: %s", key, target, format_kwargs, exc)
return value
__all__ = [
"SUPPORTED_LANGUAGES",
"DEFAULT_LANGUAGE",
"t",
"get_language",
"reset_language_cache",
]
__all__ = ["SUPPORTED_LANGUAGES", "DEFAULT_LANGUAGE", "t", "get_language", "reset_language_cache"]