From 26bbac4ab9fcc2fa3de09f8064fcd0f7cd723281 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 18:22:13 -0700
Subject: [PATCH] refactor(agent/gemini,acp,credits): compact wire adapters and
credits parser (parity-verified)
---
agent/copilot_acp_client.py | 314 +++++++++-----------------
agent/credits_tracker.py | 297 +++++++++++--------------
agent/fast_mode.py | 23 +-
agent/gemini_native_adapter.py | 396 +++++++++++++--------------------
agent/gemini_schema.py | 6 +-
agent/jiter_preload.py | 16 +-
agent/lazy_forward.py | 17 +-
agent/lmstudio_reasoning.py | 47 ++--
8 files changed, 436 insertions(+), 680 deletions(-)
diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py
index d5e1bb90a4..c52131e08a 100644
--- a/agent/copilot_acp_client.py
+++ b/agent/copilot_acp_client.py
@@ -46,6 +46,32 @@ _ROLE_LABELS = {"system": "System", "user": "User", "assistant": "Assistant", "t
# so a CLI installed mid-session is picked up.
_ACP_PROBE_CACHE: dict[str, bool] = {}
+_PROMPT_PREAMBLE = (
+ "You are being used as the active ACP agent backend for Hermes.",
+ "Use ACP capabilities to complete tasks.",
+ "IMPORTANT: If you take an action with a tool, you MUST output tool calls using {...} blocks with JSON exactly in OpenAI function-call shape.",
+ "If no tool is needed, answer normally.",
+)
+_INITIALIZE_PARAMS = {
+ "protocolVersion": 1,
+ "clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
+ "clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
+}
+_DEPRECATED_CLI_ERROR = (
+ "Hermes ACP mode requires the NEW GitHub Copilot CLI "
+ "(github.com/github/copilot-cli), but the binary it just "
+ "spawned is the deprecated `gh copilot` extension.\n\n"
+ "Install the new CLI:\n"
+ " npm install -g @github/copilot\n"
+ " # then verify with: copilot --help\n\n"
+ "If `copilot` already resolves to the new CLI but you still see this,\n"
+ "point Hermes at it explicitly:\n"
+ " export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
+ "Alternative: use the `copilot` provider (no ACP, hits the Copilot API\n"
+ "directly with a Copilot subscription token) via `hermes setup`.\n\n"
+ "Original error:\n"
+)
+
def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
"""True iff stderr looks like the deprecated gh-copilot extension's banner."""
@@ -54,11 +80,7 @@ def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
def _resolve_command() -> str:
- return (
- os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
- or os.getenv("COPILOT_CLI_PATH", "").strip()
- or "copilot"
- )
+ return os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip() or os.getenv("COPILOT_CLI_PATH", "").strip() or "copilot"
def _resolve_args() -> list[str]:
@@ -70,10 +92,10 @@ def _acp_supported(command: str, args: list[str]) -> bool | None:
"""Tri-state probe: does ``command`` accept ``--acp``?
A CLI without the flag (older releases, Claude Code v2.x) exits 1 with
- ``error: unknown option '--acp'`` and the parent then waits the full
- child timeout for stdout that never arrives. True = help advertises --acp;
- False = help ran cleanly without it (caller fast-fails); None = inconclusive
- (binary missing / --help failed), caller falls through to the normal spawn error.
+ ``error: unknown option '--acp'`` and the parent then waits the full child
+ timeout for stdout that never arrives. True = help advertises --acp; False =
+ help ran cleanly without it (caller fast-fails); None = inconclusive (binary
+ missing / --help failed), caller falls through to the normal spawn error.
Only probes when ``--acp`` is among ``args`` — a custom transport is the operator's business.
"""
if "--acp" not in args:
@@ -97,7 +119,7 @@ def _acp_supported(command: str, args: list[str]) -> bool | None:
def _resolve_home_dir() -> str:
- """Return a stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
+ """Stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
home = os.environ.get("HOME", "").strip()
if home:
return home
@@ -137,99 +159,61 @@ def _permission_denied(message_id: Any) -> dict[str, Any]:
return _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
-def _model_selection_request(
- session: dict[str, Any], requested_model: str
-) -> tuple[str, dict[str, str]] | None:
+def _enabled_ids(entries: Any, key: str) -> set[str]:
+ """Ids of ``entries`` (dicts) whose ``_meta.copilotEnablement`` is not ``disabled``."""
+ return {
+ str(e.get(key) or "").strip()
+ for e in (entries or [])
+ if isinstance(e, dict) and str((e.get("_meta") or {}).get("copilotEnablement") or "").strip().lower() != "disabled"
+ }
+
+
+def _model_selection_request(session: dict[str, Any], requested_model: str) -> tuple[str, dict[str, str]] | None:
"""Return the ACP request that selects ``requested_model`` for ``session``.
- Prefer stable v1 ``session/set_config_option``. Fall back to Copilot's
- pre-stabilization ``session/set_model`` extension only when no model
- config option is advertised. A reported model list is authoritative:
- unknown and policy-disabled ids return None instead of being sent.
+ Prefer stable v1 ``session/set_config_option``; fall back to Copilot's
+ pre-stabilization ``session/set_model`` extension only when no model config
+ option is advertised. A reported model list is authoritative: unknown and
+ policy-disabled ids return None instead of being sent.
"""
session_id = str(session.get("sessionId") or "").strip()
requested_model = str(requested_model or "").strip()
if not session_id or not requested_model or requested_model == "copilot-acp":
return None
-
- config_options = [
- o for o in (session.get("configOptions") or []) if isinstance(o, dict)
- ]
model_option = next(
- (
- o for o in config_options
- if o.get("category") == "model" or o.get("id") == "model"
- ),
+ (o for o in (session.get("configOptions") or [])
+ if isinstance(o, dict) and (o.get("category") == "model" or o.get("id") == "model")),
None,
)
if model_option is not None:
- enabled_values = {
- str(o.get("value") or "").strip()
- for o in (model_option.get("options") or [])
- if isinstance(o, dict)
- and str(
- ((o.get("_meta") or {}).get("copilotEnablement")) or ""
- ).strip().lower() != "disabled"
- }
- if requested_model not in enabled_values:
+ if requested_model not in _enabled_ids(model_option.get("options"), "value"):
return None
- return (
- "session/set_config_option",
- {
- "sessionId": session_id,
- "configId": str(model_option.get("id") or "model"),
- "value": requested_model,
- },
- )
-
- advertised = [
- m
- for m in ((session.get("models") or {}).get("availableModels") or [])
- if isinstance(m, dict)
- ]
- available = {
- str(m.get("modelId") or "").strip()
- for m in advertised
- if str(
- ((m.get("_meta") or {}).get("copilotEnablement")) or ""
- ).strip().lower() != "disabled"
- }
+ return "session/set_config_option", {
+ "sessionId": session_id, "configId": str(model_option.get("id") or "model"), "value": requested_model,
+ }
+ available = _enabled_ids((session.get("models") or {}).get("availableModels"), "modelId")
if available and requested_model not in available:
return None
- return (
- "session/set_model",
- {"sessionId": session_id, "modelId": requested_model},
- )
+ return "session/set_model", {"sessionId": session_id, "modelId": requested_model}
def _format_messages_as_prompt(
- messages: list[dict[str, Any]],
- model: str | None = None,
- tools: list[dict[str, Any]] | None = None,
+ messages: list[dict[str, Any]], model: str | None = None, tools: list[dict[str, Any]] | None = None,
tool_choice: Any = None,
) -> str:
- sections: list[str] = [
- "You are being used as the active ACP agent backend for Hermes.",
- "Use ACP capabilities to complete tasks.",
- "IMPORTANT: If you take an action with a tool, you MUST output tool calls using {...} blocks with JSON exactly in OpenAI function-call shape.",
- "If no tool is needed, answer normally.",
- ]
# Deliberately no "requested model" line: the model is applied for real via ACP
# session/set_model; a prompt-text mention makes a substituted backend model
# FALSELY self-identify as the requested one. Identity comes from the backend.
# Copilot has no tools of its own that collide with Hermes', so forward the whole toolset.
- sections.extend(_render_tool_bridge_sections(tools, tool_choice))
-
+ sections: list[str] = [*_PROMPT_PREAMBLE, *_render_tool_bridge_sections(tools, tool_choice)]
transcript: list[str] = []
for message in messages:
if not isinstance(message, dict):
continue
role = str(message.get("role") or "unknown").strip().lower()
- if role not in _ROLE_LABELS:
- role = "context"
rendered = _render_message_content(message.get("content"))
if rendered:
- transcript.append(f"{_ROLE_LABELS[role]}:\n{rendered}")
+ transcript.append(f"{_ROLE_LABELS.get(role, 'Context')}:\n{rendered}")
if transcript:
sections.append("Conversation transcript:\n\n" + "\n\n".join(transcript))
sections.append("Continue the conversation from the latest user request.")
@@ -239,23 +223,18 @@ def _format_messages_as_prompt(
def _render_message_content(content: Any) -> str:
if content is None:
return ""
- if isinstance(content, str):
- return content.strip()
if isinstance(content, dict):
if "text" in content:
return str(content.get("text") or "").strip()
if isinstance(content.get("content"), str):
- return str(content.get("content") or "").strip()
+ return content["content"].strip()
return json.dumps(content, ensure_ascii=True)
if isinstance(content, list):
- parts: list[str] = []
- for item in content:
- if isinstance(item, str):
- parts.append(item)
- elif isinstance(item, dict):
- text = item.get("text")
- if isinstance(text, str) and text.strip():
- parts.append(text.strip())
+ parts = [
+ item if isinstance(item, str) else item["text"].strip()
+ for item in content
+ if isinstance(item, str) or (isinstance(item, dict) and isinstance(item.get("text"), str) and item["text"].strip())
+ ]
return "\n".join(parts).strip()
return str(content).strip()
@@ -279,9 +258,9 @@ def _effective_timeout(timeout: Any) -> float:
return _DEFAULT_TIMEOUT_SECONDS
if isinstance(timeout, (int, float)):
return float(timeout)
- _candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
- _numeric = [float(v) for v in _candidates if isinstance(v, (int, float))]
- return max(_numeric) if _numeric else _DEFAULT_TIMEOUT_SECONDS
+ candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
+ numeric = [float(v) for v in candidates if isinstance(v, (int, float))]
+ return max(numeric) if numeric else _DEFAULT_TIMEOUT_SECONDS
def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
@@ -293,13 +272,11 @@ def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
content = path.read_text(encoding="utf-8")
except FileNotFoundError:
content = ""
- line = params.get("line")
- limit = params.get("limit")
+ line, limit = params.get("line"), params.get("limit")
if isinstance(line, int) and line > 1:
lines = content.splitlines(keepends=True)
- start = line - 1
- end = start + limit if isinstance(limit, int) and limit > 0 else None
- content = "".join(lines[start:end])
+ end = line - 1 + limit if isinstance(limit, int) and limit > 0 else None
+ content = "".join(lines[line - 1:end])
if content:
content = redact_sensitive_text(content, force=True)
return {"content": content}
@@ -325,19 +302,6 @@ def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any:
_FS_HANDLERS = {"fs/read_text_file": _fs_read_text_file, "fs/write_text_file": _fs_write_text_file}
-class _ACPChatCompletions:
- def __init__(self, client: "CopilotACPClient"):
- self._client = client
-
- def create(self, **kwargs: Any) -> Any:
- return self._client._create_chat_completion(**kwargs)
-
-
-class _ACPChatNamespace:
- def __init__(self, client: "CopilotACPClient"):
- self.completions = _ACPChatCompletions(client)
-
-
class CopilotACPClient:
"""Minimal OpenAI-client-compatible facade for Copilot ACP."""
@@ -348,17 +312,9 @@ class CopilotACPClient:
HERMES_SKIP_ASYNC_WRAP = True
def __init__(
- self,
- *,
- api_key: str | None = None,
- base_url: str | None = None,
- default_headers: dict[str, str] | None = None,
- acp_command: str | None = None,
- acp_args: list[str] | None = None,
- acp_cwd: str | None = None,
- command: str | None = None,
- args: list[str] | None = None,
- **_: Any,
+ self, *, api_key: str | None = None, base_url: str | None = None, default_headers: dict[str, str] | None = None,
+ acp_command: str | None = None, acp_args: list[str] | None = None, acp_cwd: str | None = None,
+ command: str | None = None, args: list[str] | None = None, **_: Any,
):
self.api_key = api_key or "copilot-acp"
self.base_url = base_url or ACP_MARKER_BASE_URL
@@ -366,7 +322,7 @@ class CopilotACPClient:
self._acp_command = acp_command or command or _resolve_command()
self._acp_args = list(acp_args or args or _resolve_args())
self._acp_cwd = str(Path(acp_cwd or os.getcwd()).resolve())
- self.chat = _ACPChatNamespace(self)
+ self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
self.is_closed = False
self._active_process: subprocess.Popen[str] | None = None
self._active_process_lock = threading.Lock()
@@ -388,37 +344,23 @@ class CopilotACPClient:
pass
def _create_chat_completion(
- self,
- *,
- model: str | None = None,
- messages: list[dict[str, Any]] | None = None,
- timeout: float | None = None,
- tools: list[dict[str, Any]] | None = None,
- tool_choice: Any = None,
- stream: bool = False,
- **_: Any,
+ self, *, model: str | None = None, messages: list[dict[str, Any]] | None = None, timeout: float | None = None,
+ tools: list[dict[str, Any]] | None = None, tool_choice: Any = None, stream: bool = False, **_: Any,
) -> Any:
prompt_text = _format_messages_as_prompt(messages or [], model=model, tools=tools, tool_choice=tool_choice)
- response_text, reasoning_text = self._run_prompt(
- prompt_text, timeout_seconds=_effective_timeout(timeout), model=model
- )
+ response_text, reasoning_text = self._run_prompt(prompt_text, timeout_seconds=_effective_timeout(timeout), model=model)
tool_calls, cleaned_text = _extract_tool_calls_from_text(response_text)
-
- usage = SimpleNamespace(
- prompt_tokens=0,
- completion_tokens=0,
- total_tokens=0,
- prompt_tokens_details=SimpleNamespace(cached_tokens=0),
- )
assistant_message = SimpleNamespace(
- content=cleaned_text,
- tool_calls=tool_calls,
- reasoning=reasoning_text or None,
- reasoning_content=reasoning_text or None,
- reasoning_details=None,
+ content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning_text or None,
+ reasoning_content=reasoning_text or None, reasoning_details=None,
+ )
+ completion = SimpleNamespace(
+ choices=[SimpleNamespace(message=assistant_message, finish_reason="tool_calls" if tool_calls else "stop")],
+ usage=SimpleNamespace(
+ prompt_tokens=0, completion_tokens=0, total_tokens=0, prompt_tokens_details=SimpleNamespace(cached_tokens=0)
+ ),
+ model=model or "copilot-acp",
)
- choice = SimpleNamespace(message=assistant_message, finish_reason="tool_calls" if tool_calls else "stop")
- completion = SimpleNamespace(choices=[choice], usage=usage, model=model or "copilot-acp")
return _completion_to_stream_chunks(completion) if stream else completion
def _spawn(self) -> subprocess.Popen[str]:
@@ -443,14 +385,9 @@ class CopilotACPClient:
proc = subprocess.Popen(
[self._acp_command] + self._acp_args,
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- text=True, encoding='utf-8', errors='replace',
- bufsize=1,
- cwd=self._acp_cwd,
- env=_build_subprocess_env(),
- creationflags=windows_hide_flags(),
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ text=True, encoding='utf-8', errors='replace', bufsize=1,
+ cwd=self._acp_cwd, env=_build_subprocess_env(), creationflags=windows_hide_flags(),
)
except FileNotFoundError as exc:
raise RuntimeError(
@@ -482,9 +419,7 @@ class CopilotACPClient:
inbox.put({"raw": line.rstrip("\n")})
def _stderr_reader() -> None:
- if proc.stderr is None:
- return
- for line in proc.stderr:
+ for line in proc.stderr or ():
stderr_tail.append(line.rstrip("\n"))
threading.Thread(target=_stdout_reader, daemon=True).start()
@@ -497,20 +432,15 @@ class CopilotACPClient:
request_id = next_id
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + "\n")
proc.stdin.flush()
-
deadline = time.monotonic() + timeout_seconds
- while time.monotonic() < deadline:
- if proc.poll() is not None:
- break
+ while time.monotonic() < deadline and proc.poll() is None:
try:
msg = inbox.get(timeout=0.1)
except queue.Empty:
continue
if self._handle_server_message(
msg, process=proc, cwd=self._acp_cwd, text_parts=text_parts, reasoning_parts=reasoning_parts
- ):
- continue
- if msg.get("id") != request_id:
+ ) or msg.get("id") != request_id:
continue
if "error" in msg:
err = msg.get("error") or {}
@@ -520,85 +450,48 @@ class CopilotACPClient:
stderr_text = "\n".join(stderr_tail).strip()
if proc.poll() is not None and stderr_text:
if _is_gh_copilot_deprecation_message(stderr_text):
- raise RuntimeError(
- "Hermes ACP mode requires the NEW GitHub Copilot CLI "
- "(github.com/github/copilot-cli), but the binary it just "
- "spawned is the deprecated `gh copilot` extension.\n\n"
- "Install the new CLI:\n"
- " npm install -g @github/copilot\n"
- " # then verify with: copilot --help\n\n"
- "If `copilot` already resolves to the new CLI but you still see this,\n"
- "point Hermes at it explicitly:\n"
- " export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
- "Alternative: use the `copilot` provider (no ACP, hits the Copilot API\n"
- "directly with a Copilot subscription token) via `hermes setup`.\n\n"
- f"Original error:\n{stderr_text}"
- )
+ raise RuntimeError(_DEPRECATED_CLI_ERROR + stderr_text)
raise RuntimeError(f"Copilot ACP process exited early: {stderr_text}")
raise TimeoutError(f"Timed out waiting for Copilot ACP response to {method}.")
try:
- _request(
- "initialize",
- {
- "protocolVersion": 1,
- "clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
- "clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
- },
- )
+ _request("initialize", _INITIALIZE_PARAMS)
session = _request("session/new", {"cwd": self._acp_cwd, "mcpServers": []}) or {}
session_id = str(session.get("sessionId") or "").strip()
if not session_id:
raise RuntimeError("Copilot ACP did not return a sessionId.")
-
# Prefer the stable ACP v1 session-config API (category="model" select
# option + session/set_config_option); session/set_model is the fallback.
if requested_model and requested_model != "copilot-acp":
try:
selection = _model_selection_request(session, requested_model)
if selection is not None:
- method, params = selection
- _request(method, params)
+ _request(*selection)
else:
- logger.warning(
- "Copilot ACP does not offer model %r; using the "
- "session default.",
- requested_model,
- )
+ logger.warning("Copilot ACP does not offer model %r; using the session default.", requested_model)
except Exception as exc:
logger.warning(
- "Copilot ACP model selection for %r failed; continuing "
- "with the session default: %s",
- requested_model,
- exc,
+ "Copilot ACP model selection for %r failed; continuing with the session default: %s",
+ requested_model, exc,
)
-
text_parts: list[str] = []
reasoning_parts: list[str] = []
_request(
- "session/prompt",
- {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]},
- text_parts=text_parts,
- reasoning_parts=reasoning_parts,
+ "session/prompt", {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]},
+ text_parts=text_parts, reasoning_parts=reasoning_parts,
)
return "".join(text_parts), "".join(reasoning_parts)
finally:
self.close()
def _handle_server_message(
- self,
- msg: dict[str, Any],
- *,
- process: subprocess.Popen[str],
- cwd: str,
- text_parts: list[str] | None,
- reasoning_parts: list[str] | None,
+ self, msg: dict[str, Any], *, process: subprocess.Popen[str], cwd: str,
+ text_parts: list[str] | None, reasoning_parts: list[str] | None,
) -> bool:
"""Consume a server->client message; True when handled (notification or request answered)."""
method = msg.get("method")
if not isinstance(method, str):
return False
-
if method == "session/update":
update = (msg.get("params") or {}).get("update") or {}
kind = str(update.get("sessionUpdate") or "").strip()
@@ -609,22 +502,19 @@ class CopilotACPClient:
elif kind == "agent_thought_chunk" and chunk_text and reasoning_parts is not None:
reasoning_parts.append(chunk_text)
return True
-
if process.stdin is None:
return True
message_id = msg.get("id")
- params = msg.get("params") or {}
if method == "session/request_permission":
response = _permission_denied(message_id)
elif method in _FS_HANDLERS:
try:
- response = _jsonrpc_result(message_id, _FS_HANDLERS[method](params, cwd))
+ response = _jsonrpc_result(message_id, _FS_HANDLERS[method](msg.get("params") or {}, cwd))
except Exception as exc:
response = _jsonrpc_error(message_id, -32602, str(exc))
else:
response = _jsonrpc_error(message_id, -32601, f"ACP client method '{method}' is not supported by Hermes yet.")
-
process.stdin.write(json.dumps(response) + "\n")
process.stdin.flush()
return True
diff --git a/agent/credits_tracker.py b/agent/credits_tracker.py
index 54f0de0795..8f7a1e769e 100644
--- a/agent/credits_tracker.py
+++ b/agent/credits_tracker.py
@@ -1,25 +1,13 @@
-"""Credits tracking for Nous inference API responses: parses x-nous-credits-*
-(and optional x-nous-tool-pool-*) headers into a validated CreditsState, with
+"""Credits tracking for Nous inference API responses: parses ``x-nous-credits-*``
+(and optional ``x-nous-tool-pool-*``) headers into a validated CreditsState, with
depletion detection (paid_access), subscription-cap used_fraction, and warn-once
-schema-version gating. The hardened parser used by all live consumers.
-
-Header schema (x-nous-credits-*; each *-micros balance has a *-usd twin holding
-the server's formatted USD string):
- version contract/schema version
- remaining-micros/-usd total remaining balance
- subscription-micros/-usd subscription balance (SIGNED; may be negative/debt)
- subscription-limit-* subscription cap (PAIRED/optional)
- rollover-micros rolled-over balance
- purchased-micros/-usd purchased balance
- denominator-kind "subscription_cap" | "none"
- paid-access "true" | "false" (STRING!)
- disabled-reason reason string (header omitted when null)
- as-of-ms server-side timestamp (ms epoch)
-Tool-pool headers use a SEPARATE prefix: x-nous-tool-pool-micros (balance) and
-x-nous-tool-pool-gated-off ("true" | "false" STRING!).
-
-Money is handled as micros ints only; *_usd values are preserved verbatim as
-the raw strings the server sent (never re-parsed to float).
+schema-version gating. Header schema (each ``*-micros`` balance has a ``*-usd``
+twin holding the server's formatted USD string): version, remaining, subscription
+(SIGNED; may be debt), subscription-limit (PAIRED/optional), rollover, purchased,
+denominator-kind ("subscription_cap" | "none"), paid-access ("true"|"false" STRING),
+disabled-reason (omitted when null), as-of-ms. Tool-pool headers: x-nous-tool-pool-micros,
+x-nous-tool-pool-gated-off ("true"|"false" STRING). Money is micros ints only;
+``*_usd`` strings are preserved verbatim (never re-parsed to float).
"""
from __future__ import annotations
@@ -29,7 +17,7 @@ import os
import re
import time
from dataclasses import dataclass
-from typing import Any, Mapping, Optional
+from typing import Any, Callable, Mapping, Optional
from utils import is_truthy_value
@@ -37,14 +25,10 @@ logger = logging.getLogger(__name__)
# Warn-once latch: emit the version-unsupported warning at most once per process.
_version_warning_emitted: bool = False
-
-# Valid denominator kinds (exhaustive set from the API contract).
_VALID_DENOMINATOR_KINDS = frozenset({"subscription_cap", "none"})
-
# USD format: optional leading minus, one-or-more digits, dot, exactly 2 digits.
_USD_RE = re.compile(r"^-?\d+\.\d{2}$")
-
-_SENTINEL = object() # singleton sentinel for "parse failed"
+_SENTINEL = object() # "parse failed"
def _safe_int(value: Any) -> Any:
@@ -59,7 +43,6 @@ def _safe_int(value: Any) -> Any:
def _validate_usd(value: Optional[str]) -> bool:
- """Return True iff value is a non-None string matching ^-?\\d+\\.\\d{2}$."""
return value is not None and bool(_USD_RE.match(value))
@@ -96,13 +79,13 @@ class CreditsState:
@property
def depleted(self) -> bool:
- """Keyed off ``paid_access == False`` ONLY — never ``remaining_micros == 0``,
- a false positive when the balance is zero but access is live (renewal pending)."""
+ """Keyed off ``paid_access == False`` ONLY — ``remaining_micros == 0`` is a
+ false positive when the balance is zero but access is live (renewal pending)."""
return not self.paid_access
@property
def used_fraction(self) -> Optional[float]:
- """Fraction of the subscription cap consumed, in [0.0, 1.0]; None without a
+ """Fraction of the subscription cap consumed in [0.0, 1.0]; None without a
computable denominator. Guarded on the LIMIT FIELD (the real denominator),
not ``denominator_kind`` (metadata)."""
if not isinstance(self.subscription_limit_micros, int) or self.subscription_limit_micros <= 0:
@@ -118,7 +101,7 @@ CREDITS_RESTORED_TTL_MS = 8000 # the only TTL notice in v1 (depletion-recove
# Usage-gauge bands (ascending): (threshold_fraction, level, label_pct). One
# escalating line showing the HIGHEST band reached (50 → 75 → 90); crossing up
-# replaces it, recovering steps it down. The policy derives everything from it.
+# replaces it, recovering steps it down.
CREDITS_USAGE_BANDS: tuple[tuple[float, str, int], ...] = ((0.50, "info", 50), (0.75, "warn", 75), (0.90, "warn", 90))
CREDITS_USAGE_KEY = "credits.usage" # single key for the escalating usage notice
@@ -155,14 +138,13 @@ def _sticky_notice(text: str, level: str, key: str) -> AgentNotice:
def is_free_tier_model(model: str, base_url: str = "") -> bool:
- """Return True when *model* is a Nous free-tier model, using ONLY local data.
+ """True when *model* is a Nous free-tier model, using ONLY local data.
Zero-network signals: (1) ``:free`` suffix — canonical Nous free SKU marker;
(2) ``stealth/`` prefix — stealth-preview SKUs are free without the suffix
(naming-convention trust: a PAID ``stealth/`` model would wrongly suppress
the banner); (3) a PEEK into ``hermes_cli.models``' pricing cache (filled by
- the model picker; a miss never fetches — gateway sessions never run the
- picker, so there only 1-2 apply).
+ the model picker; a miss never fetches — gateway sessions never run the picker).
Fail-open to False (depleted notice still shows): a wrong warning is
recoverable noise; hiding it on a paid model masks a real block.
@@ -178,13 +160,26 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool:
# peek_cached_pricing owns the /v1-suffix and auth-state key details.
pricing = peek_cached_pricing(base_url)
- if not pricing:
- return False
- return _is_model_free(model, pricing)
+ return bool(pricing) and _is_model_free(model, pricing)
except Exception:
return False
+def _sync_notice(
+ key: str, want: bool, make: Callable[[], AgentNotice], active: set, to_show: list, to_clear: list
+) -> Optional[str]:
+ """Reconcile one keyed sticky notice; returns ``"shown"``, ``"cleared"`` or None."""
+ if want and key not in active:
+ to_show.append(make())
+ active.add(key)
+ return "shown"
+ if key in active and not want:
+ to_clear.append(key)
+ active.discard(key)
+ return "cleared"
+ return None
+
+
def evaluate_credits_notices(
state: CreditsState, latch: dict, *, model_is_free: bool = False,
) -> tuple[list[AgentNotice], list[str]]:
@@ -201,43 +196,35 @@ def evaluate_credits_notices(
to_show: list[AgentNotice] = []
to_clear: list[str] = []
uf = state.used_fraction
+ active = latch["active"]
# Crossing latch: band notices fire only once uf was observed below the LOWEST
# band, so a session opening mid-range doesn't fire on its first observation
# (the cold-start seed primes this when it WANTS an open-high warning).
if uf is not None and uf < CREDITS_USAGE_BANDS[0][0]:
latch["seen_below_90"] = True
-
# Grant-spent gate: fires only after this session OBSERVED the grant unspent
# (≥1¢). Opening at grant-spent is a steady STATE (/usage carries it), not an
# event. Unlike seen_below_90, seeds must NOT prime this gate.
if uf is not None and uf < 1.0 and state.subscription_micros >= GRANT_UNSPENT_MIN_MICROS:
latch["seen_grant_unspent"] = True
- active = latch["active"]
# Highest band reached (ascending → last match wins); None below all.
+ # Top-up suppression: with purchased credits the cap gauge is the wrong
+ # denominator ("90% used" on $50 of top-up is noise; it used to stick
+ # PERMANENTLY beside grant_spent at >=100%). grant_spent covers the
+ # cap-reached case; a mid-session top-up flips current_band → None and the
+ # clear path removes the band line.
current_band: Optional[tuple[float, str, int]] = None
- if uf is not None:
+ if uf is not None and state.purchased_micros <= 0:
for band in CREDITS_USAGE_BANDS:
if uf >= band[0]:
current_band = band
- # Top-up suppression: with purchased credits the cap gauge is the wrong
- # denominator ("90% used" on $50 of top-up is noise; it used to stick
- # PERMANENTLY beside grant_spent at >=100%). grant_spent below covers the
- # cap-reached case; a mid-session top-up flips current_band → None and the
- # clear path removes the band line.
- if state.purchased_micros > 0:
- current_band = None
- grant_cond = (
- state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
- )
- depleted_cond = not state.paid_access
# ── usage gauge: highest crossed band only; replace on band change (climb or
# step-down); clear below the lowest band or when the denominator vanishes.
- shown_band = latch.get("usage_band")
target_band = current_band[2] if (current_band and latch["seen_below_90"]) else None
- if target_band != shown_band:
+ if target_band != latch.get("usage_band"):
if CREDITS_USAGE_KEY in active:
to_clear.append(CREDITS_USAGE_KEY)
active.discard(CREDITS_USAGE_KEY)
@@ -245,15 +232,13 @@ def evaluate_credits_notices(
# Absolute dollars used (a bare "N%" is only meaningful against a Nous
# cap): cap − remaining in micros, clamped [0, cap]; "$?" if a producer
# set the limit without its *_usd. Re-emits on band change only.
- _cap_usd = state.subscription_limit_usd or "?"
- _level = current_band[1] # type: ignore[index] (current_band set when target_band set)
- _lim = state.subscription_limit_micros or 0
- _used_micros = max(0, min(_lim, _lim - state.subscription_micros))
- _used_usd = f"{_used_micros / 1_000_000:.2f}" if _lim else "?"
- _glyph = "⚠" if _level == "warn" else "•"
- to_show.append(
- _sticky_notice(f"{_glyph} You've used ${_used_usd} of your ${_cap_usd} cap", _level, CREDITS_USAGE_KEY)
- )
+ level = current_band[1] # type: ignore[index] (current_band set when target_band set)
+ lim = state.subscription_limit_micros or 0
+ used_usd = f"{max(0, min(lim, lim - state.subscription_micros)) / 1_000_000:.2f}" if lim else "?"
+ glyph = "⚠" if level == "warn" else "•"
+ to_show.append(_sticky_notice(
+ f"{glyph} You've used ${used_usd} of your ${state.subscription_limit_usd or '?'} cap", level, CREDITS_USAGE_KEY
+ ))
active.add(CREDITS_USAGE_KEY)
latch["usage_band"] = target_band
@@ -261,52 +246,62 @@ def evaluate_credits_notices(
# announcement per crossing. A header flicker (uf → None → 1.0) clears the
# line but cannot re-announce; only a renewal re-opening the gate (fresh ≥1¢
# observation) arms the next. .get(): default closed for hand-built latches.
- if grant_cond and "credits.grant_spent" not in active and latch.get("seen_grant_unspent", False):
- to_show.append(
- _sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent")
- )
- active.add("credits.grant_spent")
+ grant_cond = (
+ state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
+ )
+ if _sync_notice(
+ "credits.grant_spent", grant_cond and latch.get("seen_grant_unspent", False),
+ lambda: _sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent"),
+ active, to_show, to_clear,
+ ) == "shown":
latch["seen_grant_unspent"] = False
- elif "credits.grant_spent" in active and not grant_cond:
- to_clear.append("credits.grant_spent")
- active.discard("credits.grant_spent")
# ── depleted: suppressed while the model is free (inference still works).
- show_depleted = depleted_cond and not model_is_free
- if show_depleted and "credits.depleted" not in active:
- to_show.append(_sticky_notice("✕ Credit access paused · run /topup to top up", "error", "credits.depleted"))
- active.add("credits.depleted")
- elif "credits.depleted" in active and not show_depleted:
- to_clear.append("credits.depleted")
- active.discard("credits.depleted")
- if not depleted_cond:
- # Genuine recovery only — switching to a free model while still
- # depleted must NOT claim access was restored.
- to_show.append(AgentNotice(
- text="✓ Credit access restored", level="success", kind="ttl",
- ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored",
- ))
+ depleted_cond = not state.paid_access
+ if _sync_notice(
+ "credits.depleted", depleted_cond and not model_is_free,
+ lambda: _sticky_notice("✕ Credit access paused · run /topup to top up", "error", "credits.depleted"),
+ active, to_show, to_clear,
+ ) == "cleared" and not depleted_cond:
+ # Genuine recovery only — switching to a free model while still
+ # depleted must NOT claim access was restored.
+ to_show.append(AgentNotice(
+ text="✓ Credit access restored", level="success", kind="ttl",
+ ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored",
+ ))
return (to_show, to_clear)
-# (field, header, signed) — required micros fields; only subscription may be negative.
-_MICROS_FIELDS: tuple[tuple[str, str, bool], ...] = (
- ("remaining_micros", "x-nous-credits-remaining-micros", False),
- ("subscription_micros", "x-nous-credits-subscription-micros", True),
- ("rollover_micros", "x-nous-credits-rollover-micros", False),
- ("purchased_micros", "x-nous-credits-purchased-micros", False), ("as_of_ms", "x-nous-credits-as-of-ms", False),
-)
-_USD_FIELDS: tuple[tuple[str, str], ...] = (
- ("remaining_usd", "x-nous-credits-remaining-usd"), ("subscription_usd", "x-nous-credits-subscription-usd"),
- ("purchased_usd", "x-nous-credits-purchased-usd"),
-)
-# (field, header, default-when-absent) — "true"/"false" (case-insensitive) STRING flags.
-_BOOL_FIELDS: tuple[tuple[str, str, bool], ...] = (
- ("paid_access", "x-nous-credits-paid-access", True), # absent → fail-open (assume access)
- ("tool_pool_gated_off", "x-nous-tool-pool-gated-off", False),
+# Header field table: (field, header, kind). micros: required, non-negative unless
+# "signed" (only subscription may be negative). usd: required ^-?\d+\.\d{2}$.
+# bool: optional "true"/"false" STRING flag; the tuple's 4th item is the default.
+_HEADER_FIELDS: tuple[tuple, ...] = (
+ ("remaining_micros", "x-nous-credits-remaining-micros", "micros"),
+ ("subscription_micros", "x-nous-credits-subscription-micros", "signed"),
+ ("rollover_micros", "x-nous-credits-rollover-micros", "micros"),
+ ("purchased_micros", "x-nous-credits-purchased-micros", "micros"),
+ ("as_of_ms", "x-nous-credits-as-of-ms", "micros"),
+ ("remaining_usd", "x-nous-credits-remaining-usd", "usd"),
+ ("subscription_usd", "x-nous-credits-subscription-usd", "usd"),
+ ("purchased_usd", "x-nous-credits-purchased-usd", "usd"),
+ ("paid_access", "x-nous-credits-paid-access", "bool", True), # absent → fail-open (assume access)
+ ("tool_pool_gated_off", "x-nous-tool-pool-gated-off", "bool", False),
)
+def _parse_field(kind: str, raw: Optional[str], default: Any = None) -> Any:
+ """One header value → field value, or ``_SENTINEL`` on a contract violation."""
+ if kind in ("micros", "signed"):
+ val = _safe_int(raw)
+ return _SENTINEL if val is _SENTINEL or (kind == "micros" and val < 0) else val
+ if kind == "usd":
+ return raw if _validate_usd(raw) else _SENTINEL
+ if raw is None:
+ return default
+ flag = raw.strip().lower()
+ return _SENTINEL if flag not in ("true", "false") else flag == "true"
+
+
def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Optional[CreditsState]:
"""Parse x-nous-credits-* (and x-nous-tool-pool-*) headers into a CreditsState.
@@ -336,45 +331,29 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt
logger.warning("credits header version %d unsupported, ignoring — update Hermes", version_val)
return None
fields: dict[str, Any] = {}
- for name, key, signed in _MICROS_FIELDS:
- val = _safe_int(lowered.get(key))
- if val is _SENTINEL or (not signed and val < 0):
+ for name, key, kind, *default in _HEADER_FIELDS:
+ val = _parse_field(kind, lowered.get(key), *default)
+ if val is _SENTINEL:
return None
fields[name] = val
-
# tool_pool_micros is OPTIONAL: absent → 0; present-but-invalid → miss.
- _tp_raw = lowered.get("x-nous-tool-pool-micros")
- _tp_val = 0 if _tp_raw is None else _safe_int(_tp_raw)
- if _tp_val is _SENTINEL or _tp_val < 0:
+ tp_raw = lowered.get("x-nous-tool-pool-micros")
+ tp_val = 0 if tp_raw is None else _parse_field("micros", tp_raw)
+ if tp_val is _SENTINEL:
return None
- fields["tool_pool_micros"] = _tp_val
- for name, key in _USD_FIELDS:
- val = lowered.get(key, "")
- if not _validate_usd(val):
- return None
- fields[name] = val
-
+ fields["tool_pool_micros"] = tp_val
# subscription_limit_* PAIRED + OPTIONAL: both present → validate both
# (any invalid → miss); half-pair or both absent → both None, parse continues.
- sub_limit_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
- sub_limit_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
- if sub_limit_micros_raw is not None and sub_limit_usd_raw is not None:
- lm = _safe_int(sub_limit_micros_raw)
- if lm is _SENTINEL or lm < 0 or not _validate_usd(sub_limit_usd_raw):
+ lim_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
+ lim_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
+ if lim_micros_raw is not None and lim_usd_raw is not None:
+ lm = _parse_field("micros", lim_micros_raw)
+ if lm is _SENTINEL or not _validate_usd(lim_usd_raw):
return None
- fields["subscription_limit_micros"] = lm
- fields["subscription_limit_usd"] = sub_limit_usd_raw
+ fields["subscription_limit_micros"], fields["subscription_limit_usd"] = lm, lim_usd_raw
denominator_kind = lowered.get("x-nous-credits-denominator-kind", "none")
if denominator_kind not in _VALID_DENOMINATOR_KINDS:
return None
- for name, key, default in _BOOL_FIELDS:
- if key not in lowered:
- fields[name] = default
- continue
- raw = lowered[key].strip().lower()
- if raw not in ("true", "false"):
- return None
- fields[name] = raw == "true"
return CreditsState(
version=version_val,
denominator_kind=denominator_kind,
@@ -426,21 +405,18 @@ _DEV_FIXTURES: dict[str, dict] = {
def dev_fixture_credits_state() -> Optional[CreditsState]:
- """Return a fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None
- (unknown name / "clear" / "none" / unset → None).
-
- Hard prod-leak guard: applies ONLY when HERMES_DEV_CREDITS is also on, so a
- stray fixture env var can never surface fabricated balances on a real account.
- """
+ """Fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None (unknown name /
+ "clear" / "none" / unset). Hard prod-leak guard: applies ONLY when
+ HERMES_DEV_CREDITS is also on, so a stray fixture env var can never surface
+ fabricated balances on a real account."""
if not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
return None
- raw = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
- if not raw:
+ name = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
+ if not name:
return None
- name = raw
- if os.path.sep in raw or "/" in raw: # looks like a path → read the name from the file
+ if os.path.sep in name or "/" in name: # looks like a path → read the name from the file
try:
- with open(raw, "r", encoding="utf-8") as fh:
+ with open(name, "r", encoding="utf-8") as fh:
name = fh.read().strip()
except OSError:
return None
@@ -450,8 +426,7 @@ def dev_fixture_credits_state() -> Optional[CreditsState]:
# Stamp what the REAL parser always guarantees so a fixture is field-identical
# to a parse_credits_headers() result (differential test): version 1, and a
# valid purchased_usd (a zero-top-up account still carries "0.00").
- merged = {"version": 1, "purchased_usd": "0.00", **spec}
- return CreditsState(**merged, from_header=True, captured_at=time.time())
+ return CreditsState(**{"version": 1, "purchased_usd": "0.00", **spec}, from_header=True, captured_at=time.time())
def _credits_state_from_account(info) -> Optional[CreditsState]:
@@ -459,26 +434,26 @@ def _credits_state_from_account(info) -> Optional[CreditsState]:
Float account dollars → micros plus a DISPLAY *_usd (formatting account floats
is allowed; parsing a server *_usd is not). Fail-open → None."""
try:
- _acc = getattr(info, "paid_service_access_info", None)
- _sub = getattr(info, "subscription", None)
+ acc = getattr(info, "paid_service_access_info", None)
+ sub = getattr(info, "subscription", None)
def _money(dollars) -> tuple[int, str]: # (micros, display usd); (0, "") when absent
if isinstance(dollars, (int, float)):
return int(round(dollars * 1_000_000)), f"{dollars:.2f}"
return 0, ""
- _remaining = _money(getattr(_acc, "total_usable_credits", None))
- _sub_rem = _money(getattr(_acc, "subscription_credits_remaining", None))
- _purchased = _money(getattr(_acc, "purchased_credits_remaining", None))
- _monthly = getattr(_sub, "monthly_credits", None)
- _cap = _money(_monthly) if isinstance(_monthly, (int, float)) and _monthly > 0 else (None, None)
- _paid = getattr(info, "paid_service_access", None)
+ remaining = _money(getattr(acc, "total_usable_credits", None))
+ sub_rem = _money(getattr(acc, "subscription_credits_remaining", None))
+ purchased = _money(getattr(acc, "purchased_credits_remaining", None))
+ monthly = getattr(sub, "monthly_credits", None)
+ cap = _money(monthly) if isinstance(monthly, (int, float)) and monthly > 0 else (None, None)
+ paid = getattr(info, "paid_service_access", None)
return CreditsState(
- remaining_micros=_remaining[0], remaining_usd=_remaining[1], subscription_micros=_sub_rem[0],
- subscription_usd=_sub_rem[1], subscription_limit_micros=_cap[0], subscription_limit_usd=_cap[1],
- purchased_micros=_purchased[0], purchased_usd=_purchased[1],
- rollover_micros=_money(getattr(_sub, "rollover_credits", None))[0],
- denominator_kind="subscription_cap" if _cap[0] is not None else "none",
- paid_access=_paid if isinstance(_paid, bool) else True, from_header=False, captured_at=time.time(),
+ remaining_micros=remaining[0], remaining_usd=remaining[1], subscription_micros=sub_rem[0],
+ subscription_usd=sub_rem[1], subscription_limit_micros=cap[0], subscription_limit_usd=cap[1],
+ purchased_micros=purchased[0], purchased_usd=purchased[1],
+ rollover_micros=_money(getattr(sub, "rollover_credits", None))[0],
+ denominator_kind="subscription_cap" if cap[0] is not None else "none",
+ paid_access=paid if isinstance(paid, bool) else True, from_header=False, captured_at=time.time(),
)
except Exception:
logger.debug("credits ▸ seed account→state mapping failed", exc_info=True)
@@ -492,11 +467,11 @@ def _hydrate_seed_state(agent, state) -> None:
agent._credits_state = state
if getattr(agent, "_credits_session_start_micros", None) is None:
agent._credits_session_start_micros = state.remaining_micros
- _latch = getattr(agent, "_credits_latch", None)
- if isinstance(_latch, dict) and state.used_fraction is not None:
+ latch = getattr(agent, "_credits_latch", None)
+ if isinstance(latch, dict) and state.used_fraction is not None:
# Prime ONLY seen_below_90. Never prime seen_grant_unspent: a seed
# observing grant-spent is a steady state; priming revives the nag.
- _latch["seen_below_90"] = True
+ latch["seen_below_90"] = True
emit = getattr(agent, "_emit_credits_notices", None)
if callable(emit):
emit()
@@ -511,9 +486,7 @@ def seed_credits_at_session_start(agent) -> bool:
Returns True iff it seeded this call. Never raises — credits must never block startup.
"""
try:
- if getattr(agent, "provider", "") != "nous":
- return False
- if getattr(agent, "_credits_state", None) is not None:
+ if getattr(agent, "provider", "") != "nous" or getattr(agent, "_credits_state", None) is not None:
return False
try:
fixture = dev_fixture_credits_state()
diff --git a/agent/fast_mode.py b/agent/fast_mode.py
index b258b00d89..042c63abf5 100644
--- a/agent/fast_mode.py
+++ b/agent/fast_mode.py
@@ -1,12 +1,11 @@
"""Bounded fast-mode windows (``/fast auto`` and ``/fast cold``).
-``agent.service_tier`` is ``None`` (normal), ``"priority"`` (static fast, pinned
-into ``agent.request_overrides`` at build time), ``"auto"`` (every user turn opens
-a window of ``agent.fast_auto_seconds``) or ``"cold"`` (only a session's first
-turn, no prior history, opens it). The provider's fast override is layered onto
-the request kwargs only while the window is open. Only per-request params
-(``service_tier`` / ``speed``) vary — system prompt, tools, and messages are
-untouched, so the prompt cache survives the boundary.
+``agent.service_tier``: ``None`` (normal), ``"priority"`` (static fast, pinned into
+``agent.request_overrides`` at build time), ``"auto"`` (every user turn opens a
+window of ``agent.fast_auto_seconds``) or ``"cold"`` (only a session's first turn,
+no prior history, opens it). The provider's fast override is layered onto request
+kwargs only while the window is open; only per-request params (``service_tier`` /
+``speed``) vary, so the prompt cache survives the boundary.
"""
from __future__ import annotations
@@ -39,17 +38,13 @@ def begin_turn(agent: Any, conversation_history: Any) -> None:
def effective_request_overrides(agent: Any) -> dict[str, Any]:
"""``agent.request_overrides`` plus the fast override while the window is open."""
overrides = dict(getattr(agent, "request_overrides", None) or {})
- if getattr(agent, "service_tier", None) not in BOUNDED_MODES or time.monotonic() >= getattr(
- agent, "_fast_until", 0.0
- ):
+ if getattr(agent, "service_tier", None) not in BOUNDED_MODES or time.monotonic() >= getattr(agent, "_fast_until", 0.0):
return overrides
from hermes_cli.models import resolve_fast_mode_overrides
base_url = getattr(agent, "base_url", None)
if getattr(agent, "api_mode", None) == "anthropic_messages":
base_url = getattr(agent, "_anthropic_base_url", None) or base_url
- fast = resolve_fast_mode_overrides(
- getattr(agent, "model", None), provider=getattr(agent, "provider", None), base_url=base_url
+ overrides.update(
+ resolve_fast_mode_overrides(getattr(agent, "model", None), provider=getattr(agent, "provider", None), base_url=base_url) or {}
)
- if fast:
- overrides.update(fast)
return overrides
diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py
index 31871509c7..4a1a689f20 100644
--- a/agent/gemini_native_adapter.py
+++ b/agent/gemini_native_adapter.py
@@ -1,11 +1,10 @@
"""OpenAI-compatible facade over Google AI Studio's native Gemini API.
-Hermes keeps ``api_mode='chat_completions'`` for the ``gemini`` provider so the
-agent loop keeps its OpenAI-shaped message flow; this shim converts those
-``messages[]`` / ``tools[]`` requests into ``models/{model}:generateContent``
-payloads and converts the responses back. Google's OpenAI-compat endpoint has
-been brittle for the multi-turn tool loop (auth churn, tool-call replay quirks,
-thought-signature requirements); the native API is the canonical path.
+The ``gemini`` provider keeps ``api_mode='chat_completions'`` so the agent loop
+stays OpenAI-shaped; this shim converts ``messages[]``/``tools[]`` requests into
+``models/{model}:generateContent`` payloads and converts responses back. Google's
+OpenAI-compat endpoint is brittle for the multi-turn tool loop (auth churn,
+tool-call replay quirks, thought-signature requirements); native is canonical.
"""
from __future__ import annotations
@@ -41,36 +40,54 @@ DEFAULT_GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
# Gemini's native API applies a low internal default and truncates output.
GEMINI_DEFAULT_MAX_OUTPUT_TOKENS = 65535
+_FREE_TIER_GUIDANCE = (
+ "\n\nYour Google API key is on the free tier (a few hundred requests/day for Gemini Flash models). "
+ "Hermes typically makes 3-10 API calls per user turn, so the free tier is exhausted in a handful of "
+ "messages and cannot sustain an agent session. Enable billing on your Google Cloud project and "
+ "regenerate the key in a billing-enabled project: https://aistudio.google.com/apikey"
+)
+_STANDARD_KEY_GUIDANCE = (
+ "\n\nGoogle Gemini rejected this API key's type — you do NOT need OAuth. Google began rejecting legacy "
+ "'Standard' Google Cloud keys for the Gemini API on June 19, 2026, and all Standard keys stop working in "
+ "September 2026. Open https://aistudio.google.com/api-keys, check the key's type and status, and create a "
+ "replacement Gemini API key (or, as a temporary bridge, restrict the Standard key to "
+ "generativelanguage.googleapis.com). Then update GEMINI_API_KEY / GOOGLE_API_KEY in ~/.hermes/.env and "
+ "restart your session. Details: https://ai.google.dev/gemini-api/docs/api-key"
+)
+# Stands in for a model turn that never arrived (stream failure / interrupt /
+# quota fallback) when a human user text turn directly follows a tool-result
+# turn, keeping the request alternation-valid while the user's message stays a
+# turn of its own (mirrors gemini-cli's placeholder repair).
+_INTERRUPTED_RESPONSE_PLACEHOLDER = "[The previous response was interrupted before it completed.]"
+# Cross-provider tool_calls (e.g. fallback from xAI/Anthropic) carry no Gemini
+# thoughtSignature; without this sentinel Gemini 3 thinking models reject
+# replayed history with 400 INVALID_ARGUMENT.
+_SKIP_SIGNATURE = "skip_thought_signature_validator"
+_TOOL_CHOICE_MODES = {"auto": "AUTO", "required": "ANY", "none": "NONE"}
+_FINISH_REASON_MAP = {
+ "STOP": "stop", "MAX_TOKENS": "length", "SAFETY": "content_filter", "RECITATION": "content_filter", "OTHER": "stop",
+}
+_HTTP_ERROR_CODES = {401: "gemini_unauthorized", 429: "gemini_rate_limited", 404: "gemini_model_not_found"}
+
def bare_gemini_model_id(model: str) -> str:
"""Strip Gemini's own provider prefix from an aggregator-style model id."""
name = (model or "").strip()
- lowered = name.lower()
for prefix in ("google/", "gemini/"):
- if lowered.startswith(prefix):
+ if name.lower().startswith(prefix):
return name[len(prefix):].strip() or name
return name
-def _gemini_major_version(model: str) -> Optional[int]:
- """Extract the major version from a Gemini model id (``gemini-3.6-flash`` → 3)."""
- match = re.match(r"gemini-(\d+)", bare_gemini_model_id(model).lower())
- return int(match.group(1)) if match else None
-
-
def gemini_requires_tool_call_ids(model: str) -> bool:
- """Whether functionCall/functionResponse parts must carry explicit ids.
-
- Gemini 3+ needs explicit ids so replayed parallel tool calls pair with
- their responses; Gemini 2.x rejects unexpected ``id`` fields, so gate on
- the major version.
- """
- version = _gemini_major_version(model)
- return version is not None and version >= 3
+ """Gemini 3+ needs explicit functionCall/functionResponse ids so replayed
+ parallel tool calls pair with their responses; 2.x rejects the field."""
+ match = re.match(r"gemini-(\d+)", bare_gemini_model_id(model).lower())
+ return match is not None and int(match.group(1)) >= 3
def is_native_gemini_base_url(base_url: str) -> bool:
- """Return True when the endpoint speaks Gemini's native REST API."""
+ """True when the endpoint speaks Gemini's native REST API (not ``/openai``)."""
normalized = str(base_url or "").strip().rstrip("/").lower()
return "generativelanguage.googleapis.com" in normalized and not normalized.endswith("/openai")
@@ -78,23 +95,19 @@ def is_native_gemini_base_url(base_url: str) -> bool:
def probe_gemini_tier(
api_key: str, base_url: str = DEFAULT_GEMINI_BASE_URL, *, model: str = "gemini-3.7-flash", timeout: float = 10.0
) -> str:
- """Probe a Google AI Studio API key; return ``"free"``, ``"paid"`` or
- ``"unknown"`` (probe failed — callers should proceed without blocking)."""
+ """Probe a Google AI Studio key; return ``"free"``, ``"paid"`` or ``"unknown"``
+ (probe failed — callers should proceed without blocking)."""
key = (api_key or "").strip()
if not key:
return "unknown"
-
- normalized_base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL
- if normalized_base.lower().endswith("/openai"):
- normalized_base = normalized_base[: -len("/openai")]
-
+ base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL
+ if base.lower().endswith("/openai"):
+ base = base[: -len("/openai")]
payload = {"contents": [{"role": "user", "parts": [{"text": "hi"}]}], "generationConfig": {"maxOutputTokens": 1}}
try:
with httpx.Client(timeout=timeout) as client:
resp = client.post(
- f"{normalized_base}/models/{model}:generateContent",
- params={"key": key},
- json=payload,
+ f"{base}/models/{model}:generateContent", params={"key": key}, json=payload,
headers={"Content-Type": "application/json", "X-Goog-Api-Client": f"hermes-agent/{_HERMES_VERSION}"},
)
except Exception as exc:
@@ -108,52 +121,32 @@ def probe_gemini_tier(
return "free" if int(rpd_header) <= 1000 else "paid"
except (TypeError, ValueError):
pass
-
if resp.status_code == 429:
- try:
- body_text = resp.text or ""
- except Exception:
- body_text = ""
- return "free" if "free_tier" in body_text.lower() else "paid"
+ return "free" if "free_tier" in _response_text(resp).lower() else "paid"
return "paid" if 200 <= resp.status_code < 300 else "unknown"
+def _response_text(response: Any) -> str:
+ try:
+ return response.text or ""
+ except Exception:
+ return ""
+
+
def is_free_tier_quota_error(error_message: str) -> bool:
- """Return True when a Gemini 429 message indicates free-tier exhaustion."""
+ """True when a Gemini 429 message indicates free-tier exhaustion."""
return bool(error_message) and "free_tier" in error_message.lower()
-_FREE_TIER_GUIDANCE = (
- "\n\nYour Google API key is on the free tier (a few hundred requests/day for Gemini Flash models). "
- "Hermes typically makes 3-10 API calls per user turn, so the free tier is exhausted in a handful of "
- "messages and cannot sustain an agent session. Enable billing on your Google Cloud project and "
- "regenerate the key in a billing-enabled project: https://aistudio.google.com/apikey"
-)
-
-
def is_standard_key_auth_error(status: int, error_message: str, reason: str = "") -> bool:
- """Return True when a Gemini 401 indicates Google rejected the key TYPE.
-
- Google rejects legacy "Standard" Google Cloud API keys with a misleading
- 401 asking for an OAuth 2 access token, optionally carrying ErrorInfo
- reason ``ACCESS_TOKEN_TYPE_UNSUPPORTED``. Scoped narrowly so a plain bad
- key (``API_KEY_INVALID``) keeps its existing message.
- """
+ """True when a Gemini 401 means Google rejected the key TYPE (legacy "Standard"
+ Cloud key → misleading "expected OAuth 2 access token" / ErrorInfo
+ ``ACCESS_TOKEN_TYPE_UNSUPPORTED``). Narrow so ``API_KEY_INVALID`` keeps its message."""
if status != 401:
return False
return reason == "ACCESS_TOKEN_TYPE_UNSUPPORTED" or "expected oauth 2 access token" in (error_message or "").lower()
-_STANDARD_KEY_GUIDANCE = (
- "\n\nGoogle Gemini rejected this API key's type — you do NOT need OAuth. Google began rejecting legacy "
- "'Standard' Google Cloud keys for the Gemini API on June 19, 2026, and all Standard keys stop working in "
- "September 2026. Open https://aistudio.google.com/api-keys, check the key's type and status, and create a "
- "replacement Gemini API key (or, as a temporary bridge, restrict the Standard key to "
- "generativelanguage.googleapis.com). Then update GEMINI_API_KEY / GOOGLE_API_KEY in ~/.hermes/.env and "
- "restart your session. Details: https://ai.google.dev/gemini-api/docs/api-key"
-)
-
-
class GeminiAPIError(Exception):
"""Error shape compatible with Hermes retry/error classification."""
@@ -163,26 +156,23 @@ class GeminiAPIError(Exception):
details: Optional[Dict[str, Any]] = None,
) -> None:
super().__init__(message)
- self.code = code
- self.status_code = status_code
- self.response = response
- self.retry_after = retry_after
- self.details = details or {}
+ self.code, self.status_code, self.response = code, status_code, response
+ self.retry_after, self.details = retry_after, details or {}
+# ── OpenAI → Gemini request translation ──────────────────────────────────────
+
def _coerce_content_to_text(content: Any) -> str:
if content is None:
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
- pieces: List[str] = []
- for part in content:
- if isinstance(part, str):
- pieces.append(part)
- elif isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
- pieces.append(part["text"])
- return "\n".join(pieces)
+ return "\n".join(
+ p if isinstance(p, str) else p["text"]
+ for p in content
+ if isinstance(p, str) or (isinstance(p, dict) and p.get("type") == "text" and isinstance(p.get("text"), str))
+ )
return str(content)
@@ -190,21 +180,17 @@ def _extract_multimodal_parts(content: Any) -> List[Dict[str, Any]]:
if not isinstance(content, list):
text = _coerce_content_to_text(content)
return [{"text": text}] if text else []
-
parts: List[Dict[str, Any]] = []
for item in content:
if isinstance(item, str):
parts.append({"text": item})
+ elif not isinstance(item, dict):
continue
- if not isinstance(item, dict):
- continue
- ptype = item.get("type")
- if ptype == "text":
- text = item.get("text")
- if isinstance(text, str) and text:
- parts.append({"text": text})
- elif ptype == "image_url":
- url = ((item.get("image_url") or {}).get("url") or "")
+ elif item.get("type") == "text":
+ if isinstance(item.get("text"), str) and item["text"]:
+ parts.append({"text": item["text"]})
+ elif item.get("type") == "image_url":
+ url = (item.get("image_url") or {}).get("url") or ""
if not isinstance(url, str) or not url.startswith("data:"):
continue
try:
@@ -226,13 +212,6 @@ def _tool_call_extra_signature(tool_call: Dict[str, Any]) -> Optional[str]:
return sig if isinstance(sig, str) and sig else None
-# Stands in for a model turn that never arrived (stream failure / interrupt /
-# quota fallback) when history leaves a human user text turn directly after a
-# tool-result turn, keeping the request alternation-valid while the user's
-# message remains a turn of its own (mirrors gemini-cli's placeholder repair).
-_INTERRUPTED_RESPONSE_PLACEHOLDER = "[The previous response was interrupted before it completed.]"
-
-
def _tool_call_id(tool_call: Dict[str, Any]) -> str:
return str(tool_call.get("id") or tool_call.get("call_id") or "")
@@ -246,25 +225,19 @@ def _translate_tool_call_to_gemini(tool_call: Dict[str, Any], include_ids: bool
args = {"_raw": args_raw}
if not isinstance(args, dict):
args = {"_value": args}
-
- part: Dict[str, Any] = {"functionCall": {"name": str(fn.get("name") or ""), "args": args}}
+ call: Dict[str, Any] = {"name": str(fn.get("name") or ""), "args": args}
if include_ids and _tool_call_id(tool_call):
- part["functionCall"]["id"] = _tool_call_id(tool_call)
- # Cross-provider tool_calls (e.g. fallback from xAI/Anthropic) carry no
- # Gemini thoughtSignature; without the sentinel, Gemini 3 thinking models
- # reject replayed history with 400 INVALID_ARGUMENT.
- part["thoughtSignature"] = _tool_call_extra_signature(tool_call) or "skip_thought_signature_validator"
- return part
+ call["id"] = _tool_call_id(tool_call)
+ return {"functionCall": call, "thoughtSignature": _tool_call_extra_signature(tool_call) or _SKIP_SIGNATURE}
def _looks_like_json_schema(node: Any) -> bool:
- """True if a parsed value contains a JSON-Schema-style ``$ref`` pointer (``#/...``).
+ """True if a parsed value contains a JSON-Schema ``$ref`` pointer (``#/...``).
- Gemini 3 resolves ``$ref``/``$defs`` inside a functionResponse.response
- payload and rejects unknown pointers with HTTP 400, so a tool result that
- is itself a JSON Schema (e.g. ``tool_describe`` output) must be forwarded
- as opaque text. Detection is structural: false positives only lose the
- structured shape, never the content.
+ Gemini 3 resolves ``$ref``/``$defs`` inside functionResponse.response and
+ rejects unknown pointers with HTTP 400, so a tool result that is itself a
+ JSON Schema (e.g. ``tool_describe`` output) must be forwarded as opaque
+ text. False positives only lose the structured shape, never the content.
"""
if isinstance(node, dict):
return any(
@@ -278,12 +251,11 @@ def _translate_tool_result_to_gemini(
message: Dict[str, Any], tool_name_by_call_id: Optional[Dict[str, str]] = None, include_ids: bool = False,
*, is_gemini3: bool = False,
) -> Dict[str, Any]:
- tool_name_by_call_id = tool_name_by_call_id or {}
tool_call_id = str(message.get("tool_call_id") or "")
- # Gemini requires functionResponse.name to echo the matching
- # functionCall.name, so the call-id mapping beats the result's own name
- # (which may be an unwrapped internal name, e.g. an MCP tool via `tool_call`).
- name = str(tool_name_by_call_id.get(tool_call_id) or message.get("name") or tool_call_id or "tool")
+ # functionResponse.name must echo the matching functionCall.name, so the
+ # call-id mapping beats the result's own name (which may be an unwrapped
+ # internal name, e.g. an MCP tool via `tool_call`).
+ name = str((tool_name_by_call_id or {}).get(tool_call_id) or message.get("name") or tool_call_id or "tool")
raw_content = message.get("content")
content = _coerce_content_to_text(raw_content)
try:
@@ -294,8 +266,7 @@ def _translate_tool_result_to_gemini(
function_response: Dict[str, Any] = {"name": name, "response": response}
if include_ids and tool_call_id:
function_response["id"] = tool_call_id
- # Gemini 3.x accepts images inside functionResponse.parts; 2.x rejects
- # the field, so older models get the text-only downgrade.
+ # Gemini 3.x accepts images inside functionResponse.parts; 2.x rejects the field.
if is_gemini3:
image_parts = [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p]
if image_parts:
@@ -318,7 +289,6 @@ def _build_gemini_contents(
if not isinstance(msg, dict):
continue
role = str(msg.get("role") or "user")
-
if role == "system":
system_text_parts.append(_coerce_content_to_text(msg.get("content")))
continue
@@ -334,47 +304,41 @@ def _build_gemini_contents(
if not isinstance(tool_call, dict):
continue
tool_call_id = _tool_call_id(tool_call)
- tool_name = str(((tool_call.get("function") or {}).get("name") or ""))
+ tool_name = str((tool_call.get("function") or {}).get("name") or "")
if tool_call_id and tool_name:
tool_name_by_call_id[tool_call_id] = tool_name
parts.append(_translate_tool_call_to_gemini(tool_call, include_ids=include_tool_call_ids))
-
if parts:
contents.append({"role": "model" if role == "assistant" else "user", "parts": parts})
# Alternation contract for generateContent:
- # 1) Adjacent same-role contents merge (consecutive same-role contents are
- # rejected with HTTP 400 "multiturn requests [must] alternate").
+ # 1) Adjacent same-role contents merge (else HTTP 400 "multiturn requests
+ # [must] alternate").
# 2) Exception: never fuse a human user text turn into a preceding user
# content that only carries functionResponse parts (or vice versa) —
# Gemini 3 accepts the fold but reads the text as a continuation of the
# tool result and returns an empty candidate. Parallel tool results
# (functionResponse + functionResponse) still merge.
- # 3) The split pair is kept API-valid by interposing a placeholder model
- # turn between the functionResponse content and the human text.
- merged_contents: List[Dict[str, Any]] = []
+ # 3) The split pair stays API-valid via an interposed placeholder model turn.
+ merged: List[Dict[str, Any]] = []
for content in contents:
- prev = merged_contents[-1] if merged_contents else None
+ prev = merged[-1] if merged else None
same_role = prev is not None and prev["role"] == content["role"]
if same_role and content["role"] == "user" and _has_function_response(prev) != _has_function_response(content):
same_role = False
- merged_contents.append({"role": "model", "parts": [{"text": _INTERRUPTED_RESPONSE_PLACEHOLDER}]})
-
+ merged.append({"role": "model", "parts": [{"text": _INTERRUPTED_RESPONSE_PLACEHOLDER}]})
if same_role:
- merged_contents[-1]["parts"].extend(content["parts"])
+ merged[-1]["parts"].extend(content["parts"])
else:
- merged_contents.append(content)
+ merged.append(content)
joined_system = "\n".join(part for part in system_text_parts if part).strip()
- system_instruction = {"role": "system", "parts": [{"text": joined_system}]} if joined_system else None
- return merged_contents, system_instruction
+ return merged, ({"role": "system", "parts": [{"text": joined_system}]} if joined_system else None)
def _translate_tools_to_gemini(tools: Any) -> List[Dict[str, Any]]:
- if not isinstance(tools, list):
- return []
declarations: List[Dict[str, Any]] = []
- for tool in tools:
+ for tool in tools if isinstance(tools, list) else []:
fn = (tool.get("function") or {}) if isinstance(tool, dict) else None
if not isinstance(fn, dict) or not (isinstance(fn.get("name"), str) and fn["name"]):
continue
@@ -387,9 +351,6 @@ def _translate_tools_to_gemini(tools: Any) -> List[Dict[str, Any]]:
return [{"functionDeclarations": declarations}] if declarations else []
-_TOOL_CHOICE_MODES = {"auto": "AUTO", "required": "ANY", "none": "NONE"}
-
-
def _translate_tool_choice_to_gemini(tool_choice: Any) -> Optional[Dict[str, Any]]:
if isinstance(tool_choice, str) and tool_choice in _TOOL_CHOICE_MODES:
return {"functionCallingConfig": {"mode": _TOOL_CHOICE_MODES[tool_choice]}}
@@ -417,12 +378,9 @@ def _normalize_thinking_config(config: Any) -> Optional[Dict[str, Any]]:
def _thinking_requests_output_headroom(thinking_config: Any) -> bool:
- """True when Gemini will spend output tokens on thinking.
-
- Thought tokens bill against ``maxOutputTokens``; a global 4096/16384
- ``max_tokens`` can be consumed entirely by high thinking, leaving
- ``finishReason=MAX_TOKENS`` with no answer.
- """
+ """True when Gemini will spend output tokens on thinking (thought tokens bill
+ against ``maxOutputTokens``; a global 4096/16384 cap can be consumed entirely
+ by high thinking, leaving ``finishReason=MAX_TOKENS`` with no answer)."""
normalized = _normalize_thinking_config(thinking_config)
if not normalized:
return False
@@ -433,10 +391,9 @@ def _thinking_requests_output_headroom(thinking_config: Any) -> bool:
def _effective_gemini_max_output_tokens(max_tokens: Optional[int], thinking_config: Any) -> int:
- """Resolve native ``maxOutputTokens``: an omitted/invalid cap becomes the
- published ceiling (Gemini truncates on its low internal default), and an
- explicit cap is raised to that ceiling when thinking is enabled so thought
- tokens do not starve the answer."""
+ """Native ``maxOutputTokens``: an omitted/invalid cap becomes the published
+ ceiling (Gemini truncates on its low internal default); an explicit cap is
+ raised to the ceiling when thinking is enabled so thoughts don't starve the answer."""
try:
requested = int(max_tokens)
except (TypeError, ValueError):
@@ -479,10 +436,7 @@ def build_gemini_request(
return request
-_FINISH_REASON_MAP = {
- "STOP": "stop", "MAX_TOKENS": "length", "SAFETY": "content_filter", "RECITATION": "content_filter", "OTHER": "stop",
-}
-
+# ── Gemini → OpenAI response translation ─────────────────────────────────────
def _map_gemini_finish_reason(reason: str) -> str:
return _FINISH_REASON_MAP.get(str(reason or "").upper(), "stop")
@@ -494,7 +448,7 @@ def _tool_call_extra_from_part(part: Dict[str, Any]) -> Optional[Dict[str, Any]]
def _new_call_id(fc: Dict[str, Any]) -> str:
- """Echo Gemini's functionCall id when present, else mint an OpenAI-style one."""
+ """Echo the functionCall/delta ``id`` when present, else mint an OpenAI-style one."""
fc_id = fc.get("id")
return fc_id if isinstance(fc_id, str) and fc_id else f"call_{uuid.uuid4().hex[:12]}"
@@ -515,13 +469,6 @@ def _usage_from_metadata(usage_meta: Dict[str, Any]) -> SimpleNamespace:
)
-def _completion(model: str, message: SimpleNamespace, finish_reason: str, usage: SimpleNamespace) -> SimpleNamespace:
- return SimpleNamespace(
- id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object="chat.completion", created=int(time.time()), model=model,
- choices=[SimpleNamespace(index=0, message=message, finish_reason=finish_reason)], usage=usage,
- )
-
-
def _assistant_message(content: Any, tool_calls: Any, reasoning: Any) -> SimpleNamespace:
return SimpleNamespace(
role="assistant", content=content, tool_calls=tool_calls,
@@ -531,41 +478,43 @@ def _assistant_message(content: Any, tool_calls: Any, reasoning: Any) -> SimpleN
def translate_gemini_response(resp: Dict[str, Any], model: str) -> SimpleNamespace:
candidates = resp.get("candidates") or []
- if not isinstance(candidates, list) or not candidates:
- return _completion(model, _assistant_message("", None, None), "stop", _usage_from_metadata({}))
- cand = candidates[0] if isinstance(candidates[0], dict) else {}
- content_obj = cand.get("content")
- parts = content_obj.get("parts") if isinstance(content_obj, dict) else []
-
+ cand = parts = None
+ if isinstance(candidates, list) and candidates:
+ cand = candidates[0] if isinstance(candidates[0], dict) else {}
+ content_obj = cand.get("content")
+ parts = content_obj.get("parts") if isinstance(content_obj, dict) else []
text_pieces: List[str] = []
reasoning_pieces: List[str] = []
tool_calls: List[SimpleNamespace] = []
-
for index, part in enumerate(parts or []):
if not isinstance(part, dict):
continue
if part.get("thought") is True and isinstance(part.get("text"), str):
reasoning_pieces.append(part["text"])
- continue
- if isinstance(part.get("text"), str):
+ elif isinstance(part.get("text"), str):
text_pieces.append(part["text"])
- continue
- fc = part.get("functionCall")
- if isinstance(fc, dict) and fc.get("name"):
+ elif isinstance(fc := part.get("functionCall"), dict) and fc.get("name"):
tool_call = SimpleNamespace(
id=_new_call_id(fc), type="function", index=index,
function=SimpleNamespace(name=str(fc["name"]), arguments=_dump_call_args(fc)),
)
- extra_content = _tool_call_extra_from_part(part)
- if extra_content:
+ if extra_content := _tool_call_extra_from_part(part):
tool_call.extra_content = extra_content
tool_calls.append(tool_call)
- finish_reason = "tool_calls" if tool_calls else _map_gemini_finish_reason(str(cand.get("finishReason") or ""))
+ if cand is None:
+ finish_reason, usage = "stop", _usage_from_metadata({})
+ else:
+ finish_reason = "tool_calls" if tool_calls else _map_gemini_finish_reason(str(cand.get("finishReason") or ""))
+ usage = _usage_from_metadata(resp.get("usageMetadata") or {})
message = _assistant_message(
- "".join(text_pieces) if text_pieces else None, tool_calls or None, "".join(reasoning_pieces) or None
+ "".join(text_pieces) if text_pieces else ("" if cand is None else None),
+ tool_calls or None, "".join(reasoning_pieces) or None,
+ )
+ return SimpleNamespace(
+ id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object="chat.completion", created=int(time.time()), model=model,
+ choices=[SimpleNamespace(index=0, message=message, finish_reason=finish_reason)], usage=usage,
)
- return _completion(model, message, finish_reason, _usage_from_metadata(resp.get("usageMetadata") or {}))
class _GeminiStreamChunk(SimpleNamespace):
@@ -579,14 +528,11 @@ def _make_stream_chunk(
tool_calls = None
if tool_call_delta is not None:
tool_delta = SimpleNamespace(
- index=tool_call_delta.get("index", 0),
- id=tool_call_delta.get("id") or f"call_{uuid.uuid4().hex[:12]}",
- type="function",
+ index=tool_call_delta.get("index", 0), id=_new_call_id(tool_call_delta), type="function",
function=SimpleNamespace(name=tool_call_delta.get("name") or "", arguments=tool_call_delta.get("arguments") or ""),
)
- extra_content = tool_call_delta.get("extra_content")
- if isinstance(extra_content, dict):
- tool_delta.extra_content = extra_content
+ if isinstance(tool_call_delta.get("extra_content"), dict):
+ tool_delta.extra_content = tool_call_delta["extra_content"]
tool_calls = [tool_delta]
delta = SimpleNamespace(
role="assistant", content=content or None, tool_calls=tool_calls,
@@ -628,7 +574,6 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
cand = candidates[0] if isinstance(candidates[0], dict) else {}
parts = (cand.get("content") or {}).get("parts") or []
chunks: List[_GeminiStreamChunk] = []
-
for part_index, part in enumerate(parts):
if not isinstance(part, dict):
continue
@@ -647,15 +592,14 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
)
slot = tool_call_indices.get(call_key)
if slot is None:
- slot = {"index": len(tool_call_indices), "id": _new_call_id(fc), "last_arguments": ""}
- tool_call_indices[call_key] = slot
+ slot = tool_call_indices[call_key] = {"index": len(tool_call_indices), "id": _new_call_id(fc), "last_arguments": ""}
# Gemini re-sends the full args each event; emit only the new suffix.
last_arguments = str(slot.get("last_arguments") or "")
- emitted_arguments = args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str
+ emitted = args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str
slot["last_arguments"] = args_str
chunks.append(_make_stream_chunk(model=model, tool_call_delta={
"index": slot["index"], "id": slot["id"], "name": name,
- "arguments": emitted_arguments, "extra_content": _tool_call_extra_from_part(part),
+ "arguments": emitted, "extra_content": _tool_call_extra_from_part(part),
}))
finish_reason_raw = str(cand.get("finishReason") or "")
@@ -663,26 +607,17 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
finish_chunk = _make_stream_chunk(
model=model, finish_reason="tool_calls" if tool_call_indices else _map_gemini_finish_reason(finish_reason_raw)
)
- # Carry usageMetadata on the finish chunk so the streaming loop can
- # record token counts like the non-streaming path does.
- usage_meta = event.get("usageMetadata") or {}
- if usage_meta:
+ # usageMetadata rides on the finish chunk so the streaming loop records
+ # token counts like the non-streaming path does.
+ if usage_meta := event.get("usageMetadata") or {}:
finish_chunk.usage = _usage_from_metadata(usage_meta)
chunks.append(finish_chunk)
return chunks
-_HTTP_ERROR_CODES = {401: "gemini_unauthorized", 429: "gemini_rate_limited", 404: "gemini_model_not_found"}
-
-
def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = None) -> GeminiAPIError:
status = response.status_code
- if body_text is None:
- try:
- body_text = response.text
- except Exception:
- body_text = ""
- body_text = body_text or ""
+ body_text = (_response_text(response) if body_text is None else body_text) or ""
err_obj: Any = None
if body_text:
try:
@@ -696,8 +631,8 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
err_message = str(err_obj.get("message") or "").strip()
details_list = err_obj.get("details")
- reason = ""
- metadata: Dict[str, Any] = {}
+ # First google.rpc.ErrorInfo detail supplies reason/metadata.
+ reason, metadata = "", {}
for detail in details_list if isinstance(details_list, list) else []:
if isinstance(detail, dict) and not reason and str(detail.get("@type") or "").endswith("/google.rpc.ErrorInfo"):
reason_value, md = detail.get("reason"), detail.get("metadata")
@@ -707,12 +642,10 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
metadata = md
retry_after: Optional[float] = None
- header_retry = response.headers.get("Retry-After") or response.headers.get("retry-after")
- if header_retry:
- try:
- retry_after = float(header_retry)
- except (TypeError, ValueError):
- pass
+ try:
+ retry_after = float(response.headers.get("Retry-After") or response.headers.get("retry-after"))
+ except (TypeError, ValueError):
+ pass
if err_message:
message = f"Gemini HTTP {status} ({err_status or 'error'}): {err_message}"
@@ -721,27 +654,21 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
# Users who bypassed the setup wizard (raw GOOGLE_API_KEY in .env) still
# need to learn that the free tier cannot sustain an agent session.
if status == 429 and is_free_tier_quota_error(err_message or body_text):
- message = message + _FREE_TIER_GUIDANCE
+ message += _FREE_TIER_GUIDANCE
# Legacy "Standard" key rejection: Google's raw 401 misleadingly asks for
# OAuth; append the actual fix (mint a new Gemini API key in AI Studio).
if is_standard_key_auth_error(status, err_message or body_text, reason):
- message = message + _STANDARD_KEY_GUIDANCE
-
+ message += _STANDARD_KEY_GUIDANCE
return GeminiAPIError(
- message,
- code=_HTTP_ERROR_CODES.get(status, f"gemini_http_{status}"),
- status_code=status,
- response=response,
+ message, code=_HTTP_ERROR_CODES.get(status, f"gemini_http_{status}"), status_code=status, response=response,
retry_after=retry_after,
details={"status": err_status, "reason": reason, "metadata": metadata, "message": err_message},
)
class GeminiNativeClient:
- """Minimal OpenAI-SDK-compatible facade over Gemini's native REST API.
-
- ``client.chat.completions.create(**kwargs)`` mirrors the OpenAI SDK surface.
- """
+ """Minimal OpenAI-SDK-compatible facade (``client.chat.completions.create(**kwargs)``)
+ over Gemini's native REST API."""
# Declared for agent/auxiliary_client.py: already a complete client, so it
# is never re-dispatched through a wire adapter. (No HERMES_SKIP_ASYNC_WRAP
@@ -804,14 +731,12 @@ class GeminiNativeClient:
max_tokens: Optional[int] = None, top_p: Optional[float] = None, stop: Any = None,
extra_body: Optional[Dict[str, Any]] = None, timeout: Any = None, **_: Any,
) -> Any:
- thinking_config = None
- if isinstance(extra_body, dict):
- thinking_config = extra_body.get("thinking_config") or extra_body.get("thinkingConfig")
+ extra = extra_body if isinstance(extra_body, dict) else {}
+ thinking_config = extra.get("thinking_config") or extra.get("thinkingConfig")
request = build_gemini_request(
messages=messages or [], tools=tools, tool_choice=tool_choice, temperature=temperature,
max_tokens=max_tokens, top_p=top_p, stop=stop, thinking_config=thinking_config, model=model,
)
-
model = bare_gemini_model_id(model)
if stream:
return self._stream_completion(model=model, request=request, timeout=timeout)
@@ -832,19 +757,15 @@ class GeminiNativeClient:
def _stream_completion(self, *, model: str, request: Dict[str, Any], timeout: Any = None) -> Iterator[_GeminiStreamChunk]:
url = f"{self.base_url}/models/{model}:streamGenerateContent?alt=sse"
stream_headers = {**self._headers(), "Accept": "text/event-stream"}
-
- def _generator() -> Iterator[_GeminiStreamChunk]:
- try:
- with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response:
- if response.status_code != 200:
- raise gemini_http_error(response, body_text=read_streaming_error_body(response))
- tool_call_indices: Dict[str, Dict[str, Any]] = {}
- for event in _iter_sse_events(response):
- yield from translate_stream_event(event, model, tool_call_indices)
- except httpx.HTTPError as exc:
- raise GeminiAPIError(f"Gemini streaming request failed: {exc}", code="gemini_stream_error") from exc
-
- return _generator()
+ try:
+ with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response:
+ if response.status_code != 200:
+ raise gemini_http_error(response, body_text=read_streaming_error_body(response))
+ tool_call_indices: Dict[str, Dict[str, Any]] = {}
+ for event in _iter_sse_events(response):
+ yield from translate_stream_event(event, model, tool_call_indices)
+ except httpx.HTTPError as exc:
+ raise GeminiAPIError(f"Gemini streaming request failed: {exc}", code="gemini_stream_error") from exc
class AsyncGeminiNativeClient:
@@ -860,16 +781,15 @@ class AsyncGeminiNativeClient:
self._real_client = sync_client
async def _create_chat_completion(self, **kwargs: Any) -> Any:
- stream = bool(kwargs.get("stream"))
result = await asyncio.to_thread(self._sync.chat.completions.create, **kwargs)
- if not stream:
+ if not kwargs.get("stream"):
return result
async def _async_stream() -> Any:
while True:
done, chunk = await asyncio.to_thread(self._sync._advance_stream_iterator, result)
if done:
- break
+ return
yield chunk
return _async_stream()
diff --git a/agent/gemini_schema.py b/agent/gemini_schema.py
index 748378a940..9f9a956f9c 100644
--- a/agent/gemini_schema.py
+++ b/agent/gemini_schema.py
@@ -53,10 +53,9 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
# Gemini requires every ``enum`` entry to be a string even for
# integer/number/boolean types; the declared type stays intact and Gemini
- # still emits typed tool arguments at runtime.
+ # still emits typed tool arguments at runtime. dict.fromkeys = ordered dedupe.
enum_val = cleaned.get("enum")
if isinstance(enum_val, list) and cleaned.get("type") in {"integer", "number", "boolean"}:
- # dict.fromkeys = order-preserving dedupe.
stringified = list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None))
if stringified:
cleaned["enum"] = stringified
@@ -71,13 +70,12 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
required_val = cleaned.get("required")
if isinstance(required_val, list):
props_val = cleaned.get("properties")
- prop_names = set(props_val.keys()) if isinstance(props_val, dict) else set()
+ prop_names = set(props_val) if isinstance(props_val, dict) else set()
valid_required = [name for name in required_val if isinstance(name, str) and name in prop_names]
if not valid_required:
cleaned.pop("required", None)
elif len(valid_required) != len(required_val):
cleaned["required"] = valid_required
-
return cleaned
diff --git a/agent/jiter_preload.py b/agent/jiter_preload.py
index 0d27b4d38b..6a69352a78 100644
--- a/agent/jiter_preload.py
+++ b/agent/jiter_preload.py
@@ -1,10 +1,8 @@
-"""Best-effort early import of the OpenAI SDK's native streaming parser.
-
-On some Windows installs ``jiter``'s native extension imports fine from the venv
-but fails when first imported later inside the threaded streaming path. Loading
-it once at agent-package import avoids that while preserving the SDK's normal
-error path for genuinely broken installs.
-"""
+"""Best-effort early import of the OpenAI SDK's native streaming parser: on some
+Windows installs ``jiter``'s native extension imports fine from the venv but fails
+when first imported inside the threaded streaming path. Loading it once at
+agent-package import avoids that while keeping the SDK's normal error path for
+genuinely broken installs."""
from __future__ import annotations
@@ -15,7 +13,6 @@ _JITER_PRELOAD_ERROR: Exception | None = None
def preload_jiter_native_extension() -> bool:
- """Import jiter's native extension early if it is available."""
global _JITER_PRELOADED, _JITER_PRELOAD_ERROR
if _JITER_PRELOADED:
return True
@@ -25,8 +22,7 @@ def preload_jiter_native_extension() -> bool:
except Exception as exc:
_JITER_PRELOAD_ERROR = exc
return False
- _JITER_PRELOADED = True
- _JITER_PRELOAD_ERROR = None
+ _JITER_PRELOADED, _JITER_PRELOAD_ERROR = True, None
return True
diff --git a/agent/lazy_forward.py b/agent/lazy_forward.py
index ae65a912fb..8c7d646f54 100644
--- a/agent/lazy_forward.py
+++ b/agent/lazy_forward.py
@@ -1,28 +1,25 @@
"""Lazy forwarding helpers shared by the AIAgent façade and its mixins.
-Targets are resolved on every call, so ``patch(".")`` in tests still intercepts and the
-heavy agent modules stay off the ``run_agent`` import path.
+Targets are resolved on every call, so ``patch(".")`` in tests still
+intercepts and the heavy agent modules stay off the ``run_agent`` import path.
"""
import importlib
def lazy_attr(module: str, name: str):
- """Resolve ``module.name`` at call time (keeps heavy/cyclic imports off the run_agent import path)."""
+ """Resolve ``module.name`` at call time."""
return getattr(importlib.import_module(module), name)
def forward(module: str, name: str, *, static: bool = False):
- """Build an AIAgent method that lazily forwards to ``module.name`` (``target(self, *args, **kwargs)``).
-
- The target is resolved on every call, so ``patch(".")`` in tests still intercepts and
- run_agent keeps its import-time cost. ``static=True`` drops ``self``.
- """
+ """Build an AIAgent method that lazily forwards to ``module.name``
+ (``target(self, *args, **kwargs)``; ``static=True`` drops ``self``)."""
if static:
def forwarder(*args, **kwargs):
- return getattr(importlib.import_module(module), name)(*args, **kwargs)
+ return lazy_attr(module, name)(*args, **kwargs)
else:
def forwarder(self, *args, **kwargs):
- return getattr(importlib.import_module(module), name)(self, *args, **kwargs)
+ return lazy_attr(module, name)(self, *args, **kwargs)
forwarder.__name__ = forwarder.__qualname__ = name
forwarder.__doc__ = f"Forwarder — see ``{module}.{name}``."
return staticmethod(forwarder) if static else forwarder
diff --git a/agent/lmstudio_reasoning.py b/agent/lmstudio_reasoning.py
index 799e8dd538..0835264a77 100644
--- a/agent/lmstudio_reasoning.py
+++ b/agent/lmstudio_reasoning.py
@@ -1,41 +1,30 @@
-"""LM Studio reasoning-effort resolution shared by the chat-completions
-transport and run_agent's iteration-limit summary path.
-
-LM Studio publishes per-model ``capabilities.reasoning.allowed_options``
-(``["off","on"]`` for toggle models, ``["off","minimal","low"]`` for graduated
-ones). We map the user's ``reasoning_config`` onto LM Studio's OpenAI-compatible
-vocabulary, then clamp against the model's allowed set so the server doesn't 400.
-"""
+"""LM Studio reasoning-effort resolution (chat-completions transport + run_agent's
+iteration-limit summary path). LM Studio publishes per-model
+``capabilities.reasoning.allowed_options`` (``["off","on"]`` for toggle models,
+``["off","minimal","low"]`` for graduated ones); the user's ``reasoning_config`` is
+mapped onto LM Studio's vocabulary, then clamped to the allowed set so the server
+doesn't 400."""
from __future__ import annotations
from typing import List, Optional
-# Top-level reasoning_effort values LM Studio's OpenAI-compatible endpoint accepts.
_LM_VALID_EFFORTS = {"none", "minimal", "low", "medium", "high", "xhigh"}
-# Toggle-style models publish allowed_options as ["off","on"]; map onto the
-# request vocabulary. Also applied to the published allowed_options themselves.
+# Toggle vocabulary → request vocabulary; also applied to published allowed_options.
_LM_EFFORT_ALIASES = {"off": "none", "on": "medium"}
-# Hermes' ladder grew past LM Studio's vocabulary ("max", "ultra"). Without this
-# ceiling clamp they miss _LM_VALID_EFFORTS, keep the "medium" default and are
-# conflated with unparseable input — asking for more yields less than "xhigh".
-# Kept separate from _LM_EFFORT_ALIASES, which must not rewrite allowed_options.
+# Hermes' ladder grew past LM Studio's vocabulary ("max", "ultra"); without this
+# ceiling clamp they'd fall to the "medium" default (more yields less than "xhigh").
+# Separate from _LM_EFFORT_ALIASES, which must not rewrite allowed_options.
_LM_EFFORT_CLAMP = {"max": "xhigh", "ultra": "xhigh"}
-def resolve_lmstudio_effort(
- reasoning_config: Optional[dict],
- allowed_options: Optional[List[str]],
-) -> Optional[str]:
- """Return the ``reasoning_effort`` to send to LM Studio, or ``None``.
-
- ``None`` means "omit the field": the user picked a level the model can't
- honor, so LM Studio falls back to the model's declared default rather than
- a silently substituted effort. Falsy ``allowed_options`` (probe failed)
- skips clamping and sends the resolved effort anyway.
- """
+def resolve_lmstudio_effort(reasoning_config: Optional[dict], allowed_options: Optional[List[str]]) -> Optional[str]:
+ """Return the ``reasoning_effort`` to send to LM Studio, or ``None`` = omit the
+ field (the user picked a level the model can't honor, so LM Studio falls back
+ to the model's declared default rather than a silently substituted effort).
+ Falsy ``allowed_options`` (probe failed) skips clamping."""
effort = "medium"
if reasoning_config and isinstance(reasoning_config, dict):
if reasoning_config.get("enabled") is False:
@@ -46,8 +35,6 @@ def resolve_lmstudio_effort(
raw = _LM_EFFORT_CLAMP.get(raw, raw)
if raw in _LM_VALID_EFFORTS:
effort = raw
- if allowed_options:
- allowed = {_LM_EFFORT_ALIASES.get(opt, opt) for opt in allowed_options}
- if effort not in allowed:
- return None
+ if allowed_options and effort not in {_LM_EFFORT_ALIASES.get(opt, opt) for opt in allowed_options}:
+ return None
return effort