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

refactor(agent/gemini,acp,credits): fold literals and single-use temporaries

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:46:27 -07:00
parent 1b61d2e9c5
commit b84223fd06
3 changed files with 18 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
agent/copilot_acp_client.py
View File

@@ -198,10 +198,9 @@ def _render_message_content(content: Any) -> str:
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, root = candidate.resolve(), Path(cwd).resolve()
resolved, root = Path(path_text).resolve(), Path(cwd).resolve()
try:
resolved.relative_to(root)
except ValueError as exc:
@@ -308,9 +307,9 @@ class CopilotACPClient:
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}': `{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 "
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:
@@ -417,15 +416,12 @@ class CopilotACPClient:
update = (msg.get("params") or {}).get("update") or {}
content = update.get("content") or {}
chunk_text = str(content.get("text") or "") if isinstance(content, dict) else ""
sink = {"agent_message_chunk": text_parts, "agent_thought_chunk": reasoning_parts}.get(
str(update.get("sessionUpdate") or "").strip()
)
if chunk_text and sink is not None:
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")
if method == "session/request_permission":
response = _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})

11
agent/credits_tracker.py
View File

@@ -367,17 +367,16 @@ def _credits_state_from_account(info) -> Optional[CreditsState]:
def _money(dollars) -> tuple[int, str]: # (micros, display usd); (0, "") when absent
return (int(round(dollars * 1_000_000)), f"{dollars:.2f}") if isinstance(dollars, (int, float)) else (0, "")
fields: dict[str, Any] = {}
for prefix, obj, attr in (("remaining", acc, "total_usable_credits"), ("subscription", acc, "subscription_credits_remaining"),
("purchased", acc, "purchased_credits_remaining")):
fields[f"{prefix}_micros"], fields[f"{prefix}_usd"] = _money(getattr(obj, attr, None))
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(
**fields, subscription_limit_micros=cap[0], subscription_limit_usd=cap[1],
rollover_micros=_money(getattr(sub, "rollover_credits", None))[0],
**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",
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)

12
agent/gemini_native_adapter.py
View File

@@ -370,9 +370,9 @@ def _thinking_requests_output_headroom(thinking_config: Any) -> bool:
def _effective_gemini_max_output_tokens(max_tokens: Optional[int], thinking_config: Any) -> int:
"""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."""
"""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):
@@ -622,9 +622,9 @@ class GeminiNativeClient:
) -> 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."
"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
self.base_url = (base_url or DEFAULT_GEMINI_BASE_URL).rstrip("/").removesuffix("/openai")