refactor(agent): one Retry-After parser and one reset-grammar table feed every retry wait

Seven sites hand-rolled `float(headers.get("Retry-After"))` (anon_auth,
shared_metrics_sender, gemini_native_adapter, extract_api_error_context,
nous_rate_guard, skills_hub_github, skills_hub_clawhub x2) and silently
dropped RFC 7231 HTTP-date values that the conversation loop already honours
via agent/retry_utils.py::parse_retry_after_seconds. They now call it; per-site
caps/floors stay at the call site.

The free-text "resets in / quotaResetDelay / retry after N s" regexes lived in
two tables (agent_runtime_helpers vs credential_pool) whose "resets in"
grammars diverged: the pool accepted only integer `Nhr Nmin` while the error
context accepted h/hr/hours + m/min/minutes + s/seconds with decimals. One table
(agent/retry_utils.py::RETRY_DELAY_PATTERNS / reset_delay_from_message) using
the wider grammar, so a pooled credential's cooldown and the UI's reset time
now agree.

agent/agent_runtime_helpers.py

@@ -26,6 +26,7 @@ from agent.credential_pool import (
     STATUS_EXHAUSTED, credential_pool_matches_provider, resolve_runtime_pool_key
 )
 from agent.error_classifier import FailoverReason
+from agent.retry_utils import parse_retry_after_seconds, reset_delay_from_message
 from agent.turn_context import drop_stale_api_content
 from utils import base_url_host_matches, base_url_hostname, env_var_enabled, atomic_json_write
 logger = logging.getLogger(__name__)
@@ -3104,34 +3105,12 @@ def cleanup_dead_connections(agent) -> bool:
     return False
 
 
-_QUOTA_RESET_DELAY_RE = re.compile(r"quotaResetDelay[:\s\"]+(\d+(?:\.\d+)?)(ms|s)", re.IGNORECASE)
-_RESETS_IN_RE = re.compile(
-    r"resets?\s+in\s+"
-    r"(?:(\d+(?:\.\d+)?)\s*(?:h|hr|hrs|hour|hours)\b\s*)?"
-    r"(?:(\d+(?:\.\d+)?)\s*(?:m|min|mins|minute|minutes)\b\s*)?"
-    r"(?:(\d+(?:\.\d+)?)\s*(?:s|sec|secs|second|seconds)\b)?", re.IGNORECASE,
-)
-_RETRY_AFTER_SECONDS_RE = re.compile(r"retry\s+(?:after\s+)?(\d+(?:\.\d+)?)\s*(?:sec|secs|seconds|s\b)", re.IGNORECASE)
-
-
-def _reset_delay_from_message(message: str) -> Optional[float]:
-    """Seconds-until-reset parsed from free-text provider messages, or None."""
-    m = _QUOTA_RESET_DELAY_RE.search(message)
-    if m:
-        value = float(m.group(1))
-        return value / 1000.0 if m.group(2).lower() == "ms" else value
-    m = _RESETS_IN_RE.search(message)
-    if m and any(m.groups()):
-        return float(m.group(1) or 0) * 3600 + float(m.group(2) or 0) * 60 + float(m.group(3) or 0)
-    m = _RETRY_AFTER_SECONDS_RE.search(message)
-    return float(m.group(1)) if m else None
-
-
 def _set_reset_from_retry_after(context: Dict[str, Any], retry_after: Any) -> None:
-    if retry_after in {None, ""} or "reset_at" in context:
+    if "reset_at" in context:
         return
-    with contextlib.suppress(TypeError, ValueError):
-        context["reset_at"] = time.time() + float(retry_after)
+    seconds = parse_retry_after_seconds(retry_after)
+    if seconds is not None:
+        context["reset_at"] = time.time() + seconds
 
 
 def extract_api_error_context(error: Exception) -> Dict[str, Any]:
@@ -3155,14 +3134,14 @@ def extract_api_error_context(error: Exception) -> Dict[str, Any]:
         _set_reset_from_retry_after(context, payload.get("retry_after"))
     headers = getattr(getattr(error, "response", None), "headers", None)
     if headers:
-        _set_reset_from_retry_after(context, headers.get("retry-after") or headers.get("Retry-After") or None)
+        _set_reset_from_retry_after(context, headers)
         ratelimit_reset = headers.get("x-ratelimit-reset")
         if ratelimit_reset and "reset_at" not in context:
             context["reset_at"] = ratelimit_reset
     if "message" not in context and str(error).strip():
         context["message"] = str(error).strip()[:500]
     if "reset_at" not in context and isinstance(context.get("message") or "", str):
-        delay = _reset_delay_from_message(context.get("message") or "")
+        delay = reset_delay_from_message(context.get("message") or "")
         if delay is not None:
             context["reset_at"] = time.time() + delay
     return context

agent/credential_pool.py

@@ -19,6 +19,7 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple
 from hermes_constants import OPENROUTER_BASE_URL
 from hermes_cli.config import load_env
 from agent.secret_scope import get_secret as _get_secret
+from agent.retry_utils import reset_delay_from_message
 from agent.credential_persistence import (
     fingerprint_secret_value,
     is_borrowed_credential_source,
@@ -367,36 +368,6 @@ def _parse_absolute_timestamp(value: Any) -> Optional[float]:
     return None
 
 
-# (regex, seconds-from-match) pairs tried in order against provider error text.
-_RETRY_DELAY_PATTERNS: Tuple[Tuple[re.Pattern, Callable[[re.Match], float]], ...] = (
-    (
-        re.compile(r"quotaResetDelay[:\s\"]+(\d+(?:\.\d+)?)(ms|s)", re.IGNORECASE),
-        lambda m: float(m.group(1)) / 1000.0 if m.group(2).lower() == "ms" else float(m.group(1)),
-    ),
-    (
-        re.compile(r"retry\s+(?:after\s+)?(\d+(?:\.\d+)?)\s*(?:sec|secs|seconds|s\b)", re.IGNORECASE),
-        lambda m: float(m.group(1)),
-    ),
-    # "Resets in 4hr 5min" format used by OpenCode Go weekly usage limits
-    (
-        re.compile(r"resets?\s+in\s+(\d+)\s*hr\s+(\d+)\s*min", re.IGNORECASE),
-        lambda m: int(m.group(1)) * 3600 + int(m.group(2)) * 60,
-    ),
-    (re.compile(r"resets?\s+in\s+(\d+)\s*hr\b", re.IGNORECASE), lambda m: int(m.group(1)) * 3600),
-    (re.compile(r"resets?\s+in\s+(\d+)\s*min\b", re.IGNORECASE), lambda m: int(m.group(1)) * 60),
-)
-
-
-def _extract_retry_delay_seconds(message: str) -> Optional[float]:
-    if not message:
-        return None
-    for pattern, to_seconds in _RETRY_DELAY_PATTERNS:
-        match = pattern.search(message)
-        if match:
-            return to_seconds(match)
-    return None
-
-
 def _normalize_error_context(error_context: Optional[Dict[str, Any]]) -> Dict[str, Any]:
     if not isinstance(error_context, dict):
         return {}
@@ -413,7 +384,7 @@ def _normalize_error_context(error_context: Optional[Dict[str, Any]]) -> Dict[st
     parsed_reset_at = _parse_absolute_timestamp(reset_at)
     message = error_context.get("message")
     if parsed_reset_at is None and isinstance(message, str):
-        retry_delay_seconds = _extract_retry_delay_seconds(message)
+        retry_delay_seconds = reset_delay_from_message(message)
         if retry_delay_seconds is not None:
             parsed_reset_at = time.time() + retry_delay_seconds
     if parsed_reset_at is not None:

agent/gemini_native_adapter.py

@@ -20,6 +20,7 @@ from typing import Any, Dict, Iterator, List, Optional
 import httpx
 
 from agent.bounded_response import read_streaming_error_body
+from agent.retry_utils import parse_retry_after_seconds
 from agent.gemini_schema import sanitize_gemini_tool_parameters
 
 logger = logging.getLogger(__name__)
@@ -608,10 +609,7 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
     err_obj = _error_object(body_text)
     err_status, err_message = (str(err_obj.get(k) or "").strip() for k in ("status", "message"))
     reason, metadata = _error_info(err_obj)
-    try:
-        retry_after: Optional[float] = float(response.headers.get("Retry-After") or response.headers.get("retry-after"))
-    except (TypeError, ValueError):
-        retry_after = None
+    retry_after = parse_retry_after_seconds(response.headers)
     message = (
         f"Gemini HTTP {status} ({err_status or 'error'}): {err_message}" if err_message
         else f"Gemini returned HTTP {status}: {body_text[:500]}"

agent/nous_rate_guard.py

@@ -15,6 +15,7 @@ import os
 import time
 from typing import Any, Mapping, Optional
 from utils import atomic_write_text
+from agent.retry_utils import parse_retry_after_seconds
 from agent.rate_limit_tracker import (
     _BUCKET_TAGS, _fmt_seconds, _safe_float, _safe_int, has_rate_limit_headers, lower_headers,
 )
@@ -41,11 +42,12 @@ def _state_path() -> str:
 def _parse_reset_seconds(headers: Optional[Mapping[str, str]]) -> Optional[float]:
     """Best reset estimate (seconds from now) from hourly, per-minute, then retry-after headers."""
     lowered = lower_headers(headers)
-    for key in ("x-ratelimit-reset-requests-1h", "x-ratelimit-reset-requests", "retry-after"):
+    for key in ("x-ratelimit-reset-requests-1h", "x-ratelimit-reset-requests"):
         val = _safe_float(lowered.get(key), 0.0)
         if val > 0:
             return val
-    return None
+    retry_after = parse_retry_after_seconds(lowered.get("retry-after"))
+    return retry_after if retry_after else None
 
 
 def record_nous_rate_limit(

agent/proxy_bypass.py

@@ -0,0 +1,94 @@
+"""NO_PROXY matching shared by the LLM transport (``agent/process_bootstrap.py``) and the
+gateway platform adapters (``gateway/platforms/base.py``).
+
+One matcher so "is this host in NO_PROXY" has one answer everywhere: exact hosts, domain
+suffixes (``example.com``, ``.example.com``, ``*.example.com``), IP literals, CIDR ranges,
+optional ``host:port`` entries and ``*``. The stdlib ``proxy_bypass_environment`` understands
+none of the CIDR / ``*.`` forms, which is why the LLM path used to route ``10.x`` endpoints
+through the corporate proxy while Telegram/Discord bypassed it.
+
+Leaf module: stdlib only, importable during early boot.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import os
+import re
+from urllib.parse import urlsplit
+
+PROXY_ENV_KEYS = ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy")
+
+
+def first_proxy_env_value() -> str:
+    """First non-empty HTTPS_PROXY / HTTP_PROXY / ALL_PROXY value (any case), or ''."""
+    return next((v for k in PROXY_ENV_KEYS if (v := (os.environ.get(k) or "").strip())), "")
+
+
+def split_host_port(value: str) -> tuple[str, int | None]:
+    """``(host, port)`` from a URL, ``[v6]:port``, ``host:port`` or bare host; host lowercased."""
+    raw = str(value or "").strip()
+    if not raw:
+        return "", None
+    if "://" in raw:
+        parsed = urlsplit(raw)
+        host, port = parsed.hostname or "", parsed.port
+    elif raw.startswith("[") and "]" in raw:
+        host, _, rest = raw[1:].partition("]")
+        port = int(rest[1:]) if rest.startswith(":") and rest[1:].isdigit() else None
+    elif raw.count(":") == 1 and raw.rpartition(":")[2].isdigit():
+        host, _, port_s = raw.rpartition(":")
+        port = int(port_s)
+    else:
+        host, port = raw.strip("[]"), None
+    return host.lower().rstrip("."), port
+
+
+def no_proxy_entries(no_proxy_value: str | None = None) -> list[str]:
+    """Comma/whitespace-separated NO_PROXY entries; from the environment (both casings) when
+    ``no_proxy_value`` is None."""
+    if no_proxy_value is None:
+        no_proxy_value = ",".join(os.environ.get(key, "") for key in ("NO_PROXY", "no_proxy"))
+    return [part for part in re.split(r"[\s,]+", no_proxy_value.strip()) if part]
+
+
+def _ip_or_none(value: str, parse=ipaddress.ip_address):
+    """``parse(value)`` or None on ``ValueError`` (``parse`` is ip_address / ip_network)."""
+    try:
+        return parse(value)
+    except ValueError:
+        return None
+
+
+def no_proxy_entry_matches(entry: str, host: str, port: int | None = None) -> bool:
+    token = str(entry or "").strip().lower()
+    if not token:
+        return False
+    if token == "*":
+        return True
+    token_host, token_port = split_host_port(token)
+    if not token_host or (token_port is not None and (port is None or token_port != port)):
+        return False
+    host_ip = _ip_or_none(host)
+    network = _ip_or_none(token_host, lambda v: ipaddress.ip_network(v, strict=False))
+    if network is not None:  # CIDR or bare IP literal (a /32 / /128 network)
+        return host_ip is not None and host_ip in network
+    if token_host.startswith("*."):
+        return host.endswith(token_host[1:])
+    if token_host.startswith("."):
+        return host == token_host[1:] or host.endswith(token_host)
+    return host == token_host or host.endswith(f".{token_host}")
+
+
+def should_bypass_proxy(
+    target_hosts: str | list[str] | tuple[str, ...] | set[str] | None, *, no_proxy_value: str | None = None,
+) -> bool:
+    """True when NO_PROXY (the environment, or ``no_proxy_value``) matches at least one target
+    host (a URL, ``host:port`` or bare host)."""
+    entries = no_proxy_entries(no_proxy_value)
+    if not entries or not target_hosts:
+        return False
+    candidates = [target_hosts] if isinstance(target_hosts, str) else list(target_hosts)
+    return any(
+        host and any(no_proxy_entry_matches(entry, host, port) for entry in entries)
+        for host, port in map(split_host_port, map(str, candidates)))

agent/retry_utils.py

@@ -5,6 +5,7 @@ when many sessions hit the same rate-limited provider concurrently.
 """
 
 import random
+import re
 import threading
 import time
 from datetime import datetime, timezone
@@ -63,6 +64,48 @@ def parse_retry_after_seconds(value_or_headers: Any) -> Optional[float]:
     return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
 
 
+# Free-text "reset" grammars providers put in error bodies, tried in order. One table so the
+# conversation loop's error context and the credential pool's cooldown agree on the same wait.
+_QUOTA_RESET_DELAY_RE = re.compile(r"quotaResetDelay[:\s\"]+(\d+(?:\.\d+)?)(ms|s)", re.IGNORECASE)
+# "Resets in 4hr 5min" (OpenCode Go weekly limits), "resets in 2 hours 5 minutes", "resets in 30s".
+_RESETS_IN_RE = re.compile(
+    r"resets?\s+in\s+"
+    r"(?:(\d+(?:\.\d+)?)\s*(?:h|hr|hrs|hour|hours)\b\s*)?"
+    r"(?:(\d+(?:\.\d+)?)\s*(?:m|min|mins|minute|minutes)\b\s*)?"
+    r"(?:(\d+(?:\.\d+)?)\s*(?:s|sec|secs|second|seconds)\b)?", re.IGNORECASE,
+)
+_RETRY_AFTER_SECONDS_RE = re.compile(r"retry\s+(?:after\s+)?(\d+(?:\.\d+)?)\s*(?:sec|secs|seconds|s\b)", re.IGNORECASE)
+
+
+def _quota_reset_seconds(m: "re.Match[str]") -> float:
+    value = float(m.group(1))
+    return value / 1000.0 if m.group(2).lower() == "ms" else value
+
+
+def _resets_in_seconds(m: "re.Match[str]") -> Optional[float]:
+    if not any(m.groups()):  # "resets in" with no unit-bearing number: not this grammar
+        return None
+    return float(m.group(1) or 0) * 3600 + float(m.group(2) or 0) * 60 + float(m.group(3) or 0)
+
+
+RETRY_DELAY_PATTERNS = (
+    (_QUOTA_RESET_DELAY_RE, _quota_reset_seconds),
+    (_RESETS_IN_RE, _resets_in_seconds),
+    (_RETRY_AFTER_SECONDS_RE, lambda m: float(m.group(1))),
+)
+
+
+def reset_delay_from_message(message: str) -> Optional[float]:
+    """Seconds-until-reset parsed from free-text provider error messages, or None."""
+    if not message:
+        return None
+    for pattern, to_seconds in RETRY_DELAY_PATTERNS:
+        m = pattern.search(message)
+        if m and (seconds := to_seconds(m)) is not None:
+            return seconds
+    return None
+
+
 def jittered_backoff(attempt: int, *, base_delay: float = 5.0, max_delay: float = 120.0, jitter_ratio: float = 0.5) -> float:
     """min(base * 2^(attempt-1), max_delay) + uniform jitter in
     [0, jitter_ratio * delay]. ``attempt`` is 1-based."""

hermes_cli/anon_auth.py

@@ -31,6 +31,7 @@ import time
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, Dict, Optional
 
+from agent.retry_utils import parse_retry_after_seconds
 from hermes_cli.auth_constants import (
     AuthError, DEFAULT_NOUS_PORTAL_URL, DEFAULT_NOUS_WELCOME_URL, _decode_jwt_claims, httpx)
 
@@ -661,11 +662,8 @@ def register_promotion_intent(
 
 
 def _retry_after_seconds(response: httpx.Response, default: float) -> float:
-    raw = (response.headers.get("retry-after") or "").strip()
-    try:
-        return max(0.0, float(raw)) if raw else default
-    except ValueError:
-        return default
+    seconds = parse_retry_after_seconds(response.headers)
+    return default if seconds is None else seconds
 
 
 def _sleep_until(wake: float, cancelled: Optional[Callable[[], bool]]) -> bool:

hermes_cli/observability/shared_metrics_sender.py

@@ -20,6 +20,7 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 
+from agent.retry_utils import parse_retry_after_seconds
 from hermes_cli.sqlite_util import write_txn
 
 from .shared_metrics import _isoformat, _utc_now
@@ -118,14 +119,11 @@ def _post(endpoint: str, payload: bytes, *, timeout: int) -> _Response:
 
 
 def _retry_after_seconds(value: str | None, default: int) -> int:
-    if not value:
-        return default
-    try:
-        # Contract sends seconds. Clamp so a bogus value cannot park a package for
-        # years, and never go below one second.
-        return max(1, min(int(float(value)), 86_400))
-    except (TypeError, ValueError):
+    seconds = parse_retry_after_seconds(value)
+    if seconds is None:
         return default
+    # Clamp so a bogus value cannot park a package for years, and never go below one second.
+    return max(1, min(int(seconds), 86_400))
 
 
 def reconcile_send_consent(

tests/agent/test_proxy_bypass_shared.py

@@ -0,0 +1,46 @@
+"""Invariant: the LLM transport (``agent/process_bootstrap``) and the platform adapters
+(``gateway/platforms/base``) answer "is this host in NO_PROXY" with the same matcher, so a
+corporate ``NO_PROXY=10.0.0.0/8`` bypasses the proxy for a self-hosted ``10.x`` model endpoint
+exactly as it does for Telegram/Discord/Slack.
+"""
+
+import pytest
+
+from agent.process_bootstrap import _get_proxy_for_base_url
+from agent.proxy_bypass import should_bypass_proxy
+from gateway.platforms.base import is_host_excluded_by_no_proxy, resolve_proxy_url
+
+_PROXY_KEYS = ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy",
+               "NO_PROXY", "no_proxy")
+
+
+@pytest.fixture
+def proxy_env(monkeypatch):
+    for key in _PROXY_KEYS:
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setenv("HTTPS_PROXY", "http://proxy.corp:3128")
+    monkeypatch.setattr("gateway.platforms.base.gateway_trust_env", lambda: True)
+    monkeypatch.setattr("gateway.platforms.base._detect_macos_system_proxy", lambda: None)
+    return monkeypatch
+
+
+@pytest.mark.parametrize("no_proxy, host", [
+    ("10.0.0.0/8", "10.1.2.3"),
+    ("*.internal", "svc.internal"),
+    ("localhost,.corp.example", "llm.corp.example"),
+    ("api.example.com:8443", "api.example.com:8443"),
+])
+def test_llm_and_adapter_paths_bypass_the_same_entries(proxy_env, no_proxy, host):
+    proxy_env.setenv("NO_PROXY", no_proxy)
+    assert should_bypass_proxy(host)
+    assert _get_proxy_for_base_url(f"https://{host}/v1") is None
+    assert resolve_proxy_url(target_hosts=host) is None
+    assert is_host_excluded_by_no_proxy(host.split(":")[0]) or ":" in host  # Slack passes bare hosts
+
+
+def test_non_matching_host_keeps_the_proxy_on_both_paths(proxy_env):
+    proxy_env.setenv("NO_PROXY", "10.0.0.0/8,*.internal")
+    assert _get_proxy_for_base_url("https://api.openai.com/v1") == "http://proxy.corp:3128"
+    assert resolve_proxy_url(target_hosts="api.telegram.org") == "http://proxy.corp:3128"
+    assert not is_host_excluded_by_no_proxy("slack.com")
+    assert is_host_excluded_by_no_proxy("files.slack.com", "slack.com")  # explicit value wins

tests/agent/test_retry_delay_parsers_shared.py

@@ -0,0 +1,75 @@
+"""Invariants for the shared retry-delay parsers in ``agent/retry_utils.py``.
+
+Cluster: every consumer of ``Retry-After`` / free-text reset grammars goes through one parser,
+so an HTTP-date header or a "resets in 2 hours 5 minutes" body yields the same wait everywhere.
+"""
+
+from datetime import datetime, timedelta, timezone
+from email.utils import format_datetime
+from types import SimpleNamespace
+
+import pytest
+
+from agent.retry_utils import parse_retry_after_seconds, reset_delay_from_message
+
+
+def _http_date(seconds_ahead: int) -> str:
+    return format_datetime(datetime.now(timezone.utc) + timedelta(seconds=seconds_ahead), usegmt=True)
+
+
+class TestRetryAfterHeaderOneParser:
+    def test_http_date_header_parsed_identically_at_formerly_divergent_sites(self):
+        """anon_auth, the error-context extractor and nous_rate_guard used to float() the header
+        and silently drop the RFC 7231 date form; all three must now agree with the canonical."""
+        from agent.agent_runtime_helpers import extract_api_error_context
+        from agent.nous_rate_guard import _parse_reset_seconds
+        from hermes_cli.anon_auth import _retry_after_seconds as anon_retry_after
+        import time
+
+        header = _http_date(90)
+        canonical = parse_retry_after_seconds(header)
+        assert 85 <= canonical <= 90
+
+        anon = anon_retry_after(SimpleNamespace(headers={"Retry-After": header}), default=1.0)
+        assert abs(anon - canonical) < 2
+
+        guard = _parse_reset_seconds({"Retry-After": header})
+        assert guard is not None and abs(guard - canonical) < 2
+
+        err = Exception("rate limited")
+        err.response = SimpleNamespace(headers={"Retry-After": header})
+        ctx = extract_api_error_context(err)
+        assert 85 <= ctx["reset_at"] - time.time() <= 91
+
+    def test_metrics_sender_clamps_on_top_of_the_shared_parser(self):
+        from hermes_cli.observability.shared_metrics_sender import _retry_after_seconds
+
+        assert _retry_after_seconds(_http_date(120), 7) in (119, 120)
+        assert _retry_after_seconds("0", 7) == 1          # floor survives
+        assert _retry_after_seconds("99999999", 7) == 86_400  # cap survives
+        assert _retry_after_seconds("garbage", 7) == 7
+
+
+class TestResetDelayOneTable:
+    @pytest.mark.parametrize("message, seconds", [
+        ("Weekly usage limit reached. Resets in 6hr 29min.", 6 * 3600 + 29 * 60),
+        ("resets in 2 hours 5 minutes", 2 * 3600 + 5 * 60),
+        ("Limit hit; resets in 45s", 45.0),
+        ('"quotaResetDelay": "1500ms"', 1.5),
+        ("please retry after 12 seconds", 12.0),
+    ])
+    def test_credential_pool_and_error_context_agree(self, message, seconds):
+        """The pooled-credential cooldown and the UI's error context read the same table, so the
+        long-form "hours/minutes" grammar (which the pool used to miss) resolves at both sites."""
+        import time
+        from agent.credential_pool import _normalize_error_context
+
+        assert reset_delay_from_message(message) == pytest.approx(seconds)
+        normalized = _normalize_error_context({"message": message})
+        assert normalized["reset_at"] - time.time() == pytest.approx(seconds, abs=2)
+
+    def test_no_grammar_means_no_reset(self):
+        from agent.credential_pool import _normalize_error_context
+
+        assert reset_delay_from_message("resets in the future, maybe") is None
+        assert "reset_at" not in _normalize_error_context({"message": "resets in the future, maybe"})

tests/agent/test_token_estimator_shared.py

@@ -0,0 +1,32 @@
+"""Invariant: every rough token estimate in the tree derives from ``estimate_tokens_rough`` /
+``CHARS_PER_TOKEN`` in ``agent/model_metadata.py``, so the ``/context`` breakdown's static
+categories, the conversation slice and native-compaction retention agree on non-Latin text.
+"""
+
+from agent.context_breakdown import _bytes_to_tokens, _chars_to_tokens
+from agent.model_metadata import CHARS_PER_TOKEN, estimate_tokens_rough
+from agent.native_compaction import _approx_tokens
+
+
+CYRILLIC = "Привет мир, это проверка оценки токенов. " * 40
+CJK = "これは日本語のテキストです。" * 40
+
+
+def test_breakdown_and_retention_use_the_canonical_estimator():
+    for text in (CYRILLIC, CJK, "plain ascii text " * 40):
+        canonical = estimate_tokens_rough(text)
+        assert _chars_to_tokens(text) == canonical
+        assert _approx_tokens(text) == canonical
+    # The old chars//4 shape under-counted these by ~2x; the canonical must not.
+    assert _chars_to_tokens(CYRILLIC) > (len(CYRILLIC) + 3) // 4 * 1.5
+    assert _chars_to_tokens(CJK) >= len(CJK)
+
+
+def test_byte_and_ratio_consumers_share_one_constant():
+    from agent.context_compressor import _CHARS_PER_TOKEN as compressor_ratio
+    from tools.budget_config import _CHARS_PER_TOKEN as budget_ratio
+    from tools.transcription_command import _PROMPT_CHARS_PER_TOKEN as whisper_ratio
+
+    assert compressor_ratio is budget_ratio is whisper_ratio is CHARS_PER_TOKEN
+    assert _bytes_to_tokens(CHARS_PER_TOKEN * 10) == 10
+    assert _bytes_to_tokens(None) is None

tests/tools/test_tool_output_truncate.py

@@ -0,0 +1,68 @@
+"""Invariant: terminal, execute_code, MCP and the bounded output collector truncate through one
+head/tail algorithm — 40% head / 60% tail, exactly one notice, kept text equal to the budget.
+"""
+
+import re
+
+import pytest
+
+from tools.tool_output_truncate import HEAD_RATIO, truncate_head_tail
+
+_NOTICE = re.compile(r"\n\n\.\.\. \[(?P<label>[A-Z ]+) TRUNCATED - (?P<omitted>[\d,]+) (?P<unit>chars|bytes) "
+                     r"omitted out of (?P<total>[\d,]+) total\] \.\.\.\n\n")
+
+TEXT = "".join(f"{i:04d}|" for i in range(2000))  # 10,000 chars, position-addressable
+
+
+def _assert_head_tail(result: str, *, budget: int, total: int, unit: str = "chars") -> dict:
+    notices = list(_NOTICE.finditer(result))
+    assert len(notices) == 1, result[:200]
+    m = notices[0]
+    head, tail = result[:m.start()], result[m.end():]
+    assert head == TEXT[:int(budget * HEAD_RATIO)]
+    assert tail == TEXT[-(budget - int(budget * HEAD_RATIO)):]
+    assert int(m["omitted"].replace(",", "")) == total - budget
+    assert int(m["total"].replace(",", "")) == total
+    assert m["unit"] == unit
+    return m.groupdict()
+
+
+def test_terminal_result_truncation(monkeypatch):
+    from tools import terminal_tool_result
+    monkeypatch.setattr("tools.tool_output_limits.get_max_bytes", lambda: 1000)
+    info = _assert_head_tail(terminal_tool_result._truncate_head_tail(TEXT), budget=1000, total=10_000)
+    assert info["label"] == "OUTPUT"
+
+
+def test_mcp_result_truncation():
+    from tools.mcp_tool_content import _truncate_mcp_text_result
+    info = _assert_head_tail(_truncate_mcp_text_result(TEXT, max_chars=1000), budget=1000, total=10_000)
+    assert info["label"] == "MCP RESULT"
+
+
+def test_execute_code_stdout_truncation(monkeypatch):
+    from tools import code_execution_tool
+    monkeypatch.setattr(code_execution_tool, "MAX_STDOUT_BYTES", 1000)
+    monkeypatch.setattr(code_execution_tool, "_spill_full_stdout", lambda _text: None)
+    text, meta = code_execution_tool._truncate_stdout_text(TEXT)
+    info = _assert_head_tail(text, budget=1000, total=10_000, unit="bytes")
+    assert info["label"] == "OUTPUT" and meta["stdout_bytes_omitted"] == 9000
+
+
+def test_bounded_output_collector_truncation():
+    from tools.environments.base_output import _BoundedOutputCollector
+    collector = _BoundedOutputCollector(1000)
+    for i in range(0, len(TEXT), 333):
+        collector.append(TEXT[i:i + 333])
+    rendered = collector.render()
+    notices = list(_NOTICE.finditer(rendered))
+    assert len(notices) == 1 and notices[0]["label"] == "OUTPUT"
+    assert len(rendered) <= 1000
+    head, tail = rendered[:notices[0].start()], rendered[notices[0].end():]
+    assert TEXT.startswith(head) and TEXT.endswith(tail)
+    assert abs(len(head) / (len(head) + len(tail)) - HEAD_RATIO) < 0.01
+
+
+@pytest.mark.parametrize("size", [0, 999, 1000])
+def test_text_within_budget_passes_through_untouched(size):
+    assert truncate_head_tail(TEXT[:size], 1000) == TEXT[:size]

tools/skills_hub_clawhub.py

@@ -9,6 +9,7 @@ from typing import Any, Dict, List, Optional, Tuple
 
 import httpx
 
+from agent.retry_utils import parse_retry_after_seconds
 from tools.skills_hub import _guarded_http_stream
 from tools.skills_hub_models import (
     GuardedFetchMixin, SkillBundle, SkillMeta, SkillSource, _cache_metas, _cached_metas, _get_json,
@@ -377,10 +378,9 @@ class ClawHubSource(GuardedFetchMixin, SkillSource):
                         return None
                     return self._owner_from_payload(self._coerce_skill_payload(raw))
                 if resp.status_code == 429:
-                    try:
-                        delay = float(resp.headers.get("Retry-After") or delay)
-                    except (TypeError, ValueError):
-                        pass
+                    retry_after = parse_retry_after_seconds(resp.headers)
+                    if retry_after is not None:
+                        delay = retry_after
                     reason = "HTTP 429"
                 elif 500 <= resp.status_code < 600:
                     reason = f"HTTP {resp.status_code}"
@@ -491,11 +491,7 @@ class ClawHubSource(GuardedFetchMixin, SkillSource):
                     if resp is None:
                         return files
                     if resp.status_code == 429:
-                        try:
-                            retry_after = int(resp.headers.get("retry-after", "5"))
-                        except (ValueError, TypeError):
-                            retry_after = 5
-                        retry_after = max(0, min(retry_after, 15))  # Cap wait time
+                        retry_after = max(0, min(int(parse_retry_after_seconds(resp.headers) or 5), 15))  # Cap wait time
                         logger.debug(
                             "ClawHub download rate-limited for %s, retrying in %ds (attempt %d/%d)",
                             slug, retry_after, attempt + 1, max_retries,

tools/skills_hub_github.py

@@ -11,6 +11,7 @@ from urllib.parse import quote
 import httpx
 
 from hermes_cli._subprocess_compat import windows_hide_flags
+from agent.retry_utils import parse_retry_after_seconds
 from tools.skills_guard import TRUSTED_REPOS
 from tools.skills_hub_models import (
     SkillBundle, SkillMeta, SkillSource, _cache_metas, _cached_metas, _dedupe_by_trust,
@@ -431,9 +432,9 @@ class GitHubSource(SkillSource):
                                            "Set GITHUB_TOKEN or install the gh CLI to raise the limit to 5,000/hr.")
                         return resp
                     reset = resp.headers.get("X-RateLimit-Reset", "")
-                    retry_after = resp.headers.get("Retry-After", "")
-                    if retry_after.isdigit():
-                        wait = min(float(retry_after), 60.0)
+                    retry_after = parse_retry_after_seconds(resp.headers)
+                    if retry_after is not None:
+                        wait = min(retry_after, 60.0)
                     elif reset.isdigit():
                         delta = float(reset) - time.time()
                         if 0 < delta <= 60.0:

tools/tool_output_truncate.py

@@ -0,0 +1,32 @@
+"""Head/tail truncation for oversized tool output (terminal, execute_code, MCP results).
+
+One algorithm and one notice wording: 40% head (errors surface early) / 60% tail (the most
+recent lines matter most) around a single ``... [<LABEL> TRUNCATED - N <unit> omitted out of
+T total] ...`` marker, so downstream code that recognises the marker sees one shape. Limits
+live in ``tools/tool_output_limits.py``; line-snapped, path-bearing footers (web/browser,
+read_file pagination) are different products and stay separate.
+"""
+
+from __future__ import annotations
+
+HEAD_RATIO = 0.4
+
+
+def truncation_notice(omitted: int, total: int, *, label: str = "OUTPUT", unit: str = "chars") -> str:
+    return f"\n\n... [{label} TRUNCATED - {omitted:,} {unit} omitted out of {total:,} total] ...\n\n"
+
+
+def head_tail_split(budget: int) -> tuple[int, int]:
+    """``(head, tail)`` character budgets for ``budget`` total."""
+    head = int(budget * HEAD_RATIO)
+    return head, budget - head
+
+
+def truncate_head_tail(text: str, max_chars: int, *, label: str = "OUTPUT") -> str:
+    """``text`` unchanged when it fits ``max_chars``; otherwise head + notice + tail (the kept
+    text is exactly ``max_chars`` long, the notice rides on top)."""
+    if len(text) <= max_chars:
+        return text
+    head, tail = head_tail_split(max_chars)
+    omitted = len(text) - head - tail
+    return text[:head] + truncation_notice(omitted, len(text), label=label) + text[-tail:]