refactor(agent/G_small): tighten provider ABC, ssl_guard, tracker formatting, summaries predicate

agent/rate_limit_tracker.py

@@ -125,8 +125,8 @@ def _fmt_seconds(seconds: float) -> str:
     if s < 3600:
         m, sec = divmod(s, 60)
         return f"{m}m {sec}s" if sec else f"{m}m"
-    h, remainder = divmod(s, 3600)
-    m = remainder // 60
+    h, m = divmod(s, 3600)
+    m //= 60
     return f"{h}h {m}m" if m else f"{h}h"
 
 
@@ -140,7 +140,6 @@ def _bucket_line(label: str, bucket: RateLimitBucket, label_width: int = 14) ->
     """Format one bucket as a single line."""
     if bucket.limit <= 0:
         return f"  {label:<{label_width}}  (no data)"
-
     pct = bucket.usage_pct
     used, limit, remaining = map(_fmt_count, (bucket.used, bucket.limit, bucket.remaining))
     reset = _fmt_seconds(bucket.remaining_seconds_now)

agent/reasoning_summaries.py

@@ -21,8 +21,5 @@ def separate_glued_reasoning_blocks(previous: str, delta: str) -> str:
     alone: its deltas carry their own whitespace, and a fragment that merely opens emphasis
     (``**`` alone) is not a part boundary — summary parts carry the whole heading in one delta.
     """
-    if not previous or not delta or previous[-1].isspace():
-        return delta
-    if not delta.startswith("**") or "**" not in delta[2:]:
-        return delta
-    return f"\n\n{delta}"
+    glued = previous and delta and not previous[-1].isspace() and delta.startswith("**") and "**" in delta[2:]
+    return f"\n\n{delta}" if glued else delta

agent/runtime_cwd.py

@@ -83,18 +83,12 @@ def _resolve_configured_cwd(*, override_is_final: bool) -> Path | None:
 
 
 def resolve_agent_cwd() -> Path:
-    """Configured cwd, else the launch dir (os.getcwd() — its OSError on a
-    deleted cwd deliberately propagates; the caller owns that guard)."""
-    p = _resolve_configured_cwd(override_is_final=False)
-    return p if p is not None else Path(os.getcwd())
+    """Configured cwd, else the launch dir (os.getcwd()'s OSError on a deleted cwd deliberately propagates)."""
+    return _resolve_configured_cwd(override_is_final=False) or Path(os.getcwd())
 
 
 def resolve_context_cwd() -> Path | None:
-    """Configured cwd for context-file discovery, or None for "no configured cwd".
-
-    None makes build_context_files_prompt fall back to the launch dir (correct for a local CLI
-    launched inside a real project). An existing configured path is honored verbatim —
-    including the Hermes source tree itself, a legitimate workspace when developing Hermes
-    (fallback-directory policy lives in build_context_files_prompt).
-    """
+    """Configured cwd for context-file discovery, or None (build_context_files_prompt then falls back to the
+    launch dir). An existing configured path is honored verbatim — including the Hermes source tree, a
+    legitimate workspace when developing Hermes; fallback-directory policy lives in the caller."""
     return _resolve_configured_cwd(override_is_final=True)

agent/side_question.py

@@ -81,11 +81,10 @@ def render_history_for_side_question(
     kept: List[str] = []
     used = 0
     for line in reversed(lines):
-        cost = len(line) + 1
-        if used + cost > char_budget and kept:
+        if used + len(line) + 1 > char_budget and kept:
             break
         kept.append(line)
-        used += cost
+        used += len(line) + 1
     if not kept:
         return "(no prior conversation)"
     prefix = "[...older conversation omitted...]\n" if len(kept) < len(lines) else ""

agent/ssl_guard.py

@@ -40,19 +40,15 @@ def _validate_bundle_path(label: str, value: str, *, require_substantial: bool =
         raise _ssl_err(f"{label} CA bundle at {value} cannot be loaded: {exc}") from exc
     try:
         loaded_certs = ctx.get_ca_certs()
-    except NotImplementedError:
-        # truststore-backed SSLContext (Windows OS trust store) lacks get_ca_certs(); loading above already validated it.
+    except NotImplementedError:  # truststore-backed SSLContext (Windows) lacks get_ca_certs(); loading validated it
         return
     if not loaded_certs:
         raise _ssl_err(f"{label} CA bundle at {value} did not load any certificates")
 
 
 def verify_ca_bundle() -> None:
-    """Verify configured and bundled CA certificates are present and loadable.
-
-    Raises SSLConfigurationError when an explicit CA-bundle env var points at a
-    bad path or certifi's bundled ``cacert.pem`` is missing/corrupt.
-    """
+    """Raise SSLConfigurationError when a CA-bundle env var points at a bad path or certifi's ``cacert.pem``
+    is missing/corrupt."""
     if is_truthy_value(os.getenv("HERMES_SKIP_SSL_GUARD", "")):
         logger.debug("SSL CA bundle guard skipped via HERMES_SKIP_SSL_GUARD")
         return

agent/terminal_env_provider.py

@@ -40,15 +40,15 @@ class TerminalEnvironmentProvider(ProviderBase):
       sandbox identity; opt in when a shared name would let two ephemeral runs destroy each other.
     """
 
+    is_remote: bool = True
+    is_container: bool = True
+    session_isolated_when_nonpersistent: bool = False
+
     @property
     def description(self) -> str:
         """One-line description shown in backend pickers."""
         return f"Run commands in a {self.display_name} environment."
 
-    is_remote: bool = True
-    is_container: bool = True
-    session_isolated_when_nonpersistent: bool = False
-
     @property
     def skip_container_guards(self) -> bool:
         return self.is_container
@@ -63,30 +63,23 @@ class TerminalEnvironmentProvider(ProviderBase):
 
     @property
     def env_description(self) -> str:
-        """Prompt-builder fallback for where commands run when the live backend probe fails
-        at system-prompt build time (e.g. ``"a Daytona workspace (Linux)"``)."""
+        """Prompt-builder fallback for where commands run when the live probe fails (e.g. ``"a Daytona workspace (Linux)"``)."""
         return f"a {self.display_name} environment (likely Linux)"
 
-    # -- Availability / setup UX -------------------------------------------
-
     @abc.abstractmethod
     def is_available(self) -> bool:
-        """True when this backend can service commands. Cheap check only — must
-        NOT make network calls; runs during requirement checks and UI paints."""
+        """True when this backend can service commands. Cheap, NO network calls: runs during UI paints."""
 
     def check_requirements(self, config: Dict[str, Any]) -> bool:
-        """Full requirements check for :func:`check_terminal_requirements` with the merged
-        terminal env config; log actionable errors before returning False."""
+        """Full requirements check with the merged terminal env config; log actionable errors before returning False."""
         return self.is_available()
 
     def probe(self) -> Tuple[str, str]:
-        """Dashboard picker health probe ``(status, detail)`` with status in
-        ``ready`` / ``needs_setup`` / ``unavailable``. Must never raise; stay fast (<~2s)."""
+        """Dashboard picker health ``(status, detail)``, status in ``ready``/``needs_setup``/``unavailable``. Never raise; <~2s."""
         return ("ready", "") if self.is_available() else ("needs_setup", f"{self.display_name} is not configured.")
 
     def setup_instructions(self) -> List[str]:
-        """Lines printed by ``hermes setup`` after this backend is selected. The wizard
-        persists ``terminal.backend`` itself; interactive flows go in :meth:`post_setup`."""
+        """Lines printed by ``hermes setup`` after selection (the wizard persists ``terminal.backend`` itself)."""
         return []
 
     def post_setup(self) -> None:
@@ -98,20 +91,13 @@ class TerminalEnvironmentProvider(ProviderBase):
             ok = bool(self.is_available())
         except Exception:
             ok = False
-        detail = "(configured)" if ok else "(not configured — see setup instructions)"
-        return [(ok, f"{self.display_name} backend", detail)]
-
-    # -- The factory -------------------------------------------------------
+        return [(ok, f"{self.display_name} backend", "(configured)" if ok else "(not configured — see setup instructions)")]
 
     @abc.abstractmethod
     def create_environment(
         self, *, cwd: str, timeout: int, task_id: str = "default", image: Optional[str] = None,
         container_config: Optional[Dict[str, Any]] = None, **kwargs: Any,
     ):
-        """Create and return an execution environment (``BaseEnvironment`` duck type).
-
-        MUST accept ``**kwargs`` and ignore unknown keys so the factory signature can evolve
-        without breaking older plugins. ``task_id`` keys environment reuse/persistence;
-        ``container_config`` carries ``container_cpu`` / ``container_memory`` /
-        ``container_disk`` / ``container_persistent`` when :attr:`is_container` is True.
-        """
+        """Create an execution environment (``BaseEnvironment`` duck type). MUST accept ``**kwargs`` and ignore
+        unknown keys so the factory can evolve without breaking older plugins. ``task_id`` keys reuse/persistence;
+        ``container_config`` carries ``container_cpu/memory/disk/persistent`` when :attr:`is_container`."""

agent/thinking_timeout_guidance.py

@@ -28,8 +28,9 @@ def is_thinking_timeout(classified: object, model: str, error_msg: str) -> bool:
     """
     from agent.reasoning_timeouts import get_reasoning_stale_timeout_floor
 
-    reason = getattr(classified, "reason", None)
-    if getattr(reason, "value", None) != "timeout" or get_reasoning_stale_timeout_floor(model) is None:
+    if getattr(getattr(classified, "reason", None), "value", None) != "timeout":
+        return False
+    if get_reasoning_stale_timeout_floor(model) is None:
         return False
     error_msg_lower = (error_msg or "").lower()
     return any(p in error_msg_lower for p in _THINKING_TIMEOUT_SUBSTRINGS)