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

refactor(agent/gemini,acp,credits): compact wire adapters and credits parser (parity-verified)

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:22:13 -07:00
parent 113f04616b
commit 26bbac4ab9
8 changed files with 436 additions and 680 deletions
Download Patch File Download Diff File Expand all files Collapse all files

314
agent/copilot_acp_client.py
View File

@@ -46,6 +46,32 @@ _ROLE_LABELS = {"system": "System", "user": "User", "assistant": "Assistant", "t
# so a CLI installed mid-session is picked up.
_ACP_PROBE_CACHE: dict[str, bool] = {}
_PROMPT_PREAMBLE = (
"You are being used as the active ACP agent backend for Hermes.",
"Use ACP capabilities to complete tasks.",
"IMPORTANT: If you take an action with a tool, you MUST output tool calls using <tool_call>{...}</tool_call> blocks with JSON exactly in OpenAI function-call shape.",
"If no tool is needed, answer normally.",
)
_INITIALIZE_PARAMS = {
"protocolVersion": 1,
"clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
"clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
}
_DEPRECATED_CLI_ERROR = (
"Hermes ACP mode requires the NEW GitHub Copilot CLI "
"(github.com/github/copilot-cli), but the binary it just "
"spawned is the deprecated `gh copilot` extension.\n\n"
"Install the new CLI:\n"
" npm install -g @github/copilot\n"
" # then verify with: copilot --help\n\n"
"If `copilot` already resolves to the new CLI but you still see this,\n"
"point Hermes at it explicitly:\n"
" export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
"Alternative: use the `copilot` provider (no ACP, hits the Copilot API\n"
"directly with a Copilot subscription token) via `hermes setup`.\n\n"
"Original error:\n"
)
def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
"""True iff stderr looks like the deprecated gh-copilot extension's banner."""
@@ -54,11 +80,7 @@ def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
def _resolve_command() -> str:
return (
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
or os.getenv("COPILOT_CLI_PATH", "").strip()
or "copilot"
)
return os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip() or os.getenv("COPILOT_CLI_PATH", "").strip() or "copilot"
def _resolve_args() -> list[str]:
@@ -70,10 +92,10 @@ def _acp_supported(command: str, args: list[str]) -> bool | None:
"""Tri-state probe: does ``command`` accept ``--acp``?
A CLI without the flag (older releases, Claude Code v2.x) exits 1 with
``error: unknown option '--acp'`` and the parent then waits the full
child timeout for stdout that never arrives. True = help advertises --acp;
False = help ran cleanly without it (caller fast-fails); None = inconclusive
(binary missing / --help failed), caller falls through to the normal spawn error.
``error: unknown option '--acp'`` and the parent then waits the full child
timeout for stdout that never arrives. True = help advertises --acp; False =
help ran cleanly without it (caller fast-fails); None = inconclusive (binary
missing / --help failed), caller falls through to the normal spawn error.
Only probes when ``--acp`` is among ``args`` — a custom transport is the operator's business.
"""
if "--acp" not in args:
@@ -97,7 +119,7 @@ def _acp_supported(command: str, args: list[str]) -> bool | None:
def _resolve_home_dir() -> str:
"""Return a stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
"""Stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
home = os.environ.get("HOME", "").strip()
if home:
return home
@@ -137,99 +159,61 @@ def _permission_denied(message_id: Any) -> dict[str, Any]:
return _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
def _model_selection_request(
session: dict[str, Any], requested_model: str
) -> tuple[str, dict[str, str]] | None:
def _enabled_ids(entries: Any, key: str) -> set[str]:
"""Ids of ``entries`` (dicts) whose ``_meta.copilotEnablement`` is not ``disabled``."""
return {
str(e.get(key) or "").strip()
for e in (entries or [])
if isinstance(e, dict) and str((e.get("_meta") or {}).get("copilotEnablement") or "").strip().lower() != "disabled"
}
def _model_selection_request(session: dict[str, Any], requested_model: str) -> tuple[str, dict[str, str]] | None:
"""Return the ACP request that selects ``requested_model`` for ``session``.
Prefer stable v1 ``session/set_config_option``. Fall back to Copilot's
pre-stabilization ``session/set_model`` extension only when no model
config option is advertised. A reported model list is authoritative:
unknown and policy-disabled ids return None instead of being sent.
Prefer stable v1 ``session/set_config_option``; fall back to Copilot's
pre-stabilization ``session/set_model`` extension only when no model config
option is advertised. A reported model list is authoritative: unknown and
policy-disabled ids return None instead of being sent.
"""
session_id = str(session.get("sessionId") or "").strip()
requested_model = str(requested_model or "").strip()
if not session_id or not requested_model or requested_model == "copilot-acp":
return None
config_options = [
o for o in (session.get("configOptions") or []) if isinstance(o, dict)
]
model_option = next(
(
o for o in config_options
if o.get("category") == "model" or o.get("id") == "model"
),
(o for o in (session.get("configOptions") or [])
if isinstance(o, dict) and (o.get("category") == "model" or o.get("id") == "model")),
None,
)
if model_option is not None:
enabled_values = {
str(o.get("value") or "").strip()
for o in (model_option.get("options") or [])
if isinstance(o, dict)
and str(
((o.get("_meta") or {}).get("copilotEnablement")) or ""
).strip().lower() != "disabled"
}
if requested_model not in enabled_values:
if requested_model not in _enabled_ids(model_option.get("options"), "value"):
return None
return (
"session/set_config_option",
{
"sessionId": session_id,
"configId": str(model_option.get("id") or "model"),
"value": requested_model,
},
)
advertised = [
m
for m in ((session.get("models") or {}).get("availableModels") or [])
if isinstance(m, dict)
]
available = {
str(m.get("modelId") or "").strip()
for m in advertised
if str(
((m.get("_meta") or {}).get("copilotEnablement")) or ""
).strip().lower() != "disabled"
}
return "session/set_config_option", {
"sessionId": session_id, "configId": str(model_option.get("id") or "model"), "value": requested_model,
}
available = _enabled_ids((session.get("models") or {}).get("availableModels"), "modelId")
if available and requested_model not in available:
return None
return (
"session/set_model",
{"sessionId": session_id, "modelId": requested_model},
)
return "session/set_model", {"sessionId": session_id, "modelId": requested_model}
def _format_messages_as_prompt(
messages: list[dict[str, Any]],
model: str | None = None,
tools: list[dict[str, Any]] | None = None,
messages: list[dict[str, Any]], model: str | None = None, tools: list[dict[str, Any]] | None = None,
tool_choice: Any = None,
) -> str:
sections: list[str] = [
"You are being used as the active ACP agent backend for Hermes.",
"Use ACP capabilities to complete tasks.",
"IMPORTANT: If you take an action with a tool, you MUST output tool calls using <tool_call>{...}</tool_call> blocks with JSON exactly in OpenAI function-call shape.",
"If no tool is needed, answer normally.",
]
# Deliberately no "requested model" line: the model is applied for real via ACP
# session/set_model; a prompt-text mention makes a substituted backend model
# FALSELY self-identify as the requested one. Identity comes from the backend.
# Copilot has no tools of its own that collide with Hermes', so forward the whole toolset.
sections.extend(_render_tool_bridge_sections(tools, tool_choice))
sections: list[str] = [*_PROMPT_PREAMBLE, *_render_tool_bridge_sections(tools, tool_choice)]
transcript: list[str] = []
for message in messages:
if not isinstance(message, dict):
continue
role = str(message.get("role") or "unknown").strip().lower()
if role not in _ROLE_LABELS:
role = "context"
rendered = _render_message_content(message.get("content"))
if rendered:
transcript.append(f"{_ROLE_LABELS[role]}:\n{rendered}")
transcript.append(f"{_ROLE_LABELS.get(role, 'Context')}:\n{rendered}")
if transcript:
sections.append("Conversation transcript:\n\n" + "\n\n".join(transcript))
sections.append("Continue the conversation from the latest user request.")
@@ -239,23 +223,18 @@ def _format_messages_as_prompt(
def _render_message_content(content: Any) -> str:
if content is None:
return ""
if isinstance(content, str):
return content.strip()
if isinstance(content, dict):
if "text" in content:
return str(content.get("text") or "").strip()
if isinstance(content.get("content"), str):
return str(content.get("content") or "").strip()
return content["content"].strip()
return json.dumps(content, ensure_ascii=True)
if isinstance(content, list):
parts: list[str] = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict):
text = item.get("text")
if isinstance(text, str) and text.strip():
parts.append(text.strip())
parts = [
item if isinstance(item, str) else item["text"].strip()
for item in content
if isinstance(item, str) or (isinstance(item, dict) and isinstance(item.get("text"), str) and item["text"].strip())
]
return "\n".join(parts).strip()
return str(content).strip()
@@ -279,9 +258,9 @@ def _effective_timeout(timeout: Any) -> float:
return _DEFAULT_TIMEOUT_SECONDS
if isinstance(timeout, (int, float)):
return float(timeout)
_candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
_numeric = [float(v) for v in _candidates if isinstance(v, (int, float))]
return max(_numeric) if _numeric else _DEFAULT_TIMEOUT_SECONDS
candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
numeric = [float(v) for v in candidates if isinstance(v, (int, float))]
return max(numeric) if numeric else _DEFAULT_TIMEOUT_SECONDS
def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
@@ -293,13 +272,11 @@ def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
content = path.read_text(encoding="utf-8")
except FileNotFoundError:
content = ""
line = params.get("line")
limit = params.get("limit")
line, limit = params.get("line"), params.get("limit")
if isinstance(line, int) and line > 1:
lines = content.splitlines(keepends=True)
start = line - 1
end = start + limit if isinstance(limit, int) and limit > 0 else None
content = "".join(lines[start:end])
end = line - 1 + limit if isinstance(limit, int) and limit > 0 else None
content = "".join(lines[line - 1:end])
if content:
content = redact_sensitive_text(content, force=True)
return {"content": content}
@@ -325,19 +302,6 @@ def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any:
_FS_HANDLERS = {"fs/read_text_file": _fs_read_text_file, "fs/write_text_file": _fs_write_text_file}
class _ACPChatCompletions:
def __init__(self, client: "CopilotACPClient"):
self._client = client
def create(self, **kwargs: Any) -> Any:
return self._client._create_chat_completion(**kwargs)
class _ACPChatNamespace:
def __init__(self, client: "CopilotACPClient"):
self.completions = _ACPChatCompletions(client)
class CopilotACPClient:
"""Minimal OpenAI-client-compatible facade for Copilot ACP."""
@@ -348,17 +312,9 @@ class CopilotACPClient:
HERMES_SKIP_ASYNC_WRAP = True
def __init__(
self,
*,
api_key: str | None = None,
base_url: str | None = None,
default_headers: dict[str, str] | None = None,
acp_command: str | None = None,
acp_args: list[str] | None = None,
acp_cwd: str | None = None,
command: str | None = None,
args: list[str] | None = None,
**_: Any,
self, *, api_key: str | None = None, base_url: str | None = None, default_headers: dict[str, str] | None = None,
acp_command: str | None = None, acp_args: list[str] | None = None, acp_cwd: str | None = None,
command: str | None = None, args: list[str] | None = None, **_: Any,
):
self.api_key = api_key or "copilot-acp"
self.base_url = base_url or ACP_MARKER_BASE_URL
@@ -366,7 +322,7 @@ class CopilotACPClient:
self._acp_command = acp_command or command or _resolve_command()
self._acp_args = list(acp_args or args or _resolve_args())
self._acp_cwd = str(Path(acp_cwd or os.getcwd()).resolve())
self.chat = _ACPChatNamespace(self)
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
self.is_closed = False
self._active_process: subprocess.Popen[str] | None = None
self._active_process_lock = threading.Lock()
@@ -388,37 +344,23 @@ class CopilotACPClient:
pass
def _create_chat_completion(
self,
*,
model: str | None = None,
messages: list[dict[str, Any]] | None = None,
timeout: float | None = None,
tools: list[dict[str, Any]] | None = None,
tool_choice: Any = None,
stream: bool = False,
**_: Any,
self, *, model: str | None = None, messages: list[dict[str, Any]] | None = None, timeout: float | None = None,
tools: list[dict[str, Any]] | None = None, tool_choice: Any = None, stream: bool = False, **_: Any,
) -> Any:
prompt_text = _format_messages_as_prompt(messages or [], model=model, tools=tools, tool_choice=tool_choice)
response_text, reasoning_text = self._run_prompt(
prompt_text, timeout_seconds=_effective_timeout(timeout), model=model
)
response_text, reasoning_text = self._run_prompt(prompt_text, timeout_seconds=_effective_timeout(timeout), model=model)
tool_calls, cleaned_text = _extract_tool_calls_from_text(response_text)
usage = SimpleNamespace(
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
prompt_tokens_details=SimpleNamespace(cached_tokens=0),
)
assistant_message = SimpleNamespace(
content=cleaned_text,
tool_calls=tool_calls,
reasoning=reasoning_text or None,
reasoning_content=reasoning_text or None,
reasoning_details=None,
content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning_text or None,
reasoning_content=reasoning_text or None, reasoning_details=None,
)
completion = SimpleNamespace(
choices=[SimpleNamespace(message=assistant_message, finish_reason="tool_calls" if tool_calls else "stop")],
usage=SimpleNamespace(
prompt_tokens=0, completion_tokens=0, total_tokens=0, prompt_tokens_details=SimpleNamespace(cached_tokens=0)
),
model=model or "copilot-acp",
)
choice = SimpleNamespace(message=assistant_message, finish_reason="tool_calls" if tool_calls else "stop")
completion = SimpleNamespace(choices=[choice], usage=usage, model=model or "copilot-acp")
return _completion_to_stream_chunks(completion) if stream else completion
def _spawn(self) -> subprocess.Popen[str]:
@@ -443,14 +385,9 @@ class CopilotACPClient:
proc = subprocess.Popen(
[self._acp_command] + self._acp_args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True, encoding='utf-8', errors='replace',
bufsize=1,
cwd=self._acp_cwd,
env=_build_subprocess_env(),
creationflags=windows_hide_flags(),
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, encoding='utf-8', errors='replace', bufsize=1,
cwd=self._acp_cwd, env=_build_subprocess_env(), creationflags=windows_hide_flags(),
)
except FileNotFoundError as exc:
raise RuntimeError(
@@ -482,9 +419,7 @@ class CopilotACPClient:
inbox.put({"raw": line.rstrip("\n")})
def _stderr_reader() -> None:
if proc.stderr is None:
return
for line in proc.stderr:
for line in proc.stderr or ():
stderr_tail.append(line.rstrip("\n"))
threading.Thread(target=_stdout_reader, daemon=True).start()
@@ -497,20 +432,15 @@ class CopilotACPClient:
request_id = next_id
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + "\n")
proc.stdin.flush()
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
if proc.poll() is not None:
break
while time.monotonic() < deadline and proc.poll() is None:
try:
msg = inbox.get(timeout=0.1)
except queue.Empty:
continue
if self._handle_server_message(
msg, process=proc, cwd=self._acp_cwd, text_parts=text_parts, reasoning_parts=reasoning_parts
):
continue
if msg.get("id") != request_id:
) or msg.get("id") != request_id:
continue
if "error" in msg:
err = msg.get("error") or {}
@@ -520,85 +450,48 @@ class CopilotACPClient:
stderr_text = "\n".join(stderr_tail).strip()
if proc.poll() is not None and stderr_text:
if _is_gh_copilot_deprecation_message(stderr_text):
raise RuntimeError(
"Hermes ACP mode requires the NEW GitHub Copilot CLI "
"(github.com/github/copilot-cli), but the binary it just "
"spawned is the deprecated `gh copilot` extension.\n\n"
"Install the new CLI:\n"
" npm install -g @github/copilot\n"
" # then verify with: copilot --help\n\n"
"If `copilot` already resolves to the new CLI but you still see this,\n"
"point Hermes at it explicitly:\n"
" export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
"Alternative: use the `copilot` provider (no ACP, hits the Copilot API\n"
"directly with a Copilot subscription token) via `hermes setup`.\n\n"
f"Original error:\n{stderr_text}"
)
raise RuntimeError(_DEPRECATED_CLI_ERROR + stderr_text)
raise RuntimeError(f"Copilot ACP process exited early: {stderr_text}")
raise TimeoutError(f"Timed out waiting for Copilot ACP response to {method}.")
try:
_request(
"initialize",
{
"protocolVersion": 1,
"clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
"clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
},
)
_request("initialize", _INITIALIZE_PARAMS)
session = _request("session/new", {"cwd": self._acp_cwd, "mcpServers": []}) or {}
session_id = str(session.get("sessionId") or "").strip()
if not session_id:
raise RuntimeError("Copilot ACP did not return a sessionId.")
# Prefer the stable ACP v1 session-config API (category="model" select
# option + session/set_config_option); session/set_model is the fallback.
if requested_model and requested_model != "copilot-acp":
try:
selection = _model_selection_request(session, requested_model)
if selection is not None:
method, params = selection
_request(method, params)
_request(*selection)
else:
logger.warning(
"Copilot ACP does not offer model %r; using the "
"session default.",
requested_model,
)
logger.warning("Copilot ACP does not offer model %r; using the session default.", requested_model)
except Exception as exc:
logger.warning(
"Copilot ACP model selection for %r failed; continuing "
"with the session default: %s",
requested_model,
exc,
"Copilot ACP model selection for %r failed; continuing with the session default: %s",
requested_model, exc,
)
text_parts: list[str] = []
reasoning_parts: list[str] = []
_request(
"session/prompt",
{"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]},
text_parts=text_parts,
reasoning_parts=reasoning_parts,
"session/prompt", {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]},
text_parts=text_parts, reasoning_parts=reasoning_parts,
)
return "".join(text_parts), "".join(reasoning_parts)
finally:
self.close()
def _handle_server_message(
self,
msg: dict[str, Any],
*,
process: subprocess.Popen[str],
cwd: str,
text_parts: list[str] | None,
reasoning_parts: list[str] | None,
self, msg: dict[str, Any], *, process: subprocess.Popen[str], cwd: str,
text_parts: list[str] | None, reasoning_parts: list[str] | None,
) -> bool:
"""Consume a server->client message; True when handled (notification or request answered)."""
method = msg.get("method")
if not isinstance(method, str):
return False
if method == "session/update":
update = (msg.get("params") or {}).get("update") or {}
kind = str(update.get("sessionUpdate") or "").strip()
@@ -609,22 +502,19 @@ class CopilotACPClient:
elif kind == "agent_thought_chunk" and chunk_text and reasoning_parts is not None:
reasoning_parts.append(chunk_text)
return True
if process.stdin is None:
return True
message_id = msg.get("id")
params = msg.get("params") or {}
if method == "session/request_permission":
response = _permission_denied(message_id)
elif method in _FS_HANDLERS:
try:
response = _jsonrpc_result(message_id, _FS_HANDLERS[method](params, cwd))
response = _jsonrpc_result(message_id, _FS_HANDLERS[method](msg.get("params") or {}, cwd))
except Exception as exc:
response = _jsonrpc_error(message_id, -32602, str(exc))
else:
response = _jsonrpc_error(message_id, -32601, f"ACP client method '{method}' is not supported by Hermes yet.")
process.stdin.write(json.dumps(response) + "\n")
process.stdin.flush()
return True

297
agent/credits_tracker.py
View File

@@ -1,25 +1,13 @@
"""Credits tracking for Nous inference API responses: parses x-nous-credits-*
(and optional x-nous-tool-pool-*) headers into a validated CreditsState, with
"""Credits tracking for Nous inference API responses: parses ``x-nous-credits-*``
(and optional ``x-nous-tool-pool-*``) headers into a validated CreditsState, with
depletion detection (paid_access), subscription-cap used_fraction, and warn-once
schema-version gating. The hardened parser used by all live consumers.
Header schema (x-nous-credits-*; each *-micros balance has a *-usd twin holding
the server's formatted USD string):
version contract/schema version
remaining-micros/-usd total remaining balance
subscription-micros/-usd subscription balance (SIGNED; may be negative/debt)
subscription-limit-* subscription cap (PAIRED/optional)
rollover-micros rolled-over balance
purchased-micros/-usd purchased balance
denominator-kind "subscription_cap" | "none"
paid-access "true" | "false" (STRING!)
disabled-reason reason string (header omitted when null)
as-of-ms server-side timestamp (ms epoch)
Tool-pool headers use a SEPARATE prefix: x-nous-tool-pool-micros (balance) and
x-nous-tool-pool-gated-off ("true" | "false" STRING!).
Money is handled as micros ints only; *_usd values are preserved verbatim as
the raw strings the server sent (never re-parsed to float).
schema-version gating. Header schema (each ``*-micros`` balance has a ``*-usd``
twin holding the server's formatted USD string): version, remaining, subscription
(SIGNED; may be debt), subscription-limit (PAIRED/optional), rollover, purchased,
denominator-kind ("subscription_cap" | "none"), paid-access ("true"|"false" STRING),
disabled-reason (omitted when null), as-of-ms. Tool-pool headers: x-nous-tool-pool-micros,
x-nous-tool-pool-gated-off ("true"|"false" STRING). Money is micros ints only;
``*_usd`` strings are preserved verbatim (never re-parsed to float).
"""
from __future__ import annotations
@@ -29,7 +17,7 @@ import os
import re
import time
from dataclasses import dataclass
from typing import Any, Mapping, Optional
from typing import Any, Callable, Mapping, Optional
from utils import is_truthy_value
@@ -37,14 +25,10 @@ logger = logging.getLogger(__name__)
# Warn-once latch: emit the version-unsupported warning at most once per process.
_version_warning_emitted: bool = False
# Valid denominator kinds (exhaustive set from the API contract).
_VALID_DENOMINATOR_KINDS = frozenset({"subscription_cap", "none"})
# USD format: optional leading minus, one-or-more digits, dot, exactly 2 digits.
_USD_RE = re.compile(r"^-?\d+\.\d{2}$")
_SENTINEL = object() # singleton sentinel for "parse failed"
_SENTINEL = object() # "parse failed"
def _safe_int(value: Any) -> Any:
@@ -59,7 +43,6 @@ def _safe_int(value: Any) -> Any:
def _validate_usd(value: Optional[str]) -> bool:
"""Return True iff value is a non-None string matching ^-?\\d+\\.\\d{2}$."""
return value is not None and bool(_USD_RE.match(value))
@@ -96,13 +79,13 @@ class CreditsState:
@property
def depleted(self) -> bool:
"""Keyed off ``paid_access == False`` ONLY — never ``remaining_micros == 0``,
a false positive when the balance is zero but access is live (renewal pending)."""
"""Keyed off ``paid_access == False`` ONLY — ``remaining_micros == 0`` is a
false positive when the balance is zero but access is live (renewal pending)."""
return not self.paid_access
@property
def used_fraction(self) -> Optional[float]:
"""Fraction of the subscription cap consumed, in [0.0, 1.0]; None without a
"""Fraction of the subscription cap consumed in [0.0, 1.0]; None without a
computable denominator. Guarded on the LIMIT FIELD (the real denominator),
not ``denominator_kind`` (metadata)."""
if not isinstance(self.subscription_limit_micros, int) or self.subscription_limit_micros <= 0:
@@ -118,7 +101,7 @@ CREDITS_RESTORED_TTL_MS = 8000 # the only TTL notice in v1 (depletion-recove
# Usage-gauge bands (ascending): (threshold_fraction, level, label_pct). One
# escalating line showing the HIGHEST band reached (50 → 75 → 90); crossing up
# replaces it, recovering steps it down. The policy derives everything from it.
# replaces it, recovering steps it down.
CREDITS_USAGE_BANDS: tuple[tuple[float, str, int], ...] = ((0.50, "info", 50), (0.75, "warn", 75), (0.90, "warn", 90))
CREDITS_USAGE_KEY = "credits.usage" # single key for the escalating usage notice
@@ -155,14 +138,13 @@ def _sticky_notice(text: str, level: str, key: str) -> AgentNotice:
def is_free_tier_model(model: str, base_url: str = "") -> bool:
"""Return True when *model* is a Nous free-tier model, using ONLY local data.
"""True when *model* is a Nous free-tier model, using ONLY local data.
Zero-network signals: (1) ``:free`` suffix — canonical Nous free SKU marker;
(2) ``stealth/`` prefix — stealth-preview SKUs are free without the suffix
(naming-convention trust: a PAID ``stealth/`` model would wrongly suppress
the banner); (3) a PEEK into ``hermes_cli.models``' pricing cache (filled by
the model picker; a miss never fetches — gateway sessions never run the
picker, so there only 1-2 apply).
the model picker; a miss never fetches — gateway sessions never run the picker).
Fail-open to False (depleted notice still shows): a wrong warning is
recoverable noise; hiding it on a paid model masks a real block.
@@ -178,13 +160,26 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool:
# peek_cached_pricing owns the /v1-suffix and auth-state key details.
pricing = peek_cached_pricing(base_url)
if not pricing:
return False
return _is_model_free(model, pricing)
return bool(pricing) and _is_model_free(model, pricing)
except Exception:
return False
def _sync_notice(
key: str, want: bool, make: Callable[[], AgentNotice], active: set, to_show: list, to_clear: list
) -> Optional[str]:
"""Reconcile one keyed sticky notice; returns ``"shown"``, ``"cleared"`` or None."""
if want and key not in active:
to_show.append(make())
active.add(key)
return "shown"
if key in active and not want:
to_clear.append(key)
active.discard(key)
return "cleared"
return None
def evaluate_credits_notices(
state: CreditsState, latch: dict, *, model_is_free: bool = False,
) -> tuple[list[AgentNotice], list[str]]:
@@ -201,43 +196,35 @@ def evaluate_credits_notices(
to_show: list[AgentNotice] = []
to_clear: list[str] = []
uf = state.used_fraction
active = latch["active"]
# Crossing latch: band notices fire only once uf was observed below the LOWEST
# band, so a session opening mid-range doesn't fire on its first observation
# (the cold-start seed primes this when it WANTS an open-high warning).
if uf is not None and uf < CREDITS_USAGE_BANDS[0][0]:
latch["seen_below_90"] = True
# Grant-spent gate: fires only after this session OBSERVED the grant unspent
# (≥1¢). Opening at grant-spent is a steady STATE (/usage carries it), not an
# event. Unlike seen_below_90, seeds must NOT prime this gate.
if uf is not None and uf < 1.0 and state.subscription_micros >= GRANT_UNSPENT_MIN_MICROS:
latch["seen_grant_unspent"] = True
active = latch["active"]
# Highest band reached (ascending → last match wins); None below all.
# Top-up suppression: with purchased credits the cap gauge is the wrong
# denominator ("90% used" on $50 of top-up is noise; it used to stick
# PERMANENTLY beside grant_spent at >=100%). grant_spent covers the
# cap-reached case; a mid-session top-up flips current_band → None and the
# clear path removes the band line.
current_band: Optional[tuple[float, str, int]] = None
if uf is not None:
if uf is not None and state.purchased_micros <= 0:
for band in CREDITS_USAGE_BANDS:
if uf >= band[0]:
current_band = band
# Top-up suppression: with purchased credits the cap gauge is the wrong
# denominator ("90% used" on $50 of top-up is noise; it used to stick
# PERMANENTLY beside grant_spent at >=100%). grant_spent below covers the
# cap-reached case; a mid-session top-up flips current_band → None and the
# clear path removes the band line.
if state.purchased_micros > 0:
current_band = None
grant_cond = (
state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
)
depleted_cond = not state.paid_access
# ── usage gauge: highest crossed band only; replace on band change (climb or
# step-down); clear below the lowest band or when the denominator vanishes.
shown_band = latch.get("usage_band")
target_band = current_band[2] if (current_band and latch["seen_below_90"]) else None
if target_band != shown_band:
if target_band != latch.get("usage_band"):
if CREDITS_USAGE_KEY in active:
to_clear.append(CREDITS_USAGE_KEY)
active.discard(CREDITS_USAGE_KEY)
@@ -245,15 +232,13 @@ def evaluate_credits_notices(
# Absolute dollars used (a bare "N%" is only meaningful against a Nous
# cap): cap − remaining in micros, clamped [0, cap]; "$?" if a producer
# set the limit without its *_usd. Re-emits on band change only.
_cap_usd = state.subscription_limit_usd or "?"
_level = current_band[1] # type: ignore[index] (current_band set when target_band set)
_lim = state.subscription_limit_micros or 0
_used_micros = max(0, min(_lim, _lim - state.subscription_micros))
_used_usd = f"{_used_micros / 1_000_000:.2f}" if _lim else "?"
_glyph = "⚠" if _level == "warn" else "•"
to_show.append(
_sticky_notice(f"{_glyph} You've used ${_used_usd} of your ${_cap_usd} cap", _level, CREDITS_USAGE_KEY)
)
level = current_band[1] # type: ignore[index] (current_band set when target_band set)
lim = state.subscription_limit_micros or 0
used_usd = f"{max(0, min(lim, lim - state.subscription_micros)) / 1_000_000:.2f}" if lim else "?"
glyph = "⚠" if level == "warn" else "•"
to_show.append(_sticky_notice(
f"{glyph} You've used ${used_usd} of your ${state.subscription_limit_usd or '?'} cap", level, CREDITS_USAGE_KEY
))
active.add(CREDITS_USAGE_KEY)
latch["usage_band"] = target_band
@@ -261,52 +246,62 @@ def evaluate_credits_notices(
# announcement per crossing. A header flicker (uf → None → 1.0) clears the
# line but cannot re-announce; only a renewal re-opening the gate (fresh ≥1¢
# observation) arms the next. .get(): default closed for hand-built latches.
if grant_cond and "credits.grant_spent" not in active and latch.get("seen_grant_unspent", False):
to_show.append(
_sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent")
)
active.add("credits.grant_spent")
grant_cond = (
state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
)
if _sync_notice(
"credits.grant_spent", grant_cond and latch.get("seen_grant_unspent", False),
lambda: _sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent"),
active, to_show, to_clear,
) == "shown":
latch["seen_grant_unspent"] = False
elif "credits.grant_spent" in active and not grant_cond:
to_clear.append("credits.grant_spent")
active.discard("credits.grant_spent")
# ── depleted: suppressed while the model is free (inference still works).
show_depleted = depleted_cond and not model_is_free
if show_depleted and "credits.depleted" not in active:
to_show.append(_sticky_notice("✕ Credit access paused · run /topup to top up", "error", "credits.depleted"))
active.add("credits.depleted")
elif "credits.depleted" in active and not show_depleted:
to_clear.append("credits.depleted")
active.discard("credits.depleted")
if not depleted_cond:
# Genuine recovery only — switching to a free model while still
# depleted must NOT claim access was restored.
to_show.append(AgentNotice(
text="✓ Credit access restored", level="success", kind="ttl",
ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored",
))
depleted_cond = not state.paid_access
if _sync_notice(
"credits.depleted", depleted_cond and not model_is_free,
lambda: _sticky_notice("✕ Credit access paused · run /topup to top up", "error", "credits.depleted"),
active, to_show, to_clear,
) == "cleared" and not depleted_cond:
# Genuine recovery only — switching to a free model while still
# depleted must NOT claim access was restored.
to_show.append(AgentNotice(
text="✓ Credit access restored", level="success", kind="ttl",
ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored",
))
return (to_show, to_clear)
# (field, header, signed) — required micros fields; only subscription may be negative.
_MICROS_FIELDS: tuple[tuple[str, str, bool], ...] = (
("remaining_micros", "x-nous-credits-remaining-micros", False),
("subscription_micros", "x-nous-credits-subscription-micros", True),
("rollover_micros", "x-nous-credits-rollover-micros", False),
("purchased_micros", "x-nous-credits-purchased-micros", False), ("as_of_ms", "x-nous-credits-as-of-ms", False),
)
_USD_FIELDS: tuple[tuple[str, str], ...] = (
("remaining_usd", "x-nous-credits-remaining-usd"), ("subscription_usd", "x-nous-credits-subscription-usd"),
("purchased_usd", "x-nous-credits-purchased-usd"),
)
# (field, header, default-when-absent) — "true"/"false" (case-insensitive) STRING flags.
_BOOL_FIELDS: tuple[tuple[str, str, bool], ...] = (
("paid_access", "x-nous-credits-paid-access", True), # absent → fail-open (assume access)
("tool_pool_gated_off", "x-nous-tool-pool-gated-off", False),
# Header field table: (field, header, kind). micros: required, non-negative unless
# "signed" (only subscription may be negative). usd: required ^-?\d+\.\d{2}$.
# bool: optional "true"/"false" STRING flag; the tuple's 4th item is the default.
_HEADER_FIELDS: tuple[tuple, ...] = (
("remaining_micros", "x-nous-credits-remaining-micros", "micros"),
("subscription_micros", "x-nous-credits-subscription-micros", "signed"),
("rollover_micros", "x-nous-credits-rollover-micros", "micros"),
("purchased_micros", "x-nous-credits-purchased-micros", "micros"),
("as_of_ms", "x-nous-credits-as-of-ms", "micros"),
("remaining_usd", "x-nous-credits-remaining-usd", "usd"),
("subscription_usd", "x-nous-credits-subscription-usd", "usd"),
("purchased_usd", "x-nous-credits-purchased-usd", "usd"),
("paid_access", "x-nous-credits-paid-access", "bool", True), # absent → fail-open (assume access)
("tool_pool_gated_off", "x-nous-tool-pool-gated-off", "bool", False),
)
def _parse_field(kind: str, raw: Optional[str], default: Any = None) -> Any:
"""One header value → field value, or ``_SENTINEL`` on a contract violation."""
if kind in ("micros", "signed"):
val = _safe_int(raw)
return _SENTINEL if val is _SENTINEL or (kind == "micros" and val < 0) else val
if kind == "usd":
return raw if _validate_usd(raw) else _SENTINEL
if raw is None:
return default
flag = raw.strip().lower()
return _SENTINEL if flag not in ("true", "false") else flag == "true"
def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Optional[CreditsState]:
"""Parse x-nous-credits-* (and x-nous-tool-pool-*) headers into a CreditsState.
@@ -336,45 +331,29 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt
logger.warning("credits header version %d unsupported, ignoring — update Hermes", version_val)
return None
fields: dict[str, Any] = {}
for name, key, signed in _MICROS_FIELDS:
val = _safe_int(lowered.get(key))
if val is _SENTINEL or (not signed and val < 0):
for name, key, kind, *default in _HEADER_FIELDS:
val = _parse_field(kind, lowered.get(key), *default)
if val is _SENTINEL:
return None
fields[name] = val
# tool_pool_micros is OPTIONAL: absent → 0; present-but-invalid → miss.
_tp_raw = lowered.get("x-nous-tool-pool-micros")
_tp_val = 0 if _tp_raw is None else _safe_int(_tp_raw)
if _tp_val is _SENTINEL or _tp_val < 0:
tp_raw = lowered.get("x-nous-tool-pool-micros")
tp_val = 0 if tp_raw is None else _parse_field("micros", tp_raw)
if tp_val is _SENTINEL:
return None
fields["tool_pool_micros"] = _tp_val
for name, key in _USD_FIELDS:
val = lowered.get(key, "")
if not _validate_usd(val):
return None
fields[name] = val
fields["tool_pool_micros"] = tp_val
# subscription_limit_* PAIRED + OPTIONAL: both present → validate both
# (any invalid → miss); half-pair or both absent → both None, parse continues.
sub_limit_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
sub_limit_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
if sub_limit_micros_raw is not None and sub_limit_usd_raw is not None:
lm = _safe_int(sub_limit_micros_raw)
if lm is _SENTINEL or lm < 0 or not _validate_usd(sub_limit_usd_raw):
lim_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
lim_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
if lim_micros_raw is not None and lim_usd_raw is not None:
lm = _parse_field("micros", lim_micros_raw)
if lm is _SENTINEL or not _validate_usd(lim_usd_raw):
return None
fields["subscription_limit_micros"] = lm
fields["subscription_limit_usd"] = sub_limit_usd_raw
fields["subscription_limit_micros"], fields["subscription_limit_usd"] = lm, lim_usd_raw
denominator_kind = lowered.get("x-nous-credits-denominator-kind", "none")
if denominator_kind not in _VALID_DENOMINATOR_KINDS:
return None
for name, key, default in _BOOL_FIELDS:
if key not in lowered:
fields[name] = default
continue
raw = lowered[key].strip().lower()
if raw not in ("true", "false"):
return None
fields[name] = raw == "true"
return CreditsState(
version=version_val,
denominator_kind=denominator_kind,
@@ -426,21 +405,18 @@ _DEV_FIXTURES: dict[str, dict] = {
def dev_fixture_credits_state() -> Optional[CreditsState]:
"""Return a fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None
(unknown name / "clear" / "none" / unset → None).
Hard prod-leak guard: applies ONLY when HERMES_DEV_CREDITS is also on, so a
stray fixture env var can never surface fabricated balances on a real account.
"""
"""Fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None (unknown name /
"clear" / "none" / unset). Hard prod-leak guard: applies ONLY when
HERMES_DEV_CREDITS is also on, so a stray fixture env var can never surface
fabricated balances on a real account."""
if not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
return None
raw = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
if not raw:
name = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
if not name:
return None
name = raw
if os.path.sep in raw or "/" in raw: # looks like a path → read the name from the file
if os.path.sep in name or "/" in name: # looks like a path → read the name from the file
try:
with open(raw, "r", encoding="utf-8") as fh:
with open(name, "r", encoding="utf-8") as fh:
name = fh.read().strip()
except OSError:
return None
@@ -450,8 +426,7 @@ def dev_fixture_credits_state() -> Optional[CreditsState]:
# Stamp what the REAL parser always guarantees so a fixture is field-identical
# to a parse_credits_headers() result (differential test): version 1, and a
# valid purchased_usd (a zero-top-up account still carries "0.00").
merged = {"version": 1, "purchased_usd": "0.00", **spec}
return CreditsState(**merged, from_header=True, captured_at=time.time())
return CreditsState(**{"version": 1, "purchased_usd": "0.00", **spec}, from_header=True, captured_at=time.time())
def _credits_state_from_account(info) -> Optional[CreditsState]:
@@ -459,26 +434,26 @@ def _credits_state_from_account(info) -> Optional[CreditsState]:
Float account dollars → micros plus a DISPLAY *_usd (formatting account floats
is allowed; parsing a server *_usd is not). Fail-open → None."""
try:
_acc = getattr(info, "paid_service_access_info", None)
_sub = getattr(info, "subscription", None)
acc = getattr(info, "paid_service_access_info", None)
sub = getattr(info, "subscription", None)
def _money(dollars) -> tuple[int, str]: # (micros, display usd); (0, "") when absent
if isinstance(dollars, (int, float)):
return int(round(dollars * 1_000_000)), f"{dollars:.2f}"
return 0, ""
_remaining = _money(getattr(_acc, "total_usable_credits", None))
_sub_rem = _money(getattr(_acc, "subscription_credits_remaining", None))
_purchased = _money(getattr(_acc, "purchased_credits_remaining", None))
_monthly = getattr(_sub, "monthly_credits", None)
_cap = _money(_monthly) if isinstance(_monthly, (int, float)) and _monthly > 0 else (None, None)
_paid = getattr(info, "paid_service_access", None)
remaining = _money(getattr(acc, "total_usable_credits", None))
sub_rem = _money(getattr(acc, "subscription_credits_remaining", None))
purchased = _money(getattr(acc, "purchased_credits_remaining", None))
monthly = getattr(sub, "monthly_credits", None)
cap = _money(monthly) if isinstance(monthly, (int, float)) and monthly > 0 else (None, None)
paid = getattr(info, "paid_service_access", None)
return CreditsState(
remaining_micros=_remaining[0], remaining_usd=_remaining[1], subscription_micros=_sub_rem[0],
subscription_usd=_sub_rem[1], subscription_limit_micros=_cap[0], subscription_limit_usd=_cap[1],
purchased_micros=_purchased[0], purchased_usd=_purchased[1],
rollover_micros=_money(getattr(_sub, "rollover_credits", None))[0],
denominator_kind="subscription_cap" if _cap[0] is not None else "none",
paid_access=_paid if isinstance(_paid, bool) else True, from_header=False, captured_at=time.time(),
remaining_micros=remaining[0], remaining_usd=remaining[1], subscription_micros=sub_rem[0],
subscription_usd=sub_rem[1], subscription_limit_micros=cap[0], subscription_limit_usd=cap[1],
purchased_micros=purchased[0], purchased_usd=purchased[1],
rollover_micros=_money(getattr(sub, "rollover_credits", None))[0],
denominator_kind="subscription_cap" if cap[0] is not None else "none",
paid_access=paid if isinstance(paid, bool) else True, from_header=False, captured_at=time.time(),
)
except Exception:
logger.debug("credits ▸ seed account→state mapping failed", exc_info=True)
@@ -492,11 +467,11 @@ def _hydrate_seed_state(agent, state) -> None:
agent._credits_state = state
if getattr(agent, "_credits_session_start_micros", None) is None:
agent._credits_session_start_micros = state.remaining_micros
_latch = getattr(agent, "_credits_latch", None)
if isinstance(_latch, dict) and state.used_fraction is not None:
latch = getattr(agent, "_credits_latch", None)
if isinstance(latch, dict) and state.used_fraction is not None:
# Prime ONLY seen_below_90. Never prime seen_grant_unspent: a seed
# observing grant-spent is a steady state; priming revives the nag.
_latch["seen_below_90"] = True
latch["seen_below_90"] = True
emit = getattr(agent, "_emit_credits_notices", None)
if callable(emit):
emit()
@@ -511,9 +486,7 @@ def seed_credits_at_session_start(agent) -> bool:
Returns True iff it seeded this call. Never raises — credits must never block startup.
"""
try:
if getattr(agent, "provider", "") != "nous":
return False
if getattr(agent, "_credits_state", None) is not None:
if getattr(agent, "provider", "") != "nous" or getattr(agent, "_credits_state", None) is not None:
return False
try:
fixture = dev_fixture_credits_state()

23
agent/fast_mode.py
View File

@@ -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

396
agent/gemini_native_adapter.py
View File

@@ -1,11 +1,10 @@
"""OpenAI-compatible facade over Google AI Studio's native Gemini API.
Hermes keeps ``api_mode='chat_completions'`` for the ``gemini`` provider so the
agent loop keeps its OpenAI-shaped message flow; this shim converts those
``messages[]`` / ``tools[]`` requests into ``models/{model}:generateContent``
payloads and converts the responses back. Google's OpenAI-compat endpoint has
been brittle for the multi-turn tool loop (auth churn, tool-call replay quirks,
thought-signature requirements); the native API is the canonical path.
The ``gemini`` provider keeps ``api_mode='chat_completions'`` so the agent loop
stays OpenAI-shaped; this shim converts ``messages[]``/``tools[]`` requests into
``models/{model}:generateContent`` payloads and converts responses back. Google's
OpenAI-compat endpoint is brittle for the multi-turn tool loop (auth churn,
tool-call replay quirks, thought-signature requirements); native is canonical.
"""
from __future__ import annotations
@@ -41,36 +40,54 @@ DEFAULT_GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
# Gemini's native API applies a low internal default and truncates output.
GEMINI_DEFAULT_MAX_OUTPUT_TOKENS = 65535
_FREE_TIER_GUIDANCE = (
"\n\nYour Google API key is on the free tier (a few hundred requests/day for Gemini Flash models). "
"Hermes typically makes 3-10 API calls per user turn, so the free tier is exhausted in a handful of "
"messages and cannot sustain an agent session. Enable billing on your Google Cloud project and "
"regenerate the key in a billing-enabled project: https://aistudio.google.com/apikey"
)
_STANDARD_KEY_GUIDANCE = (
"\n\nGoogle Gemini rejected this API key's type — you do NOT need OAuth. Google began rejecting legacy "
"'Standard' Google Cloud keys for the Gemini API on June 19, 2026, and all Standard keys stop working in "
"September 2026. Open https://aistudio.google.com/api-keys, check the key's type and status, and create a "
"replacement Gemini API key (or, as a temporary bridge, restrict the Standard key to "
"generativelanguage.googleapis.com). Then update GEMINI_API_KEY / GOOGLE_API_KEY in ~/.hermes/.env and "
"restart your session. Details: https://ai.google.dev/gemini-api/docs/api-key"
)
# Stands in for a model turn that never arrived (stream failure / interrupt /
# quota fallback) when a human user text turn directly follows a tool-result
# turn, keeping the request alternation-valid while the user's message stays a
# turn of its own (mirrors gemini-cli's placeholder repair).
_INTERRUPTED_RESPONSE_PLACEHOLDER = "[The previous response was interrupted before it completed.]"
# Cross-provider tool_calls (e.g. fallback from xAI/Anthropic) carry no Gemini
# thoughtSignature; without this sentinel Gemini 3 thinking models reject
# replayed history with 400 INVALID_ARGUMENT.
_SKIP_SIGNATURE = "skip_thought_signature_validator"
_TOOL_CHOICE_MODES = {"auto": "AUTO", "required": "ANY", "none": "NONE"}
_FINISH_REASON_MAP = {
"STOP": "stop", "MAX_TOKENS": "length", "SAFETY": "content_filter", "RECITATION": "content_filter", "OTHER": "stop",
}
_HTTP_ERROR_CODES = {401: "gemini_unauthorized", 429: "gemini_rate_limited", 404: "gemini_model_not_found"}
def bare_gemini_model_id(model: str) -> str:
"""Strip Gemini's own provider prefix from an aggregator-style model id."""
name = (model or "").strip()
lowered = name.lower()
for prefix in ("google/", "gemini/"):
if lowered.startswith(prefix):
if name.lower().startswith(prefix):
return name[len(prefix):].strip() or name
return name
def _gemini_major_version(model: str) -> Optional[int]:
"""Extract the major version from a Gemini model id (``gemini-3.6-flash`` → 3)."""
match = re.match(r"gemini-(\d+)", bare_gemini_model_id(model).lower())
return int(match.group(1)) if match else None
def gemini_requires_tool_call_ids(model: str) -> bool:
"""Whether functionCall/functionResponse parts must carry explicit ids.
Gemini 3+ needs explicit ids so replayed parallel tool calls pair with
their responses; Gemini 2.x rejects unexpected ``id`` fields, so gate on
the major version.
"""
version = _gemini_major_version(model)
return version is not None and version >= 3
"""Gemini 3+ needs explicit functionCall/functionResponse ids so replayed
parallel tool calls pair with their responses; 2.x rejects the field."""
match = re.match(r"gemini-(\d+)", bare_gemini_model_id(model).lower())
return match is not None and int(match.group(1)) >= 3
def is_native_gemini_base_url(base_url: str) -> bool:
"""Return True when the endpoint speaks Gemini's native REST API."""
"""True when the endpoint speaks Gemini's native REST API (not ``/openai``)."""
normalized = str(base_url or "").strip().rstrip("/").lower()
return "generativelanguage.googleapis.com" in normalized and not normalized.endswith("/openai")
@@ -78,23 +95,19 @@ def is_native_gemini_base_url(base_url: str) -> bool:
def probe_gemini_tier(
api_key: str, base_url: str = DEFAULT_GEMINI_BASE_URL, *, model: str = "gemini-3.7-flash", timeout: float = 10.0
) -> str:
"""Probe a Google AI Studio API key; return ``"free"``, ``"paid"`` or
``"unknown"`` (probe failed — callers should proceed without blocking)."""
"""Probe a Google AI Studio key; return ``"free"``, ``"paid"`` or ``"unknown"``
(probe failed — callers should proceed without blocking)."""
key = (api_key or "").strip()
if not key:
return "unknown"
normalized_base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL
if normalized_base.lower().endswith("/openai"):
normalized_base = normalized_base[: -len("/openai")]
base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL
if base.lower().endswith("/openai"):
base = base[: -len("/openai")]
payload = {"contents": [{"role": "user", "parts": [{"text": "hi"}]}], "generationConfig": {"maxOutputTokens": 1}}
try:
with httpx.Client(timeout=timeout) as client:
resp = client.post(
f"{normalized_base}/models/{model}:generateContent",
params={"key": key},
json=payload,
f"{base}/models/{model}:generateContent", params={"key": key}, json=payload,
headers={"Content-Type": "application/json", "X-Goog-Api-Client": f"hermes-agent/{_HERMES_VERSION}"},
)
except Exception as exc:
@@ -108,52 +121,32 @@ def probe_gemini_tier(
return "free" if int(rpd_header) <= 1000 else "paid"
except (TypeError, ValueError):
pass
if resp.status_code == 429:
try:
body_text = resp.text or ""
except Exception:
body_text = ""
return "free" if "free_tier" in body_text.lower() else "paid"
return "free" if "free_tier" in _response_text(resp).lower() else "paid"
return "paid" if 200 <= resp.status_code < 300 else "unknown"
def _response_text(response: Any) -> str:
try:
return response.text or ""
except Exception:
return ""
def is_free_tier_quota_error(error_message: str) -> bool:
"""Return True when a Gemini 429 message indicates free-tier exhaustion."""
"""True when a Gemini 429 message indicates free-tier exhaustion."""
return bool(error_message) and "free_tier" in error_message.lower()
_FREE_TIER_GUIDANCE = (
"\n\nYour Google API key is on the free tier (a few hundred requests/day for Gemini Flash models). "
"Hermes typically makes 3-10 API calls per user turn, so the free tier is exhausted in a handful of "
"messages and cannot sustain an agent session. Enable billing on your Google Cloud project and "
"regenerate the key in a billing-enabled project: https://aistudio.google.com/apikey"
)
def is_standard_key_auth_error(status: int, error_message: str, reason: str = "") -> bool:
"""Return True when a Gemini 401 indicates Google rejected the key TYPE.
Google rejects legacy "Standard" Google Cloud API keys with a misleading
401 asking for an OAuth 2 access token, optionally carrying ErrorInfo
reason ``ACCESS_TOKEN_TYPE_UNSUPPORTED``. Scoped narrowly so a plain bad
key (``API_KEY_INVALID``) keeps its existing message.
"""
"""True when a Gemini 401 means Google rejected the key TYPE (legacy "Standard"
Cloud key → misleading "expected OAuth 2 access token" / ErrorInfo
``ACCESS_TOKEN_TYPE_UNSUPPORTED``). Narrow so ``API_KEY_INVALID`` keeps its message."""
if status != 401:
return False
return reason == "ACCESS_TOKEN_TYPE_UNSUPPORTED" or "expected oauth 2 access token" in (error_message or "").lower()
_STANDARD_KEY_GUIDANCE = (
"\n\nGoogle Gemini rejected this API key's type — you do NOT need OAuth. Google began rejecting legacy "
"'Standard' Google Cloud keys for the Gemini API on June 19, 2026, and all Standard keys stop working in "
"September 2026. Open https://aistudio.google.com/api-keys, check the key's type and status, and create a "
"replacement Gemini API key (or, as a temporary bridge, restrict the Standard key to "
"generativelanguage.googleapis.com). Then update GEMINI_API_KEY / GOOGLE_API_KEY in ~/.hermes/.env and "
"restart your session. Details: https://ai.google.dev/gemini-api/docs/api-key"
)
class GeminiAPIError(Exception):
"""Error shape compatible with Hermes retry/error classification."""
@@ -163,26 +156,23 @@ class GeminiAPIError(Exception):
details: Optional[Dict[str, Any]] = None,
) -> None:
super().__init__(message)
self.code = code
self.status_code = status_code
self.response = response
self.retry_after = retry_after
self.details = details or {}
self.code, self.status_code, self.response = code, status_code, response
self.retry_after, self.details = retry_after, details or {}
# ── OpenAI → Gemini request translation ──────────────────────────────────────
def _coerce_content_to_text(content: Any) -> str:
if content is None:
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
pieces: List[str] = []
for part in content:
if isinstance(part, str):
pieces.append(part)
elif isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
pieces.append(part["text"])
return "\n".join(pieces)
return "\n".join(
p if isinstance(p, str) else p["text"]
for p in content
if isinstance(p, str) or (isinstance(p, dict) and p.get("type") == "text" and isinstance(p.get("text"), str))
)
return str(content)
@@ -190,21 +180,17 @@ def _extract_multimodal_parts(content: Any) -> List[Dict[str, Any]]:
if not isinstance(content, list):
text = _coerce_content_to_text(content)
return [{"text": text}] if text else []
parts: List[Dict[str, Any]] = []
for item in content:
if isinstance(item, str):
parts.append({"text": item})
elif not isinstance(item, dict):
continue
if not isinstance(item, dict):
continue
ptype = item.get("type")
if ptype == "text":
text = item.get("text")
if isinstance(text, str) and text:
parts.append({"text": text})
elif ptype == "image_url":
url = ((item.get("image_url") or {}).get("url") or "")
elif item.get("type") == "text":
if isinstance(item.get("text"), str) and item["text"]:
parts.append({"text": item["text"]})
elif item.get("type") == "image_url":
url = (item.get("image_url") or {}).get("url") or ""
if not isinstance(url, str) or not url.startswith("data:"):
continue
try:
@@ -226,13 +212,6 @@ def _tool_call_extra_signature(tool_call: Dict[str, Any]) -> Optional[str]:
return sig if isinstance(sig, str) and sig else None
# Stands in for a model turn that never arrived (stream failure / interrupt /
# quota fallback) when history leaves a human user text turn directly after a
# tool-result turn, keeping the request alternation-valid while the user's
# message remains a turn of its own (mirrors gemini-cli's placeholder repair).
_INTERRUPTED_RESPONSE_PLACEHOLDER = "[The previous response was interrupted before it completed.]"
def _tool_call_id(tool_call: Dict[str, Any]) -> str:
return str(tool_call.get("id") or tool_call.get("call_id") or "")
@@ -246,25 +225,19 @@ def _translate_tool_call_to_gemini(tool_call: Dict[str, Any], include_ids: bool
args = {"_raw": args_raw}
if not isinstance(args, dict):
args = {"_value": args}
part: Dict[str, Any] = {"functionCall": {"name": str(fn.get("name") or ""), "args": args}}
call: Dict[str, Any] = {"name": str(fn.get("name") or ""), "args": args}
if include_ids and _tool_call_id(tool_call):
part["functionCall"]["id"] = _tool_call_id(tool_call)
# Cross-provider tool_calls (e.g. fallback from xAI/Anthropic) carry no
# Gemini thoughtSignature; without the sentinel, Gemini 3 thinking models
# reject replayed history with 400 INVALID_ARGUMENT.
part["thoughtSignature"] = _tool_call_extra_signature(tool_call) or "skip_thought_signature_validator"
return part
call["id"] = _tool_call_id(tool_call)
return {"functionCall": call, "thoughtSignature": _tool_call_extra_signature(tool_call) or _SKIP_SIGNATURE}
def _looks_like_json_schema(node: Any) -> bool:
"""True if a parsed value contains a JSON-Schema-style ``$ref`` pointer (``#/...``).
"""True if a parsed value contains a JSON-Schema ``$ref`` pointer (``#/...``).
Gemini 3 resolves ``$ref``/``$defs`` inside a functionResponse.response
payload and rejects unknown pointers with HTTP 400, so a tool result that
is itself a JSON Schema (e.g. ``tool_describe`` output) must be forwarded
as opaque text. Detection is structural: false positives only lose the
structured shape, never the content.
Gemini 3 resolves ``$ref``/``$defs`` inside functionResponse.response and
rejects unknown pointers with HTTP 400, so a tool result that is itself a
JSON Schema (e.g. ``tool_describe`` output) must be forwarded as opaque
text. False positives only lose the structured shape, never the content.
"""
if isinstance(node, dict):
return any(
@@ -278,12 +251,11 @@ def _translate_tool_result_to_gemini(
message: Dict[str, Any], tool_name_by_call_id: Optional[Dict[str, str]] = None, include_ids: bool = False,
*, is_gemini3: bool = False,
) -> Dict[str, Any]:
tool_name_by_call_id = tool_name_by_call_id or {}
tool_call_id = str(message.get("tool_call_id") or "")
# Gemini requires functionResponse.name to echo the matching
# functionCall.name, so the call-id mapping beats the result's own name
# (which may be an unwrapped internal name, e.g. an MCP tool via `tool_call`).
name = str(tool_name_by_call_id.get(tool_call_id) or message.get("name") or tool_call_id or "tool")
# functionResponse.name must echo the matching functionCall.name, so the
# call-id mapping beats the result's own name (which may be an unwrapped
# internal name, e.g. an MCP tool via `tool_call`).
name = str((tool_name_by_call_id or {}).get(tool_call_id) or message.get("name") or tool_call_id or "tool")
raw_content = message.get("content")
content = _coerce_content_to_text(raw_content)
try:
@@ -294,8 +266,7 @@ def _translate_tool_result_to_gemini(
function_response: Dict[str, Any] = {"name": name, "response": response}
if include_ids and tool_call_id:
function_response["id"] = tool_call_id
# Gemini 3.x accepts images inside functionResponse.parts; 2.x rejects
# the field, so older models get the text-only downgrade.
# Gemini 3.x accepts images inside functionResponse.parts; 2.x rejects the field.
if is_gemini3:
image_parts = [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p]
if image_parts:
@@ -318,7 +289,6 @@ def _build_gemini_contents(
if not isinstance(msg, dict):
continue
role = str(msg.get("role") or "user")
if role == "system":
system_text_parts.append(_coerce_content_to_text(msg.get("content")))
continue
@@ -334,47 +304,41 @@ def _build_gemini_contents(
if not isinstance(tool_call, dict):
continue
tool_call_id = _tool_call_id(tool_call)
tool_name = str(((tool_call.get("function") or {}).get("name") or ""))
tool_name = str((tool_call.get("function") or {}).get("name") or "")
if tool_call_id and tool_name:
tool_name_by_call_id[tool_call_id] = tool_name
parts.append(_translate_tool_call_to_gemini(tool_call, include_ids=include_tool_call_ids))
if parts:
contents.append({"role": "model" if role == "assistant" else "user", "parts": parts})
# Alternation contract for generateContent:
# 1) Adjacent same-role contents merge (consecutive same-role contents are
# rejected with HTTP 400 "multiturn requests [must] alternate").
# 1) Adjacent same-role contents merge (else HTTP 400 "multiturn requests
# [must] alternate").
# 2) Exception: never fuse a human user text turn into a preceding user
# content that only carries functionResponse parts (or vice versa) —
# Gemini 3 accepts the fold but reads the text as a continuation of the
# tool result and returns an empty candidate. Parallel tool results
# (functionResponse + functionResponse) still merge.
# 3) The split pair is kept API-valid by interposing a placeholder model
# turn between the functionResponse content and the human text.
merged_contents: List[Dict[str, Any]] = []
# 3) The split pair stays API-valid via an interposed placeholder model turn.
merged: List[Dict[str, Any]] = []
for content in contents:
prev = merged_contents[-1] if merged_contents else None
prev = merged[-1] if merged else None
same_role = prev is not None and prev["role"] == content["role"]
if same_role and content["role"] == "user" and _has_function_response(prev) != _has_function_response(content):
same_role = False
merged_contents.append({"role": "model", "parts": [{"text": _INTERRUPTED_RESPONSE_PLACEHOLDER}]})
merged.append({"role": "model", "parts": [{"text": _INTERRUPTED_RESPONSE_PLACEHOLDER}]})
if same_role:
merged_contents[-1]["parts"].extend(content["parts"])
merged[-1]["parts"].extend(content["parts"])
else:
merged_contents.append(content)
merged.append(content)
joined_system = "\n".join(part for part in system_text_parts if part).strip()
system_instruction = {"role": "system", "parts": [{"text": joined_system}]} if joined_system else None
return merged_contents, system_instruction
return merged, ({"role": "system", "parts": [{"text": joined_system}]} if joined_system else None)
def _translate_tools_to_gemini(tools: Any) -> List[Dict[str, Any]]:
if not isinstance(tools, list):
return []
declarations: List[Dict[str, Any]] = []
for tool in tools:
for tool in tools if isinstance(tools, list) else []:
fn = (tool.get("function") or {}) if isinstance(tool, dict) else None
if not isinstance(fn, dict) or not (isinstance(fn.get("name"), str) and fn["name"]):
continue
@@ -387,9 +351,6 @@ def _translate_tools_to_gemini(tools: Any) -> List[Dict[str, Any]]:
return [{"functionDeclarations": declarations}] if declarations else []
_TOOL_CHOICE_MODES = {"auto": "AUTO", "required": "ANY", "none": "NONE"}
def _translate_tool_choice_to_gemini(tool_choice: Any) -> Optional[Dict[str, Any]]:
if isinstance(tool_choice, str) and tool_choice in _TOOL_CHOICE_MODES:
return {"functionCallingConfig": {"mode": _TOOL_CHOICE_MODES[tool_choice]}}
@@ -417,12 +378,9 @@ def _normalize_thinking_config(config: Any) -> Optional[Dict[str, Any]]:
def _thinking_requests_output_headroom(thinking_config: Any) -> bool:
"""True when Gemini will spend output tokens on thinking.
Thought tokens bill against ``maxOutputTokens``; a global 4096/16384
``max_tokens`` can be consumed entirely by high thinking, leaving
``finishReason=MAX_TOKENS`` with no answer.
"""
"""True when Gemini will spend output tokens on thinking (thought tokens bill
against ``maxOutputTokens``; a global 4096/16384 cap can be consumed entirely
by high thinking, leaving ``finishReason=MAX_TOKENS`` with no answer)."""
normalized = _normalize_thinking_config(thinking_config)
if not normalized:
return False
@@ -433,10 +391,9 @@ def _thinking_requests_output_headroom(thinking_config: Any) -> bool:
def _effective_gemini_max_output_tokens(max_tokens: Optional[int], thinking_config: Any) -> int:
"""Resolve native ``maxOutputTokens``: an omitted/invalid cap becomes the
published ceiling (Gemini truncates on its low internal default), and an
explicit cap is raised to that ceiling when thinking is enabled so thought
tokens do not starve the answer."""
"""Native ``maxOutputTokens``: an omitted/invalid cap becomes the published
ceiling (Gemini truncates on its low internal default); an explicit cap is
raised to the ceiling when thinking is enabled so thoughts don't starve the answer."""
try:
requested = int(max_tokens)
except (TypeError, ValueError):
@@ -479,10 +436,7 @@ def build_gemini_request(
return request
_FINISH_REASON_MAP = {
"STOP": "stop", "MAX_TOKENS": "length", "SAFETY": "content_filter", "RECITATION": "content_filter", "OTHER": "stop",
}
# ── Gemini → OpenAI response translation ─────────────────────────────────────
def _map_gemini_finish_reason(reason: str) -> str:
return _FINISH_REASON_MAP.get(str(reason or "").upper(), "stop")
@@ -494,7 +448,7 @@ def _tool_call_extra_from_part(part: Dict[str, Any]) -> Optional[Dict[str, Any]]
def _new_call_id(fc: Dict[str, Any]) -> str:
"""Echo Gemini's functionCall id when present, else mint an OpenAI-style one."""
"""Echo the functionCall/delta ``id`` when present, else mint an OpenAI-style one."""
fc_id = fc.get("id")
return fc_id if isinstance(fc_id, str) and fc_id else f"call_{uuid.uuid4().hex[:12]}"
@@ -515,13 +469,6 @@ def _usage_from_metadata(usage_meta: Dict[str, Any]) -> SimpleNamespace:
)
def _completion(model: str, message: SimpleNamespace, finish_reason: str, usage: SimpleNamespace) -> SimpleNamespace:
return SimpleNamespace(
id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object="chat.completion", created=int(time.time()), model=model,
choices=[SimpleNamespace(index=0, message=message, finish_reason=finish_reason)], usage=usage,
)
def _assistant_message(content: Any, tool_calls: Any, reasoning: Any) -> SimpleNamespace:
return SimpleNamespace(
role="assistant", content=content, tool_calls=tool_calls,
@@ -531,41 +478,43 @@ def _assistant_message(content: Any, tool_calls: Any, reasoning: Any) -> SimpleN
def translate_gemini_response(resp: Dict[str, Any], model: str) -> SimpleNamespace:
candidates = resp.get("candidates") or []
if not isinstance(candidates, list) or not candidates:
return _completion(model, _assistant_message("", None, None), "stop", _usage_from_metadata({}))
cand = candidates[0] if isinstance(candidates[0], dict) else {}
content_obj = cand.get("content")
parts = content_obj.get("parts") if isinstance(content_obj, dict) else []
cand = parts = None
if isinstance(candidates, list) and candidates:
cand = candidates[0] if isinstance(candidates[0], dict) else {}
content_obj = cand.get("content")
parts = content_obj.get("parts") if isinstance(content_obj, dict) else []
text_pieces: List[str] = []
reasoning_pieces: List[str] = []
tool_calls: List[SimpleNamespace] = []
for index, part in enumerate(parts or []):
if not isinstance(part, dict):
continue
if part.get("thought") is True and isinstance(part.get("text"), str):
reasoning_pieces.append(part["text"])
continue
if isinstance(part.get("text"), str):
elif isinstance(part.get("text"), str):
text_pieces.append(part["text"])
continue
fc = part.get("functionCall")
if isinstance(fc, dict) and fc.get("name"):
elif isinstance(fc := part.get("functionCall"), dict) and fc.get("name"):
tool_call = SimpleNamespace(
id=_new_call_id(fc), type="function", index=index,
function=SimpleNamespace(name=str(fc["name"]), arguments=_dump_call_args(fc)),
)
extra_content = _tool_call_extra_from_part(part)
if extra_content:
if extra_content := _tool_call_extra_from_part(part):
tool_call.extra_content = extra_content
tool_calls.append(tool_call)
finish_reason = "tool_calls" if tool_calls else _map_gemini_finish_reason(str(cand.get("finishReason") or ""))
if cand is None:
finish_reason, usage = "stop", _usage_from_metadata({})
else:
finish_reason = "tool_calls" if tool_calls else _map_gemini_finish_reason(str(cand.get("finishReason") or ""))
usage = _usage_from_metadata(resp.get("usageMetadata") or {})
message = _assistant_message(
"".join(text_pieces) if text_pieces else None, tool_calls or None, "".join(reasoning_pieces) or None
"".join(text_pieces) if text_pieces else ("" if cand is None else None),
tool_calls or None, "".join(reasoning_pieces) or None,
)
return SimpleNamespace(
id=f"chatcmpl-{uuid.uuid4().hex[:12]}", object="chat.completion", created=int(time.time()), model=model,
choices=[SimpleNamespace(index=0, message=message, finish_reason=finish_reason)], usage=usage,
)
return _completion(model, message, finish_reason, _usage_from_metadata(resp.get("usageMetadata") or {}))
class _GeminiStreamChunk(SimpleNamespace):
@@ -579,14 +528,11 @@ def _make_stream_chunk(
tool_calls = None
if tool_call_delta is not None:
tool_delta = SimpleNamespace(
index=tool_call_delta.get("index", 0),
id=tool_call_delta.get("id") or f"call_{uuid.uuid4().hex[:12]}",
type="function",
index=tool_call_delta.get("index", 0), id=_new_call_id(tool_call_delta), type="function",
function=SimpleNamespace(name=tool_call_delta.get("name") or "", arguments=tool_call_delta.get("arguments") or ""),
)
extra_content = tool_call_delta.get("extra_content")
if isinstance(extra_content, dict):
tool_delta.extra_content = extra_content
if isinstance(tool_call_delta.get("extra_content"), dict):
tool_delta.extra_content = tool_call_delta["extra_content"]
tool_calls = [tool_delta]
delta = SimpleNamespace(
role="assistant", content=content or None, tool_calls=tool_calls,
@@ -628,7 +574,6 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
cand = candidates[0] if isinstance(candidates[0], dict) else {}
parts = (cand.get("content") or {}).get("parts") or []
chunks: List[_GeminiStreamChunk] = []
for part_index, part in enumerate(parts):
if not isinstance(part, dict):
continue
@@ -647,15 +592,14 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
)
slot = tool_call_indices.get(call_key)
if slot is None:
slot = {"index": len(tool_call_indices), "id": _new_call_id(fc), "last_arguments": ""}
tool_call_indices[call_key] = slot
slot = tool_call_indices[call_key] = {"index": len(tool_call_indices), "id": _new_call_id(fc), "last_arguments": ""}
# Gemini re-sends the full args each event; emit only the new suffix.
last_arguments = str(slot.get("last_arguments") or "")
emitted_arguments = args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str
emitted = args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str
slot["last_arguments"] = args_str
chunks.append(_make_stream_chunk(model=model, tool_call_delta={
"index": slot["index"], "id": slot["id"], "name": name,
"arguments": emitted_arguments, "extra_content": _tool_call_extra_from_part(part),
"arguments": emitted, "extra_content": _tool_call_extra_from_part(part),
}))
finish_reason_raw = str(cand.get("finishReason") or "")
@@ -663,26 +607,17 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
finish_chunk = _make_stream_chunk(
model=model, finish_reason="tool_calls" if tool_call_indices else _map_gemini_finish_reason(finish_reason_raw)
)
# Carry usageMetadata on the finish chunk so the streaming loop can
# record token counts like the non-streaming path does.
usage_meta = event.get("usageMetadata") or {}
if usage_meta:
# usageMetadata rides on the finish chunk so the streaming loop records
# token counts like the non-streaming path does.
if usage_meta := event.get("usageMetadata") or {}:
finish_chunk.usage = _usage_from_metadata(usage_meta)
chunks.append(finish_chunk)
return chunks
_HTTP_ERROR_CODES = {401: "gemini_unauthorized", 429: "gemini_rate_limited", 404: "gemini_model_not_found"}
def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = None) -> GeminiAPIError:
status = response.status_code
if body_text is None:
try:
body_text = response.text
except Exception:
body_text = ""
body_text = body_text or ""
body_text = (_response_text(response) if body_text is None else body_text) or ""
err_obj: Any = None
if body_text:
try:
@@ -696,8 +631,8 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
err_message = str(err_obj.get("message") or "").strip()
details_list = err_obj.get("details")
reason = ""
metadata: Dict[str, Any] = {}
# First google.rpc.ErrorInfo detail supplies reason/metadata.
reason, metadata = "", {}
for detail in details_list if isinstance(details_list, list) else []:
if isinstance(detail, dict) and not reason and str(detail.get("@type") or "").endswith("/google.rpc.ErrorInfo"):
reason_value, md = detail.get("reason"), detail.get("metadata")
@@ -707,12 +642,10 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
metadata = md
retry_after: Optional[float] = None
header_retry = response.headers.get("Retry-After") or response.headers.get("retry-after")
if header_retry:
try:
retry_after = float(header_retry)
except (TypeError, ValueError):
pass
try:
retry_after = float(response.headers.get("Retry-After") or response.headers.get("retry-after"))
except (TypeError, ValueError):
pass
if err_message:
message = f"Gemini HTTP {status} ({err_status or 'error'}): {err_message}"
@@ -721,27 +654,21 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No
# Users who bypassed the setup wizard (raw GOOGLE_API_KEY in .env) still
# need to learn that the free tier cannot sustain an agent session.
if status == 429 and is_free_tier_quota_error(err_message or body_text):
message = message + _FREE_TIER_GUIDANCE
message += _FREE_TIER_GUIDANCE
# Legacy "Standard" key rejection: Google's raw 401 misleadingly asks for
# OAuth; append the actual fix (mint a new Gemini API key in AI Studio).
if is_standard_key_auth_error(status, err_message or body_text, reason):
message = message + _STANDARD_KEY_GUIDANCE
message += _STANDARD_KEY_GUIDANCE
return GeminiAPIError(
message,
code=_HTTP_ERROR_CODES.get(status, f"gemini_http_{status}"),
status_code=status,
response=response,
message, code=_HTTP_ERROR_CODES.get(status, f"gemini_http_{status}"), status_code=status, response=response,
retry_after=retry_after,
details={"status": err_status, "reason": reason, "metadata": metadata, "message": err_message},
)
class GeminiNativeClient:
"""Minimal OpenAI-SDK-compatible facade over Gemini's native REST API.
``client.chat.completions.create(**kwargs)`` mirrors the OpenAI SDK surface.
"""
"""Minimal OpenAI-SDK-compatible facade (``client.chat.completions.create(**kwargs)``)
over Gemini's native REST API."""
# Declared for agent/auxiliary_client.py: already a complete client, so it
# is never re-dispatched through a wire adapter. (No HERMES_SKIP_ASYNC_WRAP
@@ -804,14 +731,12 @@ class GeminiNativeClient:
max_tokens: Optional[int] = None, top_p: Optional[float] = None, stop: Any = None,
extra_body: Optional[Dict[str, Any]] = None, timeout: Any = None, **_: Any,
) -> Any:
thinking_config = None
if isinstance(extra_body, dict):
thinking_config = extra_body.get("thinking_config") or extra_body.get("thinkingConfig")
extra = extra_body if isinstance(extra_body, dict) else {}
thinking_config = extra.get("thinking_config") or extra.get("thinkingConfig")
request = build_gemini_request(
messages=messages or [], tools=tools, tool_choice=tool_choice, temperature=temperature,
max_tokens=max_tokens, top_p=top_p, stop=stop, thinking_config=thinking_config, model=model,
)
model = bare_gemini_model_id(model)
if stream:
return self._stream_completion(model=model, request=request, timeout=timeout)
@@ -832,19 +757,15 @@ class GeminiNativeClient:
def _stream_completion(self, *, model: str, request: Dict[str, Any], timeout: Any = None) -> Iterator[_GeminiStreamChunk]:
url = f"{self.base_url}/models/{model}:streamGenerateContent?alt=sse"
stream_headers = {**self._headers(), "Accept": "text/event-stream"}
def _generator() -> Iterator[_GeminiStreamChunk]:
try:
with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response:
if response.status_code != 200:
raise gemini_http_error(response, body_text=read_streaming_error_body(response))
tool_call_indices: Dict[str, Dict[str, Any]] = {}
for event in _iter_sse_events(response):
yield from translate_stream_event(event, model, tool_call_indices)
except httpx.HTTPError as exc:
raise GeminiAPIError(f"Gemini streaming request failed: {exc}", code="gemini_stream_error") from exc
return _generator()
try:
with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response:
if response.status_code != 200:
raise gemini_http_error(response, body_text=read_streaming_error_body(response))
tool_call_indices: Dict[str, Dict[str, Any]] = {}
for event in _iter_sse_events(response):
yield from translate_stream_event(event, model, tool_call_indices)
except httpx.HTTPError as exc:
raise GeminiAPIError(f"Gemini streaming request failed: {exc}", code="gemini_stream_error") from exc
class AsyncGeminiNativeClient:
@@ -860,16 +781,15 @@ class AsyncGeminiNativeClient:
self._real_client = sync_client
async def _create_chat_completion(self, **kwargs: Any) -> Any:
stream = bool(kwargs.get("stream"))
result = await asyncio.to_thread(self._sync.chat.completions.create, **kwargs)
if not stream:
if not kwargs.get("stream"):
return result
async def _async_stream() -> Any:
while True:
done, chunk = await asyncio.to_thread(self._sync._advance_stream_iterator, result)
if done:
break
return
yield chunk
return _async_stream()

6
agent/gemini_schema.py
View File

@@ -53,10 +53,9 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
# Gemini requires every ``enum`` entry to be a string even for
# integer/number/boolean types; the declared type stays intact and Gemini
# still emits typed tool arguments at runtime.
# still emits typed tool arguments at runtime. dict.fromkeys = ordered dedupe.
enum_val = cleaned.get("enum")
if isinstance(enum_val, list) and cleaned.get("type") in {"integer", "number", "boolean"}:
# dict.fromkeys = order-preserving dedupe.
stringified = list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None))
if stringified:
cleaned["enum"] = stringified
@@ -71,13 +70,12 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
required_val = cleaned.get("required")
if isinstance(required_val, list):
props_val = cleaned.get("properties")
prop_names = set(props_val.keys()) if isinstance(props_val, dict) else set()
prop_names = set(props_val) if isinstance(props_val, dict) else set()
valid_required = [name for name in required_val if isinstance(name, str) and name in prop_names]
if not valid_required:
cleaned.pop("required", None)
elif len(valid_required) != len(required_val):
cleaned["required"] = valid_required
return cleaned

16
agent/jiter_preload.py
View File

@@ -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

17
agent/lazy_forward.py
View File

@@ -1,28 +1,25 @@
"""Lazy forwarding helpers shared by the AIAgent façade and its mixins.
Targets are resolved on every call, so ``patch("<module>.<name>")`` in tests still intercepts and the
heavy agent modules stay off the ``run_agent`` import path.
Targets are resolved on every call, so ``patch("<module>.<name>")`` 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("<module>.<name>")`` 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

47
agent/lmstudio_reasoning.py
View File

@@ -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