diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py
index d5e1bb90a4..902b25092e 100644
--- a/agent/copilot_acp_client.py
+++ b/agent/copilot_acp_client.py
@@ -6,6 +6,7 @@ as one prompt, collects text chunks, and returns the minimal OpenAI-client shape
from __future__ import annotations
+import contextlib
import json
import logging
import os
@@ -32,19 +33,34 @@ from tools.environments.local import hermes_subprocess_env
ACP_MARKER_BASE_URL = "acp://copilot"
logger = logging.getLogger(__name__)
_DEFAULT_TIMEOUT_SECONDS = 900.0
-
-# Stderr fingerprint of the deprecated `gh copilot` extension. Require BOTH the
-# product name AND a deprecation marker: the NEW `@github/copilot` CLI (repo
-# github/copilot-cli) legitimately mentions "copilot-cli" in its own banners.
+# Stderr fingerprint of the deprecated `gh copilot` extension. Require BOTH the product name
+# AND a deprecation marker: the NEW `@github/copilot` CLI legitimately mentions "copilot-cli".
_DEPRECATION_REQUIRED = ("gh-copilot",)
_DEPRECATION_MARKERS = ("has been deprecated", "no commands will be executed")
-
_ROLE_LABELS = {"system": "System", "user": "User", "assistant": "Assistant", "tool": "Tool", "context": "Context"}
-
-# Probe verdicts per binary path so the ~50ms --help cost is paid once per
-# process. Only definitive True/False is cached; an inconclusive probe is not,
-# so a CLI installed mid-session is picked up.
+# Probe verdicts per binary path (~50ms --help paid once per process). Only definitive
+# True/False is cached, 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,\npoint 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\ndirectly with a Copilot subscription "
+ "token) via `hermes setup`.\n\nOriginal error:\n"
+)
def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
@@ -54,73 +70,57 @@ 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]:
- raw = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
- return shlex.split(raw) if raw else ["--acp", "--stdio"]
+ return shlex.split(os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()) or ["--acp", "--stdio"]
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.
- Only probes when ``--acp`` is among ``args`` — a custom transport is the operator's business.
- """
+ """Tri-state ``--acp`` probe (a CLI without the flag exits 1 and the parent would wait 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 → normal spawn error). Skipped when ``--acp`` is not in ``args`` (custom transport)."""
if "--acp" not in args:
return True
- cached = _ACP_PROBE_CACHE.get(command)
- if cached is not None:
+ if (cached := _ACP_PROBE_CACHE.get(command)) is not None:
return cached
try:
probe = subprocess.run(
- [command, "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace",
- timeout=5, stdin=subprocess.DEVNULL,
+ [command, "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
+ stdin=subprocess.DEVNULL,
)
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return None
if probe.returncode != 0:
return None
# ``--acp`` as a flag token; tolerate spacing and ``[--acp]`` variants.
- verdict = bool(re.search(r"(?:^|[\s\[])--acp(?:[\s=\],]|$)", probe.stdout, re.MULTILINE))
- _ACP_PROBE_CACHE[command] = verdict
+ verdict = _ACP_PROBE_CACHE[command] = bool(re.search(r"(?:^|[\s\[])--acp(?:[\s=\],]|$)", probe.stdout, re.MULTILINE))
return verdict
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."""
- home = os.environ.get("HOME", "").strip()
- if home:
+ """Stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
+ if home := os.environ.get("HOME", "").strip():
return home
- expanded = os.path.expanduser("~")
- if expanded and expanded != "~":
+ if (expanded := os.path.expanduser("~")) and expanded != "~":
return expanded
try:
import pwd
- resolved = pwd.getpwuid(os.getuid()).pw_dir.strip() # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows)
- if resolved:
- return resolved
+ return pwd.getpwuid(os.getuid()).pw_dir.strip() or "/tmp" # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows)
except Exception:
- pass
- return "/tmp"
+ return "/tmp"
def _build_subprocess_env() -> dict[str, str]:
- # Copilot ACP drives a model and legitimately needs LLM provider credentials;
- # the central helper still strips Tier-1 secrets (bot tokens, GitHub auth, infra).
+ from hermes_constants import apply_subprocess_home_env
+
+ # Copilot ACP drives a model and needs LLM provider credentials; the central helper still
+ # strips Tier-1 secrets (bot tokens, GitHub auth, infra).
env = hermes_subprocess_env(inherit_credentials=True)
env["HOME"] = _resolve_home_dir()
- from hermes_constants import apply_subprocess_home_env
apply_subprocess_home_env(env)
return env
@@ -133,103 +133,42 @@ def _jsonrpc_error(message_id: Any, code: int, message: str) -> dict[str, Any]:
return {"jsonrpc": "2.0", "id": message_id, "error": {"code": code, "message": message}}
-def _permission_denied(message_id: Any) -> dict[str, Any]:
- return _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
+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.
- """
+def _model_selection_request(session: dict[str, Any], requested_model: str) -> tuple[str, dict[str, str]] | None:
+ """ACP request selecting ``requested_model`` for ``session``: stable v1
+ ``session/set_config_option``, else Copilot's pre-stabilization ``session/set_model``
+ 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"
- ),
- 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:
+ options = [o for o in (session.get("configOptions") or []) if isinstance(o, dict) and "model" in (o.get("category"), o.get("id"))]
+ if options:
+ if requested_model not in _enabled_ids(options[0].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"
- }
- if available and requested_model not in available:
- return None
- return (
- "session/set_model",
- {"sessionId": session_id, "modelId": requested_model},
- )
+ return "session/set_config_option", {"sessionId": session_id, "configId": str(options[0].get("id") or "model"), "value": requested_model}
+ available = _enabled_ids((session.get("models") or {}).get("availableModels"), "modelId")
+ return None if available and requested_model not in available else ("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,
- tool_choice: Any = 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))
-
+ # 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. Copilot has no tools of its own that collide with Hermes', so forward the whole toolset.
+ 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
+ for message in (m for m in messages if isinstance(m, dict)):
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}")
+ if rendered := _render_message_content(message.get("content")):
+ 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,33 +178,21 @@ 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 json.dumps(content, ensure_ascii=True)
+ return content["content"].strip() if isinstance(content.get("content"), str) else 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()
def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path:
- candidate = Path(path_text)
- if not candidate.is_absolute():
+ if not Path(path_text).is_absolute():
raise PermissionError("ACP file-system paths must be absolute.")
- resolved = candidate.resolve()
- root = Path(cwd).resolve()
+ resolved, root = Path(path_text).resolve(), Path(cwd).resolve()
try:
resolved.relative_to(root)
except ValueError as exc:
@@ -275,48 +202,33 @@ def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path:
def _effective_timeout(timeout: Any) -> float:
"""Normalise a float or httpx.Timeout-like object to wall-clock seconds (largest component wins)."""
- if timeout is None:
- 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")]
+ return max((float(v) for v in candidates if isinstance(v, (int, float))), default=_DEFAULT_TIMEOUT_SECONDS)
def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd)
- block_error = get_read_block_error(str(path))
- if block_error:
+ if block_error := get_read_block_error(str(path)):
raise PermissionError(block_error)
try:
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])
- if content:
- content = redact_sensitive_text(content, force=True)
- return {"content": content}
+ end = line - 1 + limit if isinstance(limit, int) and limit > 0 else None
+ content = "".join(content.splitlines(keepends=True)[line - 1:end])
+ return {"content": redact_sensitive_text(content, force=True) if content else content}
def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any:
path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd)
- denied = get_write_denied_error(str(path))
- if denied:
+ if denied := get_write_denied_error(str(path)):
raise PermissionError(denied)
- # Approval-gated paths (e.g. ~/.ssh/config) are only soft-gated for interactive
- # tools, but the ACP shim has no human channel to confirm — fail closed.
- if is_write_approval_required(str(path)):
- raise PermissionError(
- f"Write denied: '{path}' requires interactive approval "
- "and cannot be written through the ACP file bridge."
- )
+ if is_write_approval_required(str(path)): # soft-gated for interactive tools; the ACP shim has no human channel → fail closed
+ raise PermissionError(f"Write denied: '{path}' requires interactive approval and cannot be written through the ACP file bridge.")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(str(params.get("content") or ""), encoding="utf-8")
return None
@@ -325,138 +237,80 @@ 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."""
- # Declared for agent/auxiliary_client.py: this shim drives an ACP subprocess
- # over stdio, so it is already a complete client (never re-dispatch it
- # through a wire adapter) and is safe to use from async code as-is.
+ # Declared for agent/auxiliary_client.py: this shim drives an ACP subprocess over stdio, so it is
+ # already a complete client (never re-dispatch through a wire adapter) and async-safe as-is.
HERMES_SKIP_TRANSPORT_WRAP = True
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
+ self.api_key, self.base_url = api_key or "copilot-acp", base_url or ACP_MARKER_BASE_URL
self._default_headers = dict(default_headers or {})
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.is_closed = False
- self._active_process: subprocess.Popen[str] | None = None
+ self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
+ self.is_closed, self._active_process = False, None
self._active_process_lock = threading.Lock()
def close(self) -> None:
with self._active_process_lock:
- proc = self._active_process
- self._active_process = None
+ proc, self._active_process = self._active_process, None
self.is_closed = True
- if proc is None:
- return
try:
- proc.terminate()
- proc.wait(timeout=2)
+ if proc is not None:
+ proc.terminate()
+ proc.wait(timeout=2)
except Exception:
- try:
+ with contextlib.suppress(Exception):
proc.kill()
- except Exception:
- 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 = 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,
+ message = SimpleNamespace(
+ content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning or None, reasoning_content=reasoning or None,
reasoning_details=None,
)
- 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")
+ completion = SimpleNamespace(
+ choices=[SimpleNamespace(message=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",
+ )
return _completion_to_stream_chunks(completion) if stream else completion
def _spawn(self) -> subprocess.Popen[str]:
- # Fast-fail when the CLI rejects --acp: without the probe the parent waits
- # the full child timeout for stdout that never arrives. ``None`` falls
- # through to the spawn, which raises the established start error.
+ # Fast-fail when the CLI rejects --acp (else the parent waits the full child timeout for stdout that
+ # never arrives). ``None`` falls through to the spawn's established start error.
if _acp_supported(self._acp_command, self._acp_args) is False:
preview = " ".join(self._acp_args[:3]) if self._acp_args else "(none)"
raise RuntimeError(
- f"ACP transport not supported by '{self._acp_command}': "
- f"`{preview}` is rejected as an unknown option. "
- f"This usually means the CLI is an older release (e.g. "
- f"Claude Code v2.x) or a different tool than expected. "
- f"Either install a CLI that ships with --acp support "
- f"(e.g. `@github/copilot` late 2025+), or set "
- f"HERMES_COPILOT_ACP_COMMAND / HERMES_COPILOT_ACP_ARGS "
- f"to a working pair."
+ f"ACP transport not supported by '{self._acp_command}': `{preview}` is rejected as an unknown option. This "
+ "usually means the CLI is an older release (e.g. Claude Code v2.x) or a different tool than expected. Either "
+ "install a CLI that ships with --acp support (e.g. `@github/copilot` late 2025+), or set "
+ "HERMES_COPILOT_ACP_COMMAND / HERMES_COPILOT_ACP_ARGS to a working pair."
)
try:
- # Hide the console the child would flash on Windows; stdio pipes stay intact.
- from hermes_cli._subprocess_compat import windows_hide_flags
+ from hermes_cli._subprocess_compat import windows_hide_flags # hide the Windows console flash; pipes intact
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(),
+ [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(),
)
except FileNotFoundError as exc:
- raise RuntimeError(
- f"Could not start Copilot ACP command '{self._acp_command}'. "
- "Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH."
- ) from exc
+ raise RuntimeError(f"Could not start Copilot ACP command '{self._acp_command}'. Install GitHub Copilot CLI or set "
+ "HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.") from exc
if proc.stdin is None or proc.stdout is None:
proc.kill()
raise RuntimeError("Copilot ACP process did not expose stdin/stdout pipes.")
@@ -466,165 +320,101 @@ class CopilotACPClient:
return proc
def _run_prompt(self, prompt_text: str, *, timeout_seconds: float, model: str | None = None) -> tuple[str, str]:
- # The CLI's `--model` spawn flag is deliberately NOT used: `copilot --acp`
- # validates it (unknown id aborts the spawn) but ignores it for the session.
- # The model is applied after session/new via ACP-native model selection.
+ # The CLI's `--model` spawn flag is deliberately NOT used: `copilot --acp` validates it (unknown id
+ # aborts the spawn) but ignores it for the session; the model is applied after session/new instead.
requested_model = str(model or "").strip()
proc = self._spawn()
inbox: queue.Queue[dict[str, Any]] = queue.Queue()
stderr_tail: deque[str] = deque(maxlen=40)
- def _stdout_reader() -> None:
- for line in proc.stdout:
- try:
- inbox.put(json.loads(line))
- except Exception:
- inbox.put({"raw": line.rstrip("\n")})
+ def _decode(line: str) -> dict[str, Any]:
+ try:
+ return json.loads(line)
+ except Exception:
+ return {"raw": line.rstrip("\n")}
- def _stderr_reader() -> None:
- if proc.stderr is None:
- return
- for line in proc.stderr:
- stderr_tail.append(line.rstrip("\n"))
+ def _pump(stream, sink) -> None:
+ for line in stream or ():
+ sink(line)
- threading.Thread(target=_stdout_reader, daemon=True).start()
- threading.Thread(target=_stderr_reader, daemon=True).start()
- next_id = 0
+ threading.Thread(target=_pump, args=(proc.stdout, lambda line: inbox.put(_decode(line))), daemon=True).start()
+ threading.Thread(target=_pump, args=(proc.stderr, lambda line: stderr_tail.append(line.rstrip("\n"))), daemon=True).start()
+ request_ids = iter(range(1, 1 << 62))
def _request(method: str, params: dict[str, Any], *, text_parts: list[str] | None = None, reasoning_parts: list[str] | None = None) -> Any:
- nonlocal next_id
- next_id += 1
- request_id = next_id
+ request_id = next(request_ids)
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 {}
raise RuntimeError(f"Copilot ACP {method} failed: {err.get('message') or err}")
return msg.get("result")
-
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)
+ if (selection := _model_selection_request(session, requested_model)) is not None:
+ _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,
- )
-
+ logger.warning("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,
- )
+ prompt = {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]}
+ _request("session/prompt", prompt, 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()
content = update.get("content") or {}
chunk_text = str(content.get("text") or "") if isinstance(content, dict) else ""
- if kind == "agent_message_chunk" and chunk_text and text_parts is not None:
- text_parts.append(chunk_text)
- elif kind == "agent_thought_chunk" and chunk_text and reasoning_parts is not None:
- reasoning_parts.append(chunk_text)
+ sinks = {"agent_message_chunk": text_parts, "agent_thought_chunk": reasoning_parts}
+ if chunk_text and (sink := sinks.get(str(update.get("sessionUpdate") or "").strip())) is not None:
+ sink.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)
+ response = _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
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..0b54133550 100644
--- a/agent/credits_tracker.py
+++ b/agent/credits_tracker.py
@@ -1,76 +1,47 @@
-"""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).
-"""
+"""Nous credits: parse ``x-nous-credits-*`` / ``x-nous-tool-pool-*`` response
+headers into a validated CreditsState (depletion = paid_access, subscription-cap
+used_fraction, warn-once schema-version gating) and drive the notice policy.
+Header contract: see ``_HEADER_FIELDS``. Money is micros ints only; ``*_usd``
+strings are preserved verbatim (never re-parsed to float)."""
from __future__ import annotations
import logging
import os
import re
+import threading
import time
from dataclasses import dataclass
+from decimal import Decimal
from typing import Any, Mapping, Optional
from utils import is_truthy_value
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).
+_version_warning_emitted: bool = False # warn-once latch (per process)
_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"
+_USD_RE = re.compile(r"^-?\d+\.\d{2}$") # optional minus, digits, exactly 2 decimals
+_SENTINEL = object() # "parse failed"
def _safe_int(value: Any) -> Any:
"""Exact int (money-safe) or ``_SENTINEL``. ``int()`` directly, NOT ``int(float())``
(precision loss above 2**53 corrupts money); float-shaped strings fail."""
- if value is None:
- return _SENTINEL
try:
- return int(str(value))
+ return _SENTINEL if value is None else int(str(value))
except (TypeError, ValueError):
return _SENTINEL
-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))
-
-
@dataclass
class CreditsState:
- """Full credits state parsed from x-nous-credits-* response headers."""
+ """Credits state parsed from x-nous-credits-* response headers."""
version: int = 0
remaining_micros: int = 0
remaining_usd: str = ""
- subscription_micros: int = 0 # SIGNED — may be negative (debt). ONLY field allowed negative.
+ subscription_micros: int = 0 # SIGNED — the ONLY field allowed negative (debt)
subscription_usd: str = ""
subscription_limit_micros: Optional[int] = None # PAIRED + OPTIONAL (only when subscription_cap)
subscription_limit_usd: Optional[str] = None
@@ -83,7 +54,7 @@ class CreditsState:
paid_access: bool = True # depletion keys off THIS == False, NEVER remaining==0
disabled_reason: Optional[str] = None # header omitted entirely when null
as_of_ms: int = 0
- captured_at: float = 0.0 # time.time() when this was captured
+ captured_at: float = 0.0 # time.time() when captured
from_header: bool = False # True only when populated by parse_credits_headers()
@property
@@ -96,51 +67,44 @@ 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)."""
+ """``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
- 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:
+ """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``."""
+ lim = self.subscription_limit_micros
+ if not isinstance(lim, int) or lim <= 0:
return None
- used = self.subscription_limit_micros - self.subscription_micros
- return max(0.0, min(1.0, used / self.subscription_limit_micros))
+ return max(0.0, min(1.0, (lim - self.subscription_micros) / lim))
# ── Credits policy constants. Switching notices sticky→TTL later also needs a
# paired *_TTL_MS per notice kind (AgentNotice has the field; not plumbed yet).
CREDITS_NOTICE_KIND = "sticky" # v1: credits notices are sticky
CREDITS_RESTORED_TTL_MS = 8000 # the only TTL notice in v1 (depletion-recovery confirmation)
-
# 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.
+# escalating line shows the HIGHEST band reached; climbing replaces it, recovery steps 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
-
+CREDITS_USAGE_KEY = "credits.usage"
# Min subscription balance counting as "grant not yet spent" for the grant_spent
-# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent
-# residue where headers report 0 — without the floor such a seed opens the gate
-# and the first header re-creates the at-open nag.
+# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent residue
+# where headers report 0 — without the floor a seed opens the gate and the first
+# header re-creates the at-open nag.
GRANT_UNSPENT_MIN_MICROS = 10_000
def new_credits_latch() -> dict:
- """Fresh notice latch for :func:`evaluate_credits_notices`. Every producer must
- build it here so a new gate key lands everywhere at once."""
+ """Fresh notice latch for :func:`evaluate_credits_notices`; every producer builds it here so a new gate key lands everywhere."""
return {"active": set(), "seen_below_90": False, "usage_band": None, "seen_grant_unspent": False}
@dataclass
class AgentNotice:
- """Driver-agnostic out-of-band notice, fired via ``AIAgent.notice_callback``
- (cleared via ``notice_clear_callback``); each driver renders its own way.
- ``kind``/``ttl_ms`` stay expressive so a future config can switch v1's
- sticky credits notices to TTL without touching the policy."""
+ """Driver-agnostic out-of-band notice (``AIAgent.notice_callback`` / ``notice_clear_callback``); each driver
+ renders its own way. ``kind``/``ttl_ms`` stay expressive so a future config can switch v1's sticky notices to TTL."""
text: str
level: str = "info" # info | warn | error | success
@@ -155,18 +119,11 @@ 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.
-
- 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).
-
- Fail-open to False (depleted notice still shows): a wrong warning is
- recoverable noise; hiding it on a paid model masks a real block.
- """
+ """True when *model* is a Nous free-tier model, using ONLY local data: (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). Fail-open to
+ False (depleted notice still shows): a wrong warning is recoverable noise; hiding it masks a real block."""
if not model:
return False
if model.endswith(":free") or model.startswith("stealth/"):
@@ -176,84 +133,55 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool:
try:
from hermes_cli.models import _is_model_free, peek_cached_pricing
- # 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)
+ pricing = peek_cached_pricing(base_url) # owns the /v1-suffix and auth-state key details
+ return bool(pricing) and _is_model_free(model, pricing)
except Exception:
return False
-def evaluate_credits_notices(
- state: CreditsState, latch: dict, *, model_is_free: bool = False,
-) -> tuple[list[AgentNotice], list[str]]:
- """Reconcile credits notices against the latch (see :func:`new_credits_latch`).
- Mutates ``latch`` IN PLACE. Pure — no I/O, no agent/run_agent imports.
-
- ``model_is_free`` (see :func:`is_free_tier_model`) suppresses
- ``credits.depleted`` — a depleted account on a free model keeps inferencing,
- so the banner is noise. Suppression does NOT emit "restored"; that fires
- only on a genuine ``paid_access`` flip back to True.
-
- Returns ``(to_show, to_clear)``; caller emits to_clear FIRST, then to_show.
- """
+def evaluate_credits_notices(state: CreditsState, latch: dict, *, model_is_free: bool = False) -> tuple[list[AgentNotice], list[str]]:
+ """Reconcile credits notices against the latch (see :func:`new_credits_latch`); mutates ``latch`` IN
+ PLACE. Pure — no I/O, no agent/run_agent imports. ``model_is_free`` suppresses ``credits.depleted`` (a
+ depleted account on a free model keeps inferencing) WITHOUT emitting "restored" — that fires only on a
+ genuine ``paid_access`` flip back to True. Returns ``(to_show, to_clear)``; caller emits to_clear FIRST."""
to_show: list[AgentNotice] = []
to_clear: list[str] = []
- uf = state.used_fraction
-
+ uf, active = state.used_fraction, 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.
+ # 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, and a
+ # mid-session top-up flips current_band → None so the clear path removes the line.
current_band: Optional[tuple[float, str, int]] = None
- if uf is not None:
- 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
+ if uf is not None and state.purchased_micros <= 0:
+ current_band = next((b for b in reversed(CREDITS_USAGE_BANDS) if uf >= b[0]), None)
# ── 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)
if target_band is not None:
- # 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)
- )
+ # Absolute dollars used (a bare "N%" is only meaningful against a Nous cap): cap − remaining,
+ # clamped [0, cap]; "$?" if a producer set the limit without its *_usd. Re-emits on band change only.
+ 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 "?"
+ text = f"{'⚠' if level == 'warn' else '•'} You've used ${used_usd} of your ${state.subscription_limit_usd or '?'} cap"
+ to_show.append(_sticky_notice(text, level, CREDITS_USAGE_KEY))
active.add(CREDITS_USAGE_KEY)
latch["usage_band"] = target_band
@@ -261,10 +189,11 @@ 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.
+ grant_cond = (
+ state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
+ )
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")
- )
+ to_show.append(_sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent"))
active.add("credits.grant_spent")
latch["seen_grant_unspent"] = False
elif "credits.grant_spent" in active and not grant_cond:
@@ -272,6 +201,7 @@ def evaluate_credits_notices(
active.discard("credits.grant_spent")
# ── depleted: suppressed while the model is free (inference still works).
+ depleted_cond = not state.paid_access
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"))
@@ -279,9 +209,7 @@ def evaluate_credits_notices(
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.
+ if not depleted_cond: # genuine recovery only — a free-model switch while depleted is NOT "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",
@@ -289,41 +217,47 @@ def evaluate_credits_notices(
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 contract: (field, kind[, default-when-absent]); a field is REQUIRED unless it has a default.
+# Header name = ``x-nous-credits-`` (``x-nous-`` for tool_pool_*), underscores → dashes.
+# micros: int >= 0 ("signed": may be negative); usd: the server's formatted string ^-?\d+\.\d{2}$
+# (never re-parsed); bool: "true"/"false" STRING. Handled inline: subscription-limit-* (PAIRED/optional),
+# denominator-kind ("subscription_cap" | "none"), disabled-reason (omitted when null).
+_HEADER_FIELDS: tuple[tuple, ...] = (
+ ("remaining_micros", "micros"), ("subscription_micros", "signed"), ("rollover_micros", "micros"),
+ ("purchased_micros", "micros"), ("as_of_ms", "micros"), ("tool_pool_micros", "micros", 0),
+ ("remaining_usd", "usd"), ("subscription_usd", "usd"), ("purchased_usd", "usd"),
+ ("paid_access", "bool", True), # absent → fail-open (assume access)
+ ("tool_pool_gated_off", "bool", False),
)
+def _header_name(field: str) -> str:
+ return "x-nous-" + ("" if field.startswith("tool_pool_") else "credits-") + field.replace("_", "-")
+
+
+def _parse_field(kind: str, raw: Optional[str], default: Any = _SENTINEL) -> Any:
+ """One header value → field value; ``default`` when absent, ``_SENTINEL`` on a contract violation."""
+ if raw is None:
+ return default
+ 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 _USD_RE.match(raw) else _SENTINEL
+ 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.
-
- Returns None (miss) on ANY of: no ``x-nous-credits-version`` header; version
- != 1 (> 1 also warns once); a ``*_micros``/``as_of_ms`` field non-integer or
- negative (subscription excepted); a ``*_usd`` not matching
- ``^-?\\d+\\.\\d{2}$``; ``denominator_kind`` outside {"subscription_cap",
- "none"}; ``paid_access``/``tool_pool_gated_off`` not exactly "true"/"false";
- any unexpected exception.
-
- Fail-open on the subscription_limit pair: a half-pair (only -micros or only
- -usd) is treated as both-absent — the parse STILL SUCCEEDS with both None.
- """
+ None (miss) on ANY of: no version header; version != 1 (> 1 also warns once);
+ a required field violating ``_HEADER_FIELDS``; unknown ``denominator_kind``;
+ any unexpected exception. Fail-open on the subscription_limit pair: a
+ half-pair (only -micros or only -usd) parses as both-absent (both None)."""
global _version_warning_emitted
try:
- # Cheap probe before the lowercase copy (header names are case-insensitive):
- # bail when the version header is absent — the hot path for non-Nous providers.
+ # Cheap probe before the lowercase copy (header names are case-insensitive): bail when the
+ # version header is absent — the hot path for non-Nous providers.
if not any(k.lower() == "x-nous-credits-version" for k in headers):
return None
lowered = {k.lower(): v for k, v in headers.items()}
@@ -335,80 +269,39 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt
_version_warning_emitted = True
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):
- 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:
- 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
-
- # 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):
- return None
- fields["subscription_limit_micros"] = lm
- fields["subscription_limit_usd"] = sub_limit_usd_raw
+ fields: dict[str, Any] = {
+ name: _parse_field(kind, lowered.get(_header_name(name)), *default) for name, kind, *default in _HEADER_FIELDS
+ }
+ 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:
+ fields["subscription_limit_micros"] = _parse_field("micros", lim_micros_raw)
+ fields["subscription_limit_usd"] = _parse_field("usd", lim_usd_raw)
denominator_kind = lowered.get("x-nous-credits-denominator-kind", "none")
- if denominator_kind not in _VALID_DENOMINATOR_KINDS:
+ if _SENTINEL in fields.values() or 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,
- disabled_reason=lowered.get("x-nous-credits-disabled-reason"), # None if absent (omitted when null)
- captured_at=time.time(),
- from_header=True,
- **fields,
- )
- except Exception:
- # Fail-open → miss; breadcrumb distinguishes a parser regression from a
- # legitimate no-headers response.
+ disabled_reason = lowered.get("x-nous-credits-disabled-reason") # None if absent (omitted when null)
+ return CreditsState(version=version_val, denominator_kind=denominator_kind, disabled_reason=disabled_reason,
+ captured_at=time.time(), from_header=True, **fields)
+ except Exception: # fail-open → miss; the breadcrumb distinguishes a parser regression from a no-headers response
logger.debug("credits ▸ parse_credits_headers raised (fail-open miss)", exc_info=True)
return None
-# ── Dev fixtures (HERMES_DEV_CREDITS_FIXTURE): throwaway scaffolding to trigger
-# any notice state without real spend. Value is a state NAME or a FILE PATH whose
-# contents are a name (re-read every turn → `echo depleted > /tmp/cf` flips live).
-# Drives the per-turn capture/notice path, the cold-start seed, and /usage.
+# ── Dev fixtures (HERMES_DEV_CREDITS_FIXTURE): throwaway scaffolding to trigger any notice state
+# without real spend. Value is a state NAME or a FILE PATH whose contents are a name (re-read every
+# turn → `echo depleted > /tmp/cf` flips live). Drives per-turn notices, the cold-start seed, and /usage.
def _fixture(remaining: str, subscription: str, limit: Optional[str] = None, purchased: Optional[str] = None,
*, paid: bool = True, reason: Optional[str] = None) -> dict:
"""Fixture spec from *_usd strings; micros derived exactly (Decimal)."""
- from decimal import Decimal
d: dict = {}
- for field, usd in (("remaining", remaining), ("subscription", subscription),
- ("subscription_limit", limit), ("purchased", purchased)):
+ for field, usd in (("remaining", remaining), ("subscription", subscription), ("subscription_limit", limit), ("purchased", purchased)):
if usd is not None:
d[f"{field}_micros"], d[f"{field}_usd"] = int(Decimal(usd) * 1_000_000), usd
if limit is not None:
d["denominator_kind"] = "subscription_cap"
d["paid_access"] = paid
- if reason is not None:
- d["disabled_reason"] = reason
- return d
+ return d if reason is None else {**d, "disabled_reason": reason}
_DEV_FIXTURES: dict[str, dict] = {
@@ -426,59 +319,44 @@ _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.
- """
- if not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
+ """Fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None (unknown name / unset). Prod-leak guard:
+ applies ONLY when HERMES_DEV_CREDITS is also on, so a stray fixture env var never surfaces fabricated balances."""
+ name = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
+ if not name or not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
return None
- raw = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
- if not raw:
- 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
- spec = _DEV_FIXTURES.get(name.lower())
- if not spec:
+ if not (spec := _DEV_FIXTURES.get(name.lower())):
return None
- # 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())
+ # Stamp what the REAL parser always guarantees so a fixture is field-identical to a
+ # parse_credits_headers() result: version 1 and a valid purchased_usd (zero top-up = "0.00").
+ return CreditsState(**{"version": 1, "purchased_usd": "0.00", **spec}, from_header=True, captured_at=time.time())
def _credits_state_from_account(info) -> Optional[CreditsState]:
- """Map a NousPortalAccountInfo into a header-shaped CreditsState for the seed.
- Float account dollars → micros plus a DISPLAY *_usd (formatting account floats
- is allowed; parsing a server *_usd is not). Fail-open → None."""
+ """Map a NousPortalAccountInfo into a header-shaped CreditsState for the seed. 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)
+ return (int(round(dollars * 1_000_000)), f"{dollars:.2f}") if isinstance(dollars, (int, float)) else (0, "")
+ fields: dict[str, Any] = {}
+ for prefix, attr in (("remaining", "total_usable_credits"), ("subscription", "subscription_credits_remaining"),
+ ("purchased", "purchased_credits_remaining")):
+ fields[f"{prefix}_micros"], fields[f"{prefix}_usd"] = _money(getattr(acc, attr, 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(),
+ **fields, subscription_limit_micros=cap[0], subscription_limit_usd=cap[1], from_header=False, captured_at=time.time(),
+ rollover_micros=_money(getattr(sub, "rollover_credits", None))[0], paid_access=paid if isinstance(paid, bool) else True,
+ denominator_kind="subscription_cap" if cap[0] is not None else "none",
)
except Exception:
logger.debug("credits ▸ seed account→state mapping failed", exc_info=True)
@@ -486,63 +364,45 @@ def _credits_state_from_account(info) -> Optional[CreditsState]:
def _hydrate_seed_state(agent, state) -> None:
- """Install a seed CreditsState on the agent and fire the notice policy once.
- Primes the crossing gate: the cold-start snapshot IS the first observation, so
- a session opening in a band warns immediately. Safe from a worker thread."""
+ """Install a seed CreditsState on the agent and fire the notice policy once. Primes the crossing gate:
+ the cold-start snapshot IS the first observation, so a session opening in a band warns immediately."""
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:
- # 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
- emit = getattr(agent, "_emit_credits_notices", None)
- if callable(emit):
+ latch = getattr(agent, "_credits_latch", None)
+ if isinstance(latch, dict) and state.used_fraction is not None:
+ latch["seen_below_90"] = True # ONLY this gate — priming seen_grant_unspent would revive the steady-state nag
+ if callable(emit := getattr(agent, "_emit_credits_notices", None)):
emit()
def seed_credits_at_session_start(agent) -> bool:
- """Hydrate agent._credits_state from the portal account (or a dev fixture) and
- fire the notice policy so warnings show at session OPEN. Shared by the
- TUI/desktop build ("ready") and first-turn setup (plain-CLI fallback).
- Idempotent once a seed or real header populated _credits_state.
-
- Returns True iff it seeded this call. Never raises — credits must never block startup.
- """
+ """Hydrate agent._credits_state from the portal account (or a dev fixture) and fire the notice policy so
+ warnings show at session OPEN (TUI/desktop "ready" and plain-CLI first-turn setup). Idempotent once a seed
+ or real header populated _credits_state. Returns True iff it seeded this call. Never raises."""
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()
except Exception:
fixture = None
- if fixture is not None:
- # Synchronous: a fixture is instant, and tests rely on the state +
- # notice landing before this returns.
+ if fixture is not None: # synchronous: instant, and tests rely on state + notice landing before return
_hydrate_seed_state(agent, fixture)
return True
- # FIRE-AND-FORGET: a slow portal must never delay "ready". The daemon
- # thread re-checks idempotency (a live header may land first).
- import threading
-
- def _bg_seed() -> None:
+ def _bg_seed() -> None: # FIRE-AND-FORGET: a slow portal must never delay "ready"
try:
from hermes_cli.nous_account import get_nous_portal_account_info
info = get_nous_portal_account_info(force_fresh=True)
if getattr(agent, "_credits_state", None) is not None:
return # a live inference header beat us — don't clobber it
- state = _credits_state_from_account(info)
- if state is not None:
+ if (state := _credits_state_from_account(info)) is not None:
_hydrate_seed_state(agent, state)
except Exception:
logger.debug("credits ▸ session-start seed (background) failed", exc_info=True)
threading.Thread(target=_bg_seed, name="credits-seed", daemon=True).start()
return True
except Exception:
- # Innermost log across all call sites so a dead seed is diagnosable.
- logger.debug("credits ▸ session-start seed failed (fail-open)", exc_info=True)
+ logger.debug("credits ▸ session-start seed failed (fail-open)", exc_info=True) # innermost log: diagnosable dead seed
return False
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..fb693723c0 100644
--- a/agent/gemini_native_adapter.py
+++ b/agent/gemini_native_adapter.py
@@ -1,17 +1,14 @@
-"""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.
-"""
+"""OpenAI-compatible facade over Google AI Studio's native Gemini API: the ``gemini``
+provider keeps ``api_mode='chat_completions'`` so the agent loop stays OpenAI-shaped,
+and this shim converts ``messages[]``/``tools[]`` into ``models/{model}:generateContent``
+payloads and responses back. Google's OpenAI-compat endpoint is brittle for the
+multi-turn tool loop (auth churn, tool-call replay, thought signatures); native is canonical."""
from __future__ import annotations
import asyncio
import base64
+import contextlib
import json
import logging
import re
@@ -33,117 +30,20 @@ try:
_HERMES_VERSION = str(_hermes_cli.__version__)
except Exception:
_HERMES_VERSION = "0.0.0"
+_API_CLIENT = f"hermes-agent/{_HERMES_VERSION}" # client context per Gemini's partner-integration guidance
DEFAULT_GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
-# Published max output-token ceiling shared by every current Gemini text model.
-# Used when the caller passes max_tokens=None: unlike OpenAI-compat endpoints,
-# Gemini's native API applies a low internal default and truncates output.
+# Published max output-token ceiling shared by every current Gemini text model; used
+# for max_tokens=None because the native API's low internal default truncates output.
GEMINI_DEFAULT_MAX_OUTPUT_TOKENS = 65535
-
-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):
- 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
-
-
-def is_native_gemini_base_url(base_url: str) -> bool:
- """Return True when the endpoint speaks Gemini's native REST API."""
- normalized = str(base_url or "").strip().rstrip("/").lower()
- return "generativelanguage.googleapis.com" in normalized and not normalized.endswith("/openai")
-
-
-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)."""
- 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")]
-
- 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,
- headers={"Content-Type": "application/json", "X-Goog-Api-Client": f"hermes-agent/{_HERMES_VERSION}"},
- )
- except Exception as exc:
- logger.debug("probe_gemini_tier: network error: %s", exc)
- return "unknown"
-
- rpd_header = {k.lower(): v for k, v in resp.headers.items()}.get("x-ratelimit-limit-requests-per-day")
- if rpd_header:
- # Free-tier daily caps top out at 1000 (flash-lite); Tier 1 starts ~1500+.
- try:
- 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 "paid" if 200 <= resp.status_code < 300 else "unknown"
-
-
-def is_free_tier_quota_error(error_message: str) -> bool:
- """Return 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.
- """
- 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 "
@@ -152,87 +52,156 @@ _STANDARD_KEY_GUIDANCE = (
"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"
+_END = object() # stream-exhausted marker for _advance_stream_iterator
+_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"}
+_MISSING_KEY_ERROR = (
+ "Gemini native client requires an API key, but none was provided. Set GOOGLE_API_KEY or GEMINI_API_KEY in your "
+ "environment / ~/.hermes/.env (get one at https://aistudio.google.com/app/apikey), or run `hermes setup` to "
+ "configure the Google provider."
+)
+
+
+def bare_gemini_model_id(model: str) -> str:
+ """Strip Gemini's own provider prefix from an aggregator-style model id."""
+ name = (model or "").strip()
+ for prefix in ("google/", "gemini/"):
+ if name.lower().startswith(prefix):
+ return name[len(prefix):].strip() or name
+ return name
+
+
+def gemini_requires_tool_call_ids(model: str) -> bool:
+ """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:
+ """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")
+
+
+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 key → ``"free"`` | ``"paid"`` | ``"unknown"`` (probe failed; callers proceed without blocking)."""
+ key = (api_key or "").strip()
+ if not key:
+ return "unknown"
+ base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL
+ base = re.sub(r"/openai\Z", "", base, flags=re.IGNORECASE)
+ payload = {"contents": [{"role": "user", "parts": [{"text": "hi"}]}], "generationConfig": {"maxOutputTokens": 1}}
+ headers = {"Content-Type": "application/json", "X-Goog-Api-Client": _API_CLIENT}
+ try:
+ with httpx.Client(timeout=timeout) as client:
+ resp = client.post(f"{base}/models/{model}:generateContent", params={"key": key}, json=payload, headers=headers)
+ except Exception as exc:
+ logger.debug("probe_gemini_tier: network error: %s", exc)
+ return "unknown"
+ rpd_header = {k.lower(): v for k, v in resp.headers.items()}.get("x-ratelimit-limit-requests-per-day")
+ try:
+ if rpd_header: # free-tier daily caps top out at 1000 (flash-lite); Tier 1 starts ~1500+
+ return "free" if int(rpd_header) <= 1000 else "paid"
+ except (TypeError, ValueError):
+ pass
+ if resp.status_code == 429:
+ 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:
+ """True when a Gemini 429 message indicates free-tier exhaustion."""
+ return bool(error_message) and "free_tier" in error_message.lower()
+
+
+def is_standard_key_auth_error(status: int, error_message: str, reason: str = "") -> bool:
+ """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."""
+ return status == 401 and (reason == "ACCESS_TOKEN_TYPE_UNSUPPORTED" or "expected oauth 2 access token" in (error_message or "").lower())
class GeminiAPIError(Exception):
"""Error shape compatible with Hermes retry/error classification."""
- def __init__(
- self, message: str, *, code: str = "gemini_api_error", status_code: Optional[int] = None,
- response: Optional[httpx.Response] = None, retry_after: Optional[float] = None,
- details: Optional[Dict[str, Any]] = None,
- ) -> None:
+ def __init__(self, message: str, *, code: str = "gemini_api_error", status_code: Optional[int] = None,
+ response: Optional[httpx.Response] = None, retry_after: Optional[float] = None, details: Optional[Dict[str, Any]] = 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 _text_of(item: Any) -> Optional[str]:
+ """Text of an OpenAI content item (plain str or ``{"type": "text"}`` dict), else None."""
+ if isinstance(item, dict) and item.get("type") == "text" and isinstance(item.get("text"), str):
+ return item["text"]
+ return item if isinstance(item, str) else None
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 str(content)
+ return "\n".join(t for t in map(_text_of, content) if t is not None)
+ return "" if content is None else str(content)
+
+
+def _inline_data_part(item: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+ """``inlineData`` part for an ``image_url`` item carrying a ``data:`` URL; None otherwise."""
+ url = (item.get("image_url") or {}).get("url") or ""
+ if item.get("type") != "image_url" or not isinstance(url, str) or not url.startswith("data:"):
+ return None
+ try:
+ header, encoded = url.split(",", 1)
+ data = base64.b64encode(base64.b64decode(encoded)).decode("ascii")
+ return {"inlineData": {"mimeType": header.split(":", 1)[1].split(";", 1)[0], "data": data}}
+ except Exception:
+ return None
+
+
+def _multimodal_part(item: Any) -> Optional[Dict[str, Any]]:
+ text = _text_of(item)
+ if text or isinstance(item, str):
+ return {"text": text}
+ return _inline_data_part(item) if isinstance(item, dict) else None
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})
- 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 "")
- if not isinstance(url, str) or not url.startswith("data:"):
- continue
- try:
- header, encoded = url.split(",", 1)
- mime = header.split(":", 1)[1].split(";", 1)[0]
- raw = base64.b64decode(encoded)
- except Exception:
- continue
- parts.append({"inlineData": {"mimeType": mime, "data": base64.b64encode(raw).decode("ascii")}})
- return parts
+ if isinstance(content, list):
+ return [p for p in map(_multimodal_part, content) if p]
+ text = _coerce_content_to_text(content)
+ return [{"text": text}] if text else []
def _tool_call_extra_signature(tool_call: Dict[str, Any]) -> Optional[str]:
+ """Replayed Gemini thoughtSignature from ``extra_content`` (``google.thought_signature`` or flat)."""
extra = tool_call.get("extra_content") or {}
- if not isinstance(extra, dict):
- return None
- google = extra.get("google") or extra.get("thought_signature")
- sig = (google.get("thought_signature") or google.get("thoughtSignature")) if isinstance(google, dict) else google
+ sig = (extra.get("google") or extra.get("thought_signature")) if isinstance(extra, dict) else None
+ if isinstance(sig, dict):
+ sig = sig.get("thought_signature") or sig.get("thoughtSignature")
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 "")
@@ -244,62 +213,42 @@ def _translate_tool_call_to_gemini(tool_call: Dict[str, Any], include_ids: bool
args = json.loads(args_raw) if isinstance(args_raw, str) and args_raw else {}
except json.JSONDecodeError:
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}}
- 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: Dict[str, Any] = {"name": str(fn.get("name") or ""), "args": args if isinstance(args, dict) else {"_value": args}}
+ if include_ids and (call_id := _tool_call_id(tool_call)):
+ call["id"] = call_id
+ 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 (``#/...``).
-
- 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.
- """
+ """True if a parsed value contains a JSON-Schema ``$ref`` pointer (``#/...``): 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) is forwarded as opaque text.
+ False positives only lose the structured shape, never the content."""
if isinstance(node, dict):
- return any(
- (key == "$ref" and isinstance(value, str) and value.startswith("#/")) or _looks_like_json_schema(value)
- for key, value in node.items()
- )
+ return any((k == "$ref" and isinstance(v, str) and v.startswith("#/")) or _looks_like_json_schema(v) for k, v in node.items())
return isinstance(node, list) and any(_looks_like_json_schema(item) for item in node)
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,
+ 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 (may be an unwrapped MCP name 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:
parsed = json.loads(content) if content.strip().startswith(("{", "[")) else None
except json.JSONDecodeError:
parsed = None
- response = parsed if isinstance(parsed, dict) and not _looks_like_json_schema(parsed) else {"output": content}
- function_response: Dict[str, Any] = {"name": name, "response": response}
+ structured = isinstance(parsed, dict) and not _looks_like_json_schema(parsed)
+ function_response: Dict[str, Any] = {"name": name, "response": parsed if structured else {"output": content}}
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.
- if is_gemini3:
- image_parts = [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p]
- if image_parts:
- function_response["parts"] = image_parts
+ # Gemini 3.x accepts images inside functionResponse.parts; 2.x rejects the field.
+ if image_parts := [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p] if is_gemini3 else []:
+ function_response["parts"] = image_parts
return {"functionResponse": function_response}
@@ -307,143 +256,117 @@ def _has_function_response(content: Dict[str, Any]) -> bool:
return any(isinstance(part, dict) and "functionResponse" in part for part in content.get("parts", []))
+def _merge_alternating(contents: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+ """Alternation contract for generateContent: 1) adjacent same-role contents merge (else HTTP 400
+ "multiturn requests [must] alternate"); 2) EXCEPT 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
+ functionResponse + functionResponse still merge); 3) the split pair stays API-valid via an interposed
+ placeholder model turn."""
+ merged: List[Dict[str, Any]] = []
+ for content in contents:
+ 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.append({"role": "model", "parts": [{"text": _INTERRUPTED_RESPONSE_PLACEHOLDER}]})
+ if same_role:
+ merged[-1]["parts"].extend(content["parts"])
+ else:
+ merged.append(content)
+ return merged
+
+
def _build_gemini_contents(
messages: List[Dict[str, Any]], include_tool_call_ids: bool = False, *, is_gemini3: bool = False
) -> tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]]:
system_text_parts: List[str] = []
contents: List[Dict[str, Any]] = []
tool_name_by_call_id: Dict[str, str] = {}
-
for msg in messages:
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
if role in {"tool", "function"}:
- part = _translate_tool_result_to_gemini(
- msg, tool_name_by_call_id=tool_name_by_call_id, include_ids=include_tool_call_ids, is_gemini3=is_gemini3
- )
+ part = _translate_tool_result_to_gemini(msg, tool_name_by_call_id, include_tool_call_ids, is_gemini3=is_gemini3)
contents.append({"role": "user", "parts": [part]})
continue
parts = _extract_multimodal_parts(msg.get("content"))
tool_calls = msg.get("tool_calls") or []
- for tool_call in tool_calls if isinstance(tool_calls, list) else []:
- 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 ""))
- if tool_call_id and tool_name:
- tool_name_by_call_id[tool_call_id] = tool_name
+ for tool_call in (tc for tc in tool_calls if isinstance(tc, dict)) if isinstance(tool_calls, list) else ():
+ tool_name = str((tool_call.get("function") or {}).get("name") or "")
+ if (call_id := _tool_call_id(tool_call)) and tool_name:
+ tool_name_by_call_id[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").
- # 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]] = []
- for content in contents:
- prev = merged_contents[-1] if merged_contents 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}]})
-
- if same_role:
- merged_contents[-1]["parts"].extend(content["parts"])
- else:
- merged_contents.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 _merge_alternating(contents), ({"role": "system", "parts": [{"text": joined_system}]} if joined_system else None)
+
+
+def _function_declaration(tool: Any) -> Optional[Dict[str, Any]]:
+ 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"]):
+ return None
+ decl: Dict[str, Any] = {"name": fn["name"]}
+ if isinstance(fn.get("description"), str) and fn["description"]:
+ decl["description"] = fn["description"]
+ if isinstance(fn.get("parameters"), dict):
+ decl["parameters"] = sanitize_gemini_tool_parameters(fn["parameters"])
+ return decl
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:
- 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
- decl: Dict[str, Any] = {"name": fn["name"]}
- if isinstance(fn.get("description"), str) and fn["description"]:
- decl["description"] = fn["description"]
- if isinstance(fn.get("parameters"), dict):
- decl["parameters"] = sanitize_gemini_tool_parameters(fn["parameters"])
- declarations.append(decl)
+ declarations = [d for d in map(_function_declaration, tools if isinstance(tools, list) else []) if d]
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]}}
- if isinstance(tool_choice, dict):
- name = (tool_choice.get("function") or {}).get("name")
- if isinstance(name, str) and name:
- return {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [name]}}
- return None
+ name = (tool_choice.get("function") or {}).get("name") if isinstance(tool_choice, dict) else None
+ return {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [name]}} if isinstance(name, str) and name else None
+
+
+# (camelCase key, snake_case alias, accepted types, normalizer)
+_THINKING_KEYS = (
+ ("thinkingBudget", "thinking_budget", (int, float), int),
+ ("includeThoughts", "include_thoughts", bool, lambda v: v),
+ ("thinkingLevel", "thinking_level", str, lambda v: v.strip().lower()),
+)
def _normalize_thinking_config(config: Any) -> Optional[Dict[str, Any]]:
- if not isinstance(config, dict) or not config:
+ if not isinstance(config, dict):
return None
- budget = config.get("thinkingBudget", config.get("thinking_budget"))
- include = config.get("includeThoughts", config.get("include_thoughts"))
- level = config.get("thinkingLevel", config.get("thinking_level"))
- normalized: Dict[str, Any] = {}
- if isinstance(budget, (int, float)):
- normalized["thinkingBudget"] = int(budget)
- if isinstance(include, bool):
- normalized["includeThoughts"] = include
- if isinstance(level, str) and level.strip():
- normalized["thinkingLevel"] = level.strip().lower()
+ values = {key: config.get(key, config.get(alias)) for key, alias, _, _ in _THINKING_KEYS}
+ normalized = {key: norm(values[key]) for key, _, types, norm in _THINKING_KEYS
+ if isinstance(values[key], types) and (values[key].strip() if isinstance(values[key], str) else True)}
return normalized or None
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.
- """
- normalized = _normalize_thinking_config(thinking_config)
- if not normalized:
- return False
+ """True when Gemini will spend output tokens on thinking: thought tokens bill against ``maxOutputTokens``,
+ so a global 4096/16384 cap can be consumed entirely by high thinking (``finishReason=MAX_TOKENS``, no answer)."""
+ normalized = _normalize_thinking_config(thinking_config) or {}
+ budget, has_level = normalized.get("thinkingBudget"), "thinkingLevel" in normalized
if normalized.get("includeThoughts") is False:
- return "thinkingLevel" in normalized or bool(normalized.get("thinkingBudget"))
- budget = normalized.get("thinkingBudget")
- return not (isinstance(budget, int) and budget <= 0 and "thinkingLevel" not in normalized)
+ return has_level or bool(budget)
+ return bool(normalized) and not (isinstance(budget, int) and budget <= 0 and not has_level)
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):
requested = 0
- if requested <= 0:
- return GEMINI_DEFAULT_MAX_OUTPUT_TOKENS
- if _thinking_requests_output_headroom(thinking_config):
+ if requested <= 0 or _thinking_requests_output_headroom(thinking_config):
return max(requested, GEMINI_DEFAULT_MAX_OUTPUT_TOKENS)
return requested
@@ -456,45 +379,29 @@ def build_gemini_request(
# Gemini 3+ both requires tool-call ids and accepts multimodal functionResponse parts.
is_gemini3 = gemini_requires_tool_call_ids(model)
contents, system_instruction = _build_gemini_contents(messages, include_tool_call_ids=is_gemini3, is_gemini3=is_gemini3)
- request: Dict[str, Any] = {"contents": contents}
- for key, value in (
- ("systemInstruction", system_instruction),
- ("tools", _translate_tools_to_gemini(tools)),
+ optional = (
+ ("systemInstruction", system_instruction), ("tools", _translate_tools_to_gemini(tools)),
("toolConfig", _translate_tool_choice_to_gemini(tool_choice)),
- ):
- if value:
- request[key] = value
-
- generation_config: Dict[str, Any] = {}
- if temperature is not None:
- generation_config["temperature"] = temperature
- generation_config["maxOutputTokens"] = _effective_gemini_max_output_tokens(max_tokens, thinking_config)
- if top_p is not None:
- generation_config["topP"] = top_p
- if stop:
- generation_config["stopSequences"] = stop if isinstance(stop, list) else [str(stop)]
- if normalized_thinking := _normalize_thinking_config(thinking_config):
- generation_config["thinkingConfig"] = normalized_thinking
- request["generationConfig"] = generation_config
+ )
+ request: Dict[str, Any] = {"contents": contents, **{k: v for k, v in optional if v}}
+ # Key order is part of the wire format (prompt-cache parity): temperature, maxOutputTokens, topP, stop, thinking.
+ generation = (
+ ("temperature", temperature), ("maxOutputTokens", _effective_gemini_max_output_tokens(max_tokens, thinking_config)),
+ ("topP", top_p), ("stopSequences", (stop if isinstance(stop, list) else [str(stop)]) if stop else None),
+ ("thinkingConfig", _normalize_thinking_config(thinking_config)),
+ )
+ request["generationConfig"] = {k: v for k, v in generation if v is not None}
return request
-_FINISH_REASON_MAP = {
- "STOP": "stop", "MAX_TOKENS": "length", "SAFETY": "content_filter", "RECITATION": "content_filter", "OTHER": "stop",
-}
-
-
-def _map_gemini_finish_reason(reason: str) -> str:
- return _FINISH_REASON_MAP.get(str(reason or "").upper(), "stop")
-
-
+# ── Gemini → OpenAI response translation ─────────────────────────────────────
def _tool_call_extra_from_part(part: Dict[str, Any]) -> Optional[Dict[str, Any]]:
sig = part.get("thoughtSignature")
return {"google": {"thought_signature": sig}} if isinstance(sig, str) and sig else None
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]}"
@@ -507,110 +414,87 @@ def _dump_call_args(fc: Dict[str, Any], **kwargs: Any) -> str:
def _usage_from_metadata(usage_meta: Dict[str, Any]) -> SimpleNamespace:
+ count = lambda key: int(usage_meta.get(key) or 0) # noqa: E731
return SimpleNamespace(
- prompt_tokens=int(usage_meta.get("promptTokenCount") or 0),
- completion_tokens=int(usage_meta.get("candidatesTokenCount") or 0),
- total_tokens=int(usage_meta.get("totalTokenCount") or 0),
- prompt_tokens_details=SimpleNamespace(cached_tokens=int(usage_meta.get("cachedContentTokenCount") or 0)),
+ prompt_tokens=count("promptTokenCount"), completion_tokens=count("candidatesTokenCount"),
+ total_tokens=count("totalTokenCount"), prompt_tokens_details=SimpleNamespace(cached_tokens=count("cachedContentTokenCount")),
)
-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 _envelope(model: str, object_: str, choice: SimpleNamespace, usage: Any, cls: type = SimpleNamespace) -> Any:
+ """OpenAI chat.completion / chat.completion.chunk envelope around one choice."""
+ return cls(id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object=object_, created=int(time.time()), model=model,
+ choices=[choice], usage=usage)
-def _assistant_message(content: Any, tool_calls: Any, reasoning: Any) -> SimpleNamespace:
- return SimpleNamespace(
- role="assistant", content=content, tool_calls=tool_calls,
- reasoning=reasoning, reasoning_content=reasoning, reasoning_details=None,
- )
+def _tool_call_ns(name: str, arguments: str, index: int, call_id: str, extra_content: Any) -> SimpleNamespace:
+ """OpenAI-shaped tool call; ``extra_content`` attached only when it is a dict."""
+ extra = {"extra_content": extra_content} if isinstance(extra_content, dict) else {}
+ return SimpleNamespace(id=call_id, type="function", index=index, function=SimpleNamespace(name=name, arguments=arguments), **extra)
+
+
+def _part_text(part: Dict[str, Any]) -> tuple[Optional[str], bool]:
+ """``(text, is_thought)`` for a candidate part; ``(None, False)`` when it carries no text."""
+ text = part.get("text")
+ return (text, part.get("thought") is True) if isinstance(text, str) else (None, False)
+
+
+def _part_function_call(part: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+ fc = part.get("functionCall")
+ return fc if isinstance(fc, dict) and fc.get("name") else None
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 []
-
- text_pieces: List[str] = []
- reasoning_pieces: List[str] = []
+ 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 []
+ pieces: Dict[bool, List[str]] = {False: [], True: []} # is_thought → text pieces
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):
- text_pieces.append(part["text"])
- continue
- fc = part.get("functionCall")
- if isinstance(fc, 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:
- 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 ""))
- message = _assistant_message(
- "".join(text_pieces) if text_pieces else None, tool_calls or None, "".join(reasoning_pieces) or None
- )
- return _completion(model, message, finish_reason, _usage_from_metadata(resp.get("usageMetadata") or {}))
+ text, is_thought = _part_text(part)
+ if text is not None:
+ pieces[is_thought].append(text)
+ elif fc := _part_function_call(part):
+ tool_calls.append(_tool_call_ns(str(fc["name"]), _dump_call_args(fc), index, _new_call_id(fc), _tool_call_extra_from_part(part)))
+ finish_reason = "tool_calls" if tool_calls else _FINISH_REASON_MAP.get(str((cand or {}).get("finishReason") or "").upper(), "stop")
+ usage = _usage_from_metadata((resp.get("usageMetadata") or {}) if cand is not None else {})
+ reasoning = "".join(pieces[True]) or None
+ message = SimpleNamespace(role="assistant", content="".join(pieces[False]) if pieces[False] else ("" if cand is None else None),
+ tool_calls=tool_calls or None, reasoning=reasoning, reasoning_content=reasoning, reasoning_details=None)
+ return _envelope(model, "chat.completion", SimpleNamespace(index=0, message=message, finish_reason=finish_reason), usage)
-class _GeminiStreamChunk(SimpleNamespace):
- pass
+class _GeminiStreamChunk(SimpleNamespace): ...
def _make_stream_chunk(
- *, model: str, content: str = "", tool_call_delta: Optional[Dict[str, Any]] = None,
- finish_reason: Optional[str] = None, reasoning: str = "",
+ *, model: str, content: str = "", tool_call_delta: Optional[Dict[str, Any]] = None, finish_reason: Optional[str] = None, reasoning: str = "",
) -> _GeminiStreamChunk:
- 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",
- 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
- tool_calls = [tool_delta]
- delta = SimpleNamespace(
- role="assistant", content=content or None, tool_calls=tool_calls,
- reasoning=reasoning or None, reasoning_content=reasoning or None,
- )
- return _GeminiStreamChunk(
- id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object="chat.completion.chunk", created=int(time.time()), model=model,
- choices=[SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason)], usage=None,
- )
+ d = tool_call_delta
+ tool_calls = None if d is None else [
+ _tool_call_ns(d.get("name") or "", d.get("arguments") or "", d.get("index", 0), _new_call_id(d), d.get("extra_content"))
+ ]
+ delta = SimpleNamespace(role="assistant", content=content or None, tool_calls=tool_calls, reasoning=reasoning or None,
+ reasoning_content=reasoning or None)
+ choice = SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason)
+ return _envelope(model, "chat.completion.chunk", choice, None, cls=_GeminiStreamChunk)
def _iter_sse_events(response: httpx.Response) -> Iterator[Dict[str, Any]]:
buffer = ""
for chunk in response.iter_text():
- if not chunk:
- continue
- buffer += chunk
+ buffer += chunk or ""
while "\n" in buffer:
line, buffer = buffer.split("\n", 1)
line = line.rstrip("\r")
if not line.startswith("data: "):
continue
- data = line[6:]
- if data == "[DONE]":
+ if (data := line[6:]) == "[DONE]":
return
try:
payload = json.loads(data)
@@ -628,124 +512,89 @@ 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
- if part.get("thought") is True and isinstance(part.get("text"), str):
- chunks.append(_make_stream_chunk(model=model, reasoning=part["text"]))
+ text, is_thought = _part_text(part)
+ if is_thought:
+ chunks.append(_make_stream_chunk(model=model, reasoning=text))
continue
- if isinstance(part.get("text"), str) and part["text"]:
- chunks.append(_make_stream_chunk(model=model, content=part["text"]))
- fc = part.get("functionCall")
- if isinstance(fc, dict) and fc.get("name"):
+ if text:
+ chunks.append(_make_stream_chunk(model=model, content=text))
+ if fc := _part_function_call(part):
name = str(fc["name"])
args_str = _dump_call_args(fc, sort_keys=True)
thought_signature = part.get("thoughtSignature") if isinstance(part.get("thoughtSignature"), str) else ""
- call_key = json.dumps(
- {"part_index": part_index, "name": name, "thought_signature": thought_signature}, sort_keys=True
- )
- 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
+ call_key = json.dumps({"part_index": part_index, "name": name, "thought_signature": thought_signature}, sort_keys=True)
+ if (slot := tool_call_indices.get(call_key)) is None:
+ 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
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),
- }))
-
- finish_reason_raw = str(cand.get("finishReason") or "")
- if finish_reason_raw:
- 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:
+ delta = {"index": slot["index"], "id": slot["id"], "name": name, "extra_content": _tool_call_extra_from_part(part),
+ "arguments": args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str}
+ chunks.append(_make_stream_chunk(model=model, tool_call_delta=delta))
+ if finish_reason_raw := str(cand.get("finishReason") or ""):
+ finish_reason = "tool_calls" if tool_call_indices else _FINISH_REASON_MAP.get(finish_reason_raw.upper(), "stop")
+ finish_chunk = _make_stream_chunk(model=model, finish_reason=finish_reason)
+ if usage_meta := event.get("usageMetadata") or {}: # rides on the finish chunk so the stream loop records tokens
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 _error_info(err_obj: Dict[str, Any]) -> tuple[str, Dict[str, Any]]:
+ """``(reason, metadata)`` from the first google.rpc.ErrorInfo detail (later ones fill gaps until reason is set)."""
+ reason, metadata = "", {}
+ details = err_obj.get("details")
+ for detail in details if isinstance(details, list) else []:
+ if isinstance(detail, dict) and not reason and str(detail.get("@type") or "").endswith("/google.rpc.ErrorInfo"):
+ reason = detail["reason"] if isinstance(detail.get("reason"), str) else reason
+ metadata = detail["metadata"] if isinstance(detail.get("metadata"), dict) else metadata
+ return reason, metadata
+
+
+def _error_object(body_text: str) -> Dict[str, Any]:
+ """The ``error`` object of a Google JSON error body, or ``{}``."""
+ try:
+ parsed = json.loads(body_text) if body_text else None
+ except (ValueError, TypeError):
+ return {}
+ err_obj = parsed.get("error") if isinstance(parsed, dict) else None
+ return err_obj if isinstance(err_obj, dict) else {}
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 ""
- err_obj: Any = None
- if body_text:
- try:
- parsed = json.loads(body_text)
- err_obj = parsed.get("error") if isinstance(parsed, dict) else None
- except (ValueError, TypeError):
- pass
- if not isinstance(err_obj, dict):
- err_obj = {}
- err_status = str(err_obj.get("status") or "").strip()
- err_message = str(err_obj.get("message") or "").strip()
- details_list = err_obj.get("details")
-
- reason = ""
- metadata: Dict[str, Any] = {}
- 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")
- if isinstance(reason_value, str):
- reason = reason_value
- if isinstance(md, dict):
- 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
-
- if err_message:
- message = f"Gemini HTTP {status} ({err_status or 'error'}): {err_message}"
- else:
- message = f"Gemini returned HTTP {status}: {body_text[:500]}"
- # 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.
+ body_text = (_response_text(response) if body_text is None else body_text) or ""
+ err_obj = _error_object(body_text)
+ err_status, err_message = (str(err_obj.get(k) or "").strip() for k in ("status", "message"))
+ reason, metadata = _error_info(err_obj)
+ try:
+ retry_after: Optional[float] = float(response.headers.get("Retry-After") or response.headers.get("retry-after"))
+ except (TypeError, ValueError):
+ retry_after = None
+ message = (
+ f"Gemini HTTP {status} ({err_status or 'error'}): {err_message}" if err_message
+ else f"Gemini returned HTTP {status}: {body_text[:500]}"
+ )
+ # Users who bypassed the setup wizard (raw GOOGLE_API_KEY in .env) still need to learn the free
+ # tier cannot sustain an agent session; a legacy "Standard" key gets the real fix (Google's raw 401 asks for OAuth).
if status == 429 and is_free_tier_quota_error(err_message or body_text):
- message = 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).
+ message += _FREE_TIER_GUIDANCE
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,
- retry_after=retry_after,
- details={"status": err_status, "reason": reason, "metadata": metadata, "message": err_message},
+ 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.
+ """Minimal OpenAI-SDK-compatible facade (``client.chat.completions.create(**kwargs)``) over Gemini's native REST API."""
- ``client.chat.completions.create(**kwargs)`` mirrors the OpenAI SDK surface.
- """
-
- # 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
- # — the async path has a real conversion, AsyncGeminiNativeClient.)
+ # For agent/auxiliary_client.py: a complete client, never re-dispatched through a wire adapter.
+ # (No HERMES_SKIP_ASYNC_WRAP — the async path has a real conversion, AsyncGeminiNativeClient.)
HERMES_SKIP_TRANSPORT_WRAP = True
def __init__(
@@ -753,126 +602,81 @@ class GeminiNativeClient:
timeout: Any = None, http_client: Optional[httpx.Client] = None, **_: Any,
) -> None:
if not (api_key or "").strip():
- raise RuntimeError(
- "Gemini native client requires an API key, but none was provided. Set GOOGLE_API_KEY or "
- "GEMINI_API_KEY in your environment / ~/.hermes/.env (get one at https://aistudio.google.com/app/apikey), "
- "or run `hermes setup` to configure the Google provider."
- )
- self.api_key = api_key
+ raise RuntimeError(_MISSING_KEY_ERROR)
+ self.api_key, self.is_closed = api_key, False
self.base_url = (base_url or DEFAULT_GEMINI_BASE_URL).rstrip("/").removesuffix("/openai")
self._default_headers = dict(default_headers or {})
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
- self.is_closed = False
- self._http = http_client or httpx.Client(
- timeout=timeout or httpx.Timeout(connect=15.0, read=600.0, write=30.0, pool=30.0)
- )
+ self._http = http_client or httpx.Client(timeout=timeout or httpx.Timeout(connect=15.0, read=600.0, write=30.0, pool=30.0))
def close(self) -> None:
self.is_closed = True
- try:
+ with contextlib.suppress(Exception):
self._http.close()
- except Exception:
- pass
-
- def __enter__(self):
- return self
-
- def __exit__(self, exc_type, exc_val, exc_tb):
- self.close()
def _headers(self) -> Dict[str, str]:
- return {
- "Content-Type": "application/json",
- "Accept": "application/json",
- "x-goog-api-key": self.api_key,
- # Client context per Gemini's partner-integration guidance.
- "User-Agent": f"hermes-agent/{_HERMES_VERSION} (gemini-native)",
- "X-Goog-Api-Client": f"hermes-agent/{_HERMES_VERSION}",
- **self._default_headers,
- }
+ return {"Content-Type": "application/json", "Accept": "application/json", "x-goog-api-key": self.api_key,
+ "User-Agent": f"{_API_CLIENT} (gemini-native)", "X-Goog-Api-Client": _API_CLIENT, **self._default_headers}
@staticmethod
def _advance_stream_iterator(iterator: Iterator[_GeminiStreamChunk]) -> tuple[bool, Optional[_GeminiStreamChunk]]:
- try:
- return False, next(iterator)
- except StopIteration:
- return True, None
+ chunk = next(iterator, _END)
+ return (True, None) if chunk is _END else (False, chunk)
def _create_chat_completion(
- self, *, model: str = "gemini-3.7-flash", messages: Optional[List[Dict[str, Any]]] = None,
- stream: bool = False, tools: Any = None, tool_choice: Any = None, temperature: Optional[float] = None,
- max_tokens: Optional[int] = None, top_p: Optional[float] = None, stop: Any = None,
- extra_body: Optional[Dict[str, Any]] = None, timeout: Any = None, **_: Any,
+ self, *, model: str = "gemini-3.7-flash", messages: Optional[List[Dict[str, Any]]] = None, stream: bool = False,
+ tools: Any = None, tool_choice: Any = None, temperature: Optional[float] = None, 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 {}
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,
+ messages=messages or [], tools=tools, tool_choice=tool_choice, temperature=temperature, max_tokens=max_tokens,
+ top_p=top_p, stop=stop, thinking_config=extra.get("thinking_config") or extra.get("thinkingConfig"), model=model,
)
-
model = bare_gemini_model_id(model)
+ url = f"{self.base_url}/models/{model}:"
if stream:
- return self._stream_completion(model=model, request=request, timeout=timeout)
-
- url = f"{self.base_url}/models/{model}:generateContent"
- response = self._http.post(url, json=request, headers=self._headers(), timeout=timeout)
+ return self._stream_completion(model, url + "streamGenerateContent?alt=sse", request, timeout)
+ response = self._http.post(url + "generateContent", json=request, headers=self._headers(), timeout=timeout)
if response.status_code != 200:
raise gemini_http_error(response)
try:
payload = response.json()
except ValueError as exc:
raise GeminiAPIError(
- f"Invalid JSON from Gemini native API: {exc}", code="gemini_invalid_json",
- status_code=response.status_code, response=response,
+ f"Invalid JSON from Gemini native API: {exc}", code="gemini_invalid_json", status_code=response.status_code, response=response,
) from exc
return translate_gemini_response(payload, model=model)
- 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()
+ def _stream_completion(self, model: str, url: str, request: Dict[str, Any], timeout: Any) -> Iterator[_GeminiStreamChunk]:
+ try:
+ headers = {**self._headers(), "Accept": "text/event-stream"}
+ with self._http.stream("POST", url, json=request, headers=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:
"""Async wrapper used by auxiliary_client for native Gemini calls."""
def __init__(self, sync_client: GeminiNativeClient):
- self._sync = sync_client
- self.api_key = sync_client.api_key
- self.base_url = sync_client.base_url
+ # ``_real_client``: the auxiliary cache evicts entries by leaf client; GeminiNativeClient is itself the leaf.
+ self._sync = self._real_client = sync_client
+ self.api_key, self.base_url = sync_client.api_key, sync_client.base_url
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
- # The auxiliary cache evicts entries by leaf client; GeminiNativeClient
- # is itself the leaf (no OpenAI client beneath it).
- 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:
- return result
+ return self._async_stream(result) if kwargs.get("stream") else result
- async def _async_stream() -> Any:
- while True:
- done, chunk = await asyncio.to_thread(self._sync._advance_stream_iterator, result)
- if done:
- break
- yield chunk
-
- return _async_stream()
+ async def _async_stream(self, iterator: Iterator[_GeminiStreamChunk]) -> Any:
+ while not (step := await asyncio.to_thread(self._sync._advance_stream_iterator, iterator))[0]:
+ yield step[1]
async def close(self) -> None:
await asyncio.to_thread(self._sync.close)
diff --git a/agent/gemini_schema.py b/agent/gemini_schema.py
index 748378a940..41b4a3e03f 100644
--- a/agent/gemini_schema.py
+++ b/agent/gemini_schema.py
@@ -5,37 +5,30 @@ from __future__ import annotations
import math
from typing import Any, Dict
-# Gemini's ``FunctionDeclaration.parameters`` accepts the ``Schema`` object,
-# only a subset of OpenAPI 3.0 / JSON Schema; everything else is stripped.
+# Gemini's ``FunctionDeclaration.parameters`` accepts only a subset of OpenAPI 3.0 /
+# JSON Schema (the ``Schema`` object); everything else is stripped.
_GEMINI_SCHEMA_ALLOWED_KEYS = {
- "type", "format", "title", "description", "nullable", "enum",
- "maxItems", "minItems", "properties", "required", "minProperties",
- "maxProperties", "minLength", "maxLength", "pattern", "example",
- "anyOf", "propertyOrdering", "default", "items", "minimum", "maximum",
+ "type", "format", "title", "description", "nullable", "enum", "maxItems", "minItems", "properties", "required",
+ "minProperties", "maxProperties", "minLength", "maxLength", "pattern", "example", "anyOf", "propertyOrdering",
+ "default", "items", "minimum", "maximum",
}
def _stringify_enum_value(item: Any) -> Any:
"""Gemini-safe string for a scalar enum entry, or None to drop it."""
- if isinstance(item, str):
- return item
if isinstance(item, bool):
return "true" if item else "false"
if isinstance(item, (int, float)) and math.isfinite(item):
return str(item)
- return None
+ return item if isinstance(item, str) else None
def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
- """Return a Gemini-compatible copy of a tool parameter schema.
-
- Keeps only the documented Gemini subset (drops e.g. ``$schema`` /
- ``additionalProperties``) and recursively sanitizes nested
- ``properties`` / ``items`` / ``anyOf`` definitions.
- """
+ """Gemini-compatible copy of a tool parameter schema: keeps only the documented subset
+ (drops e.g. ``$schema`` / ``additionalProperties``) and recursively sanitizes nested
+ ``properties`` / ``items`` / ``anyOf``."""
if not isinstance(schema, dict):
return {}
-
cleaned: Dict[str, Any] = {}
for key, value in schema.items():
if key not in _GEMINI_SCHEMA_ALLOWED_KEYS:
@@ -53,31 +46,27 @@ 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:
+ if stringified := list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None)):
cleaned["enum"] = stringified
else:
cleaned.pop("enum", None)
- # Gemini validates ``required`` strictly against the same node's
- # ``properties`` (HTTP 400 "property is not defined"), and one bad tool
- # schema fails the ENTIRE request. MCP servers routinely emit ``required``
- # without ``properties``, so keep only names that exist here; the tool
- # handler still validates required fields at execution time.
+ # Gemini validates ``required`` strictly against the same node's ``properties`` (HTTP 400
+ # "property is not defined") and one bad tool schema fails the ENTIRE request. MCP servers
+ # routinely emit ``required`` without ``properties``, so keep only names that exist here;
+ # the tool handler still validates required fields at execution time.
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