From 7b7d3bc18877f76fa30eab707f5cce65e4028f24 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:39:24 -0700 Subject: [PATCH] refactor(agent/acp,credits,gemini): compact handlers, dispatch session/update sinks, tighten docstrings --- agent/copilot_acp_client.py | 86 +++++++---------- agent/credits_tracker.py | 168 ++++++++++++--------------------- agent/gemini_native_adapter.py | 70 ++++++-------- 3 files changed, 123 insertions(+), 201 deletions(-) diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py index c52131e08a..85059964dc 100644 --- a/agent/copilot_acp_client.py +++ b/agent/copilot_acp_client.py @@ -89,19 +89,16 @@ def _resolve_args() -> list[str]: def _acp_supported(command: str, args: list[str]) -> bool | None: - """Tri-state probe: does ``command`` accept ``--acp``? - - A CLI without the flag (older releases, Claude Code v2.x) exits 1 with - ``error: unknown option '--acp'`` and the parent then waits the full child - timeout for stdout that never arrives. True = help advertises --acp; False = - help ran cleanly without it (caller fast-fails); None = inconclusive (binary - missing / --help failed), caller falls through to the normal spawn error. - Only probes when ``--acp`` is among ``args`` — a custom transport is the operator's business. - """ + """Tri-state probe: does ``command`` accept ``--acp``? A CLI without the flag + (older releases, Claude Code v2.x) exits 1 with ``unknown option '--acp'`` and the + parent would wait the full child timeout for stdout that never arrives. + True = help advertises --acp; False = help ran cleanly without it (caller + fast-fails); None = inconclusive (binary missing / --help failed → normal spawn + error). Only probes when ``--acp`` is among ``args`` — a custom transport is the + operator's business.""" if "--acp" not in args: return True - cached = _ACP_PROBE_CACHE.get(command) - if cached is not None: + if (cached := _ACP_PROBE_CACHE.get(command)) is not None: return cached try: probe = subprocess.run( @@ -120,8 +117,7 @@ def _acp_supported(command: str, args: list[str]) -> bool | None: def _resolve_home_dir() -> str: """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: + if home := os.environ.get("HOME", "").strip(): return home expanded = os.path.expanduser("~") if expanded and expanded != "~": @@ -129,12 +125,9 @@ def _resolve_home_dir() -> str: try: import pwd - resolved = pwd.getpwuid(os.getuid()).pw_dir.strip() # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows) - if resolved: - return resolved + return pwd.getpwuid(os.getuid()).pw_dir.strip() or "/tmp" # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows) except Exception: - pass - return "/tmp" + return "/tmp" def _build_subprocess_env() -> dict[str, str]: @@ -169,13 +162,10 @@ def _enabled_ids(entries: Any, key: str) -> set[str]: 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. - """ + """ACP request selecting ``requested_model`` for ``session``: stable v1 + ``session/set_config_option``, else Copilot's pre-stabilization ``session/set_model`` + when no model config option is advertised. A reported model list is authoritative: + unknown and policy-disabled ids return None instead of being sent.""" session_id = str(session.get("sessionId") or "").strip() requested_model = str(requested_model or "").strip() if not session_id or not requested_model or requested_model == "copilot-acp": @@ -265,8 +255,7 @@ def _effective_timeout(timeout: Any) -> float: def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any: path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd) - block_error = get_read_block_error(str(path)) - if block_error: + if block_error := get_read_block_error(str(path)): raise PermissionError(block_error) try: content = path.read_text(encoding="utf-8") @@ -274,18 +263,14 @@ def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any: content = "" line, limit = params.get("line"), params.get("limit") if isinstance(line, int) and line > 1: - lines = content.splitlines(keepends=True) 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} + content = "".join(content.splitlines(keepends=True)[line - 1:end]) + return {"content": redact_sensitive_text(content, force=True) if content else content} def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any: path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd) - denied = get_write_denied_error(str(path)) - if denied: + if denied := get_write_denied_error(str(path)): raise PermissionError(denied) # Approval-gated paths (e.g. ~/.ssh/config) are only soft-gated for interactive # tools, but the ACP shim has no human channel to confirm — fail closed. @@ -329,8 +314,7 @@ class CopilotACPClient: def close(self) -> None: with self._active_process_lock: - proc = self._active_process - self._active_process = None + proc, self._active_process = self._active_process, None self.is_closed = True if proc is None: return @@ -348,17 +332,15 @@ class CopilotACPClient: tools: list[dict[str, Any]] | None = None, tool_choice: Any = None, stream: bool = False, **_: Any, ) -> Any: prompt_text = _format_messages_as_prompt(messages or [], model=model, tools=tools, tool_choice=tool_choice) - response_text, reasoning_text = self._run_prompt(prompt_text, timeout_seconds=_effective_timeout(timeout), model=model) + response_text, reasoning = self._run_prompt(prompt_text, timeout_seconds=_effective_timeout(timeout), model=model) tool_calls, cleaned_text = _extract_tool_calls_from_text(response_text) - assistant_message = SimpleNamespace( - content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning_text or None, - reasoning_content=reasoning_text or None, reasoning_details=None, + message = SimpleNamespace( + content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning or None, + reasoning_content=reasoning 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) - ), + choices=[SimpleNamespace(message=message, finish_reason="tool_calls" if tool_calls else "stop")], + usage=SimpleNamespace(prompt_tokens=0, completion_tokens=0, total_tokens=0, prompt_tokens_details=SimpleNamespace(cached_tokens=0)), model=model or "copilot-acp", ) return _completion_to_stream_chunks(completion) if stream else completion @@ -424,12 +406,10 @@ class CopilotACPClient: threading.Thread(target=_stdout_reader, daemon=True).start() threading.Thread(target=_stderr_reader, daemon=True).start() - next_id = 0 + request_ids = iter(range(1, 1 << 62)) def _request(method: str, params: dict[str, Any], *, text_parts: list[str] | None = None, reasoning_parts: list[str] | None = None) -> Any: - nonlocal next_id - next_id += 1 - request_id = next_id + request_id = next(request_ids) proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + "\n") proc.stdin.flush() deadline = time.monotonic() + timeout_seconds @@ -494,13 +474,13 @@ class CopilotACPClient: return False if method == "session/update": update = (msg.get("params") or {}).get("update") or {} - kind = str(update.get("sessionUpdate") or "").strip() content = update.get("content") or {} chunk_text = str(content.get("text") or "") if isinstance(content, dict) else "" - if kind == "agent_message_chunk" and chunk_text and text_parts is not None: - text_parts.append(chunk_text) - elif kind == "agent_thought_chunk" and chunk_text and reasoning_parts is not None: - reasoning_parts.append(chunk_text) + 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: + sink.append(chunk_text) return True if process.stdin is None: return True diff --git a/agent/credits_tracker.py b/agent/credits_tracker.py index ba9b571635..6c1880428a 100644 --- a/agent/credits_tracker.py +++ b/agent/credits_tracker.py @@ -1,14 +1,8 @@ -"""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. 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). -""" +"""Nous credits: parse ``x-nous-credits-*`` / ``x-nous-tool-pool-*`` response +headers into a validated CreditsState (depletion = paid_access, subscription-cap +used_fraction, warn-once schema-version gating) and drive the notice policy. +Header contract: see ``_HEADER_FIELDS``. Money is micros ints only; ``*_usd`` +strings are preserved verbatim (never re-parsed to float).""" from __future__ import annotations @@ -23,11 +17,9 @@ from utils import is_truthy_value logger = logging.getLogger(__name__) -# Warn-once latch: emit the version-unsupported warning at most once per process. -_version_warning_emitted: bool = False +_version_warning_emitted: bool = False # warn-once latch (per process) _VALID_DENOMINATOR_KINDS = frozenset({"subscription_cap", "none"}) -# USD format: optional leading minus, one-or-more digits, dot, exactly 2 digits. -_USD_RE = re.compile(r"^-?\d+\.\d{2}$") +_USD_RE = re.compile(r"^-?\d+\.\d{2}$") # optional minus, digits, exactly 2 decimals _SENTINEL = object() # "parse failed" @@ -48,12 +40,12 @@ def _validate_usd(value: Optional[str]) -> bool: @dataclass class CreditsState: - """Full credits state parsed from x-nous-credits-* response headers.""" + """Credits state parsed from x-nous-credits-* response headers.""" version: int = 0 remaining_micros: int = 0 remaining_usd: str = "" - subscription_micros: int = 0 # SIGNED — may be negative (debt). ONLY field allowed negative. + subscription_micros: int = 0 # SIGNED — the ONLY field allowed negative (debt) subscription_usd: str = "" subscription_limit_micros: Optional[int] = None # PAIRED + OPTIONAL (only when subscription_cap) subscription_limit_usd: Optional[str] = None @@ -66,7 +58,7 @@ class CreditsState: paid_access: bool = True # depletion keys off THIS == False, NEVER remaining==0 disabled_reason: Optional[str] = None # header omitted entirely when null as_of_ms: int = 0 - captured_at: float = 0.0 # time.time() when this was captured + captured_at: float = 0.0 # time.time() when captured from_header: bool = False # True only when populated by parse_credits_headers() @property @@ -98,17 +90,14 @@ class CreditsState: # paired *_TTL_MS per notice kind (AgentNotice has the field; not plumbed yet). CREDITS_NOTICE_KIND = "sticky" # v1: credits notices are sticky CREDITS_RESTORED_TTL_MS = 8000 # the only TTL notice in v1 (depletion-recovery confirmation) - # Usage-gauge bands (ascending): (threshold_fraction, level, label_pct). One -# escalating line showing the HIGHEST band reached (50 → 75 → 90); crossing up -# replaces it, recovering steps it down. +# escalating line shows the HIGHEST band reached; climbing replaces it, recovery steps down. CREDITS_USAGE_BANDS: tuple[tuple[float, str, int], ...] = ((0.50, "info", 50), (0.75, "warn", 75), (0.90, "warn", 90)) -CREDITS_USAGE_KEY = "credits.usage" # single key for the escalating usage notice - +CREDITS_USAGE_KEY = "credits.usage" # Min subscription balance counting as "grant not yet spent" for the grant_spent -# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent -# residue where headers report 0 — without the floor such a seed opens the gate -# and the first header re-creates the at-open nag. +# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent residue +# where headers report 0 — without the floor a seed opens the gate and the first +# header re-creates the at-open nag. GRANT_UNSPENT_MIN_MICROS = 10_000 @@ -120,10 +109,9 @@ def new_credits_latch() -> dict: @dataclass class AgentNotice: - """Driver-agnostic out-of-band notice, fired via ``AIAgent.notice_callback`` - (cleared via ``notice_clear_callback``); each driver renders its own way. - ``kind``/``ttl_ms`` stay expressive so a future config can switch v1's - sticky credits notices to TTL without touching the policy.""" + """Driver-agnostic out-of-band notice (``AIAgent.notice_callback`` / ``notice_clear_callback``); + each driver renders its own way. ``kind``/``ttl_ms`` stay expressive so a future + config can switch v1's sticky credits notices to TTL without touching the policy.""" text: str level: str = "info" # info | warn | error | success @@ -138,17 +126,13 @@ def _sticky_notice(text: str, level: str, key: str) -> AgentNotice: def is_free_tier_model(model: str, base_url: str = "") -> bool: - """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). - - Fail-open to False (depleted notice still shows): a wrong warning is - recoverable noise; hiding it on a paid model masks a real block. - """ + """True when *model* is a Nous free-tier model, using ONLY local data: + (1) ``:free`` suffix — canonical Nous free SKU marker; (2) ``stealth/`` prefix — + stealth-preview SKUs are free without the suffix (naming-convention trust: a PAID + ``stealth/`` model would wrongly suppress the banner); (3) a PEEK into + ``hermes_cli.models``' pricing cache (filled by the model picker; a miss never + fetches). Fail-open to False (depleted notice still shows): a wrong warning is + recoverable noise; hiding it on a paid model masks a real block.""" if not model: return False if model.endswith(":free") or model.startswith("stealth/"): @@ -158,8 +142,7 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool: try: from hermes_cli.models import _is_model_free, peek_cached_pricing - # peek_cached_pricing owns the /v1-suffix and auth-state key details. - pricing = peek_cached_pricing(base_url) + pricing = peek_cached_pricing(base_url) # owns the /v1-suffix and auth-state key details return bool(pricing) and _is_model_free(model, pricing) except Exception: return False @@ -168,16 +151,12 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool: def evaluate_credits_notices( state: CreditsState, latch: dict, *, model_is_free: bool = False, ) -> tuple[list[AgentNotice], list[str]]: - """Reconcile credits notices against the latch (see :func:`new_credits_latch`). - Mutates ``latch`` IN PLACE. Pure — no I/O, no agent/run_agent imports. - - ``model_is_free`` (see :func:`is_free_tier_model`) suppresses - ``credits.depleted`` — a depleted account on a free model keeps inferencing, - so the banner is noise. Suppression does NOT emit "restored"; that fires - only on a genuine ``paid_access`` flip back to True. - - Returns ``(to_show, to_clear)``; caller emits to_clear FIRST, then to_show. - """ + """Reconcile credits notices against the latch (see :func:`new_credits_latch`); + mutates ``latch`` IN PLACE. Pure — no I/O, no agent/run_agent imports. + ``model_is_free`` suppresses ``credits.depleted`` (a depleted account on a free + model keeps inferencing) WITHOUT emitting "restored" — that fires only on a + genuine ``paid_access`` flip back to True. + Returns ``(to_show, to_clear)``; caller emits to_clear FIRST, then to_show.""" to_show: list[AgentNotice] = [] to_clear: list[str] = [] uf = state.used_fraction @@ -194,12 +173,11 @@ def evaluate_credits_notices( if uf is not None and uf < 1.0 and state.subscription_micros >= GRANT_UNSPENT_MIN_MICROS: latch["seen_grant_unspent"] = True - # 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. + # Highest band reached (ascending → last match wins); None below all. Top-up + # suppression: with purchased credits the cap gauge is the wrong denominator + # ("90% used" on $50 of top-up is noise; it used to stick PERMANENTLY beside + # grant_spent at >=100%) — grant_spent covers the cap-reached case, and a + # mid-session top-up flips current_band → None so the clear path removes the line. current_band: Optional[tuple[float, str, int]] = None if uf is not None and state.purchased_micros <= 0: for band in CREDITS_USAGE_BANDS: @@ -215,8 +193,8 @@ def evaluate_credits_notices( active.discard(CREDITS_USAGE_KEY) if target_band is not None: # Absolute dollars used (a bare "N%" is only meaningful against a Nous - # cap): cap − remaining in micros, clamped [0, cap]; "$?" if a producer - # set the limit without its *_usd. Re-emits on band change only. + # cap): cap − remaining, clamped [0, cap]; "$?" if a producer set the + # limit without its *_usd. Re-emits on band change only. level = current_band[1] # type: ignore[index] (current_band set when target_band set) lim = state.subscription_limit_micros or 0 used_usd = f"{max(0, min(lim, lim - state.subscription_micros)) / 1_000_000:.2f}" if lim else "?" @@ -251,9 +229,7 @@ def evaluate_credits_notices( elif "credits.depleted" in active and not show_depleted: to_clear.append("credits.depleted") active.discard("credits.depleted") - if not depleted_cond: - # Genuine recovery only — switching to a free model while still - # depleted must NOT claim access was restored. + if not depleted_cond: # genuine recovery only — a free-model switch while depleted is NOT "restored" to_show.append(AgentNotice( text="✓ Credit access restored", level="success", kind="ttl", ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored", @@ -261,9 +237,11 @@ def evaluate_credits_notices( return (to_show, to_clear) -# 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 contract: (field, header, kind[, default]). Each *-micros balance has a *-usd +# twin holding the server's formatted USD string. micros: required int >= 0 ("signed": +# may be negative). usd: required ^-?\d+\.\d{2}$. bool: optional "true"/"false" STRING. +# Not in the table: subscription-limit-* (PAIRED/optional), tool-pool-micros (optional), +# denominator-kind ("subscription_cap" | "none"), disabled-reason (omitted when null). _HEADER_FIELDS: tuple[tuple, ...] = ( ("remaining_micros", "x-nous-credits-remaining-micros", "micros"), ("subscription_micros", "x-nous-credits-subscription-micros", "signed"), @@ -293,17 +271,10 @@ def _parse_field(kind: str, raw: Optional[str], default: Any = None) -> Any: def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Optional[CreditsState]: """Parse x-nous-credits-* (and x-nous-tool-pool-*) headers into a CreditsState. - - Returns None (miss) on ANY of: no ``x-nous-credits-version`` header; version - != 1 (> 1 also warns once); a ``*_micros``/``as_of_ms`` field non-integer or - negative (subscription excepted); a ``*_usd`` not matching - ``^-?\\d+\\.\\d{2}$``; ``denominator_kind`` outside {"subscription_cap", - "none"}; ``paid_access``/``tool_pool_gated_off`` not exactly "true"/"false"; - any unexpected exception. - - Fail-open on the subscription_limit pair: a half-pair (only -micros or only - -usd) is treated as both-absent — the parse STILL SUCCEEDS with both None. - """ + None (miss) on ANY of: no version header; version != 1 (> 1 also warns once); + a required field violating ``_HEADER_FIELDS``; unknown ``denominator_kind``; + any unexpected exception. Fail-open on the subscription_limit pair: a + half-pair (only -micros or only -usd) parses as both-absent (both None).""" global _version_warning_emitted try: # Cheap probe before the lowercase copy (header names are case-insensitive): @@ -327,12 +298,9 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt 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 _parse_field("micros", tp_raw) - if tp_val is _SENTINEL: + fields["tool_pool_micros"] = 0 if tp_raw is None else _parse_field("micros", tp_raw) + if fields["tool_pool_micros"] is _SENTINEL: return None - 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. 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: @@ -344,16 +312,12 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt if denominator_kind not in _VALID_DENOMINATOR_KINDS: return None return CreditsState( - version=version_val, - denominator_kind=denominator_kind, + version=version_val, denominator_kind=denominator_kind, disabled_reason=lowered.get("x-nous-credits-disabled-reason"), # None if absent (omitted when null) - captured_at=time.time(), - from_header=True, - **fields, + captured_at=time.time(), from_header=True, **fields, ) except Exception: - # Fail-open → miss; breadcrumb distinguishes a parser regression from a - # legitimate no-headers response. + # Fail-open → miss; the breadcrumb distinguishes a parser regression from a no-headers response. logger.debug("credits ▸ parse_credits_headers raised (fail-open miss)", exc_info=True) return None @@ -394,10 +358,9 @@ _DEV_FIXTURES: dict[str, dict] = { def dev_fixture_credits_state() -> Optional[CreditsState]: - """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.""" + """Fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None (unknown name / unset). + Prod-leak guard: applies ONLY when HERMES_DEV_CREDITS is also on, so a stray + fixture env var can never surface fabricated balances on a real account.""" if not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")): return None name = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip() @@ -427,9 +390,7 @@ def _credits_state_from_account(info) -> Optional[CreditsState]: 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, "" + return (int(round(dollars * 1_000_000)), f"{dollars:.2f}") if isinstance(dollars, (int, float)) else (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)) @@ -468,12 +429,9 @@ def _hydrate_seed_state(agent, state) -> None: def seed_credits_at_session_start(agent) -> bool: """Hydrate agent._credits_state from the portal account (or a dev fixture) and - fire the notice policy so warnings show at session OPEN. Shared by the - TUI/desktop build ("ready") and first-turn setup (plain-CLI fallback). - Idempotent once a seed or real header populated _credits_state. - - Returns True iff it seeded this call. Never raises — credits must never block startup. - """ + fire the notice policy so warnings show at session OPEN (TUI/desktop "ready" and + plain-CLI first-turn setup). Idempotent once a seed or real header populated + _credits_state. Returns True iff it seeded this call. Never raises.""" try: if getattr(agent, "provider", "") != "nous" or getattr(agent, "_credits_state", None) is not None: return False @@ -481,9 +439,7 @@ def seed_credits_at_session_start(agent) -> bool: fixture = dev_fixture_credits_state() except Exception: fixture = None - if fixture is not None: - # Synchronous: a fixture is instant, and tests rely on the state + - # notice landing before this returns. + if fixture is not None: # synchronous: instant, and tests rely on state + notice landing before return _hydrate_seed_state(agent, fixture) return True diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py index 51db2062a4..02bec9115a 100644 --- a/agent/gemini_native_adapter.py +++ b/agent/gemini_native_adapter.py @@ -272,10 +272,9 @@ def _translate_tool_result_to_gemini( 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. - if is_gemini3: - image_parts = [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p] - if image_parts: - function_response["parts"] = image_parts + image_parts = [p for p in _extract_multimodal_parts(raw_content) if "inlineData" in p] if is_gemini3 else [] + if image_parts: + function_response["parts"] = image_parts return {"functionResponse": function_response} @@ -308,10 +307,9 @@ def _build_gemini_contents( for tool_call in tool_calls if isinstance(tool_calls, list) else []: if not isinstance(tool_call, dict): continue - tool_call_id = _tool_call_id(tool_call) tool_name = str((tool_call.get("function") or {}).get("name") or "") - if tool_call_id and tool_name: - tool_name_by_call_id[tool_call_id] = tool_name + if _tool_call_id(tool_call) and tool_name: + tool_name_by_call_id[_tool_call_id(tool_call)] = 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}) @@ -485,12 +483,10 @@ def _envelope(model: str, object_: str, choice: SimpleNamespace, usage: Any, cls ) -def _tool_call_ns(fc: Dict[str, Any], part: Dict[str, Any], index: int) -> SimpleNamespace: - tool_call = SimpleNamespace( - id=_new_call_id(fc), type="function", index=index, - function=SimpleNamespace(name=str(fc["name"]), arguments=_dump_call_args(fc)), - ) - if extra_content := _tool_call_extra_from_part(part): +def _tool_call_ns(name: str, arguments: str, index: int, call_id: str, extra_content: Any) -> SimpleNamespace: + """OpenAI-shaped tool call; ``extra_content`` attached only when it is a dict.""" + tool_call = SimpleNamespace(id=call_id, type="function", index=index, function=SimpleNamespace(name=name, arguments=arguments)) + if isinstance(extra_content, dict): tool_call.extra_content = extra_content return tool_call @@ -513,7 +509,7 @@ def translate_gemini_response(resp: Dict[str, Any], model: str) -> SimpleNamespa elif isinstance(part.get("text"), str): text_pieces.append(part["text"]) elif isinstance(fc := part.get("functionCall"), dict) and fc.get("name"): - tool_calls.append(_tool_call_ns(fc, part, index)) + tool_calls.append(_tool_call_ns(str(fc["name"]), _dump_call_args(fc), index, _new_call_id(fc), _tool_call_extra_from_part(part))) if cand is None: finish_reason, usage = "stop", _usage_from_metadata({}) @@ -538,13 +534,8 @@ def _make_stream_chunk( ) -> _GeminiStreamChunk: tool_calls = None if tool_call_delta is not None: - tool_delta = SimpleNamespace( - 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 ""), - ) - if isinstance(tool_call_delta.get("extra_content"), dict): - tool_delta.extra_content = tool_call_delta["extra_content"] - tool_calls = [tool_delta] + d = tool_call_delta + tool_calls = [_tool_call_ns(d.get("name") or "", d.get("arguments") or "", d.get("index", 0), _new_call_id(d), d.get("extra_content"))] delta = SimpleNamespace( role="assistant", content=content or None, tool_calls=tool_calls, reasoning=reasoning or None, reasoning_content=reasoning or None, @@ -556,9 +547,7 @@ def _make_stream_chunk( def _iter_sse_events(response: httpx.Response) -> Iterator[Dict[str, Any]]: buffer = "" for chunk in response.iter_text(): - if not chunk: - continue - buffer += chunk + buffer += chunk or "" while "\n" in buffer: line, buffer = buffer.split("\n", 1) line = line.rstrip("\r") @@ -604,11 +593,11 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices: 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 = 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, "extra_content": _tool_call_extra_from_part(part), + "arguments": args_str[len(last_arguments):] if args_str.startswith(last_arguments) else args_str, + "extra_content": _tool_call_extra_from_part(part), })) finish_reason_raw = str(cand.get("finishReason") or "") @@ -628,12 +617,11 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No status = response.status_code body_text = (_response_text(response) if body_text is None else body_text) or "" err_obj: Any = None - if body_text: - try: - parsed = json.loads(body_text) - err_obj = parsed.get("error") if isinstance(parsed, dict) else None - except (ValueError, TypeError): - pass + try: + parsed = json.loads(body_text) if body_text else None + err_obj = parsed.get("error") if isinstance(parsed, dict) else None + except (ValueError, TypeError): + pass if not isinstance(err_obj, dict): err_obj = {} err_status = str(err_obj.get("status") or "").strip() @@ -645,10 +633,8 @@ def gemini_http_error(response: httpx.Response, *, body_text: Optional[str] = No for detail in details_list if isinstance(details_list, list) else []: if isinstance(detail, dict) and not reason and str(detail.get("@type") or "").endswith("/google.rpc.ErrorInfo"): reason_value, md = detail.get("reason"), detail.get("metadata") - if isinstance(reason_value, str): - reason = reason_value - if isinstance(md, dict): - metadata = md + reason = reason_value if isinstance(reason_value, str) else reason + metadata = md if isinstance(md, dict) else metadata retry_after: Optional[float] = None try: @@ -744,8 +730,9 @@ class GeminiNativeClient: if stream: return self._stream_completion(model=model, request=request, timeout=timeout) - url = f"{self.base_url}/models/{model}:generateContent" - response = self._http.post(url, json=request, headers=self._headers(), timeout=timeout) + response = self._http.post( + f"{self.base_url}/models/{model}:generateContent", json=request, headers=self._headers(), timeout=timeout + ) if response.status_code != 200: raise gemini_http_error(response) try: @@ -759,9 +746,9 @@ 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"} + headers = {**self._headers(), "Accept": "text/event-stream"} try: - with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response: + with self._http.stream("POST", url, json=request, headers=headers, timeout=timeout) as response: if response.status_code != 200: raise gemini_http_error(response, body_text=read_streaming_error_body(response)) tool_call_indices: Dict[str, Dict[str, Any]] = {} @@ -776,8 +763,7 @@ class AsyncGeminiNativeClient: def __init__(self, sync_client: GeminiNativeClient): self._sync = sync_client - self.api_key = sync_client.api_key - self.base_url = sync_client.base_url + self.api_key, self.base_url = sync_client.api_key, sync_client.base_url self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion)) # The auxiliary cache evicts entries by leaf client; GeminiNativeClient # is itself the leaf (no OpenAI client beneath it).