From f49413fde645f8d6cb5b900e1dbaa13cd6788655 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:23:20 -0700 Subject: [PATCH] refactor(agent/gemini_schema,acp): compact schema sanitizer and client ctor --- agent/copilot_acp_client.py | 24 +++++++++--------------- agent/gemini_schema.py | 37 ++++++++++++++----------------------- 2 files changed, 23 insertions(+), 38 deletions(-) diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py index 5b2d2e46a6..e15026108f 100644 --- a/agent/copilot_acp_client.py +++ b/agent/copilot_acp_client.py @@ -276,8 +276,7 @@ class CopilotACPClient: acp_command: str | None = None, acp_args: list[str] | None = None, acp_cwd: str | None = None, command: str | None = None, args: list[str] | None = None, **_: Any, ): - self.api_key = api_key or "copilot-acp" - self.base_url = base_url or ACP_MARKER_BASE_URL + self.api_key, self.base_url = api_key or "copilot-acp", base_url or ACP_MARKER_BASE_URL self._default_headers = dict(default_headers or {}) self._acp_command = acp_command or command or _resolve_command() self._acp_args = list(acp_args or args or _resolve_args()) @@ -308,8 +307,8 @@ class CopilotACPClient: 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) message = SimpleNamespace( - content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning or None, - reasoning_content=reasoning or None, reasoning_details=None, + content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning or None, reasoning_content=reasoning or None, + reasoning_details=None, ) completion = SimpleNamespace( choices=[SimpleNamespace(message=message, finish_reason="tool_calls" if tool_calls else "stop")], @@ -330,14 +329,12 @@ class CopilotACPClient: "HERMES_COPILOT_ACP_COMMAND / HERMES_COPILOT_ACP_ARGS to a working pair." ) try: - # Hide the console the child would flash on Windows; stdio pipes stay intact. - from hermes_cli._subprocess_compat import windows_hide_flags + from hermes_cli._subprocess_compat import windows_hide_flags # hide the Windows console flash; pipes intact proc = subprocess.Popen( - [self._acp_command] + self._acp_args, - stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - text=True, encoding='utf-8', errors='replace', bufsize=1, - cwd=self._acp_cwd, env=_build_subprocess_env(), creationflags=windows_hide_flags(), + [self._acp_command] + self._acp_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, encoding='utf-8', errors='replace', bufsize=1, cwd=self._acp_cwd, env=_build_subprocess_env(), + creationflags=windows_hide_flags(), ) except FileNotFoundError as exc: raise RuntimeError( @@ -392,7 +389,6 @@ class CopilotACPClient: err = msg.get("error") or {} raise RuntimeError(f"Copilot ACP {method} failed: {err.get('message') or err}") return msg.get("result") - stderr_text = "\n".join(stderr_tail).strip() if proc.poll() is not None and stderr_text: if _is_gh_copilot_deprecation_message(stderr_text): @@ -418,10 +414,8 @@ class CopilotACPClient: ) text_parts: list[str] = [] reasoning_parts: list[str] = [] - _request( - "session/prompt", {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]}, - text_parts=text_parts, reasoning_parts=reasoning_parts, - ) + prompt = {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]} + _request("session/prompt", prompt, text_parts=text_parts, reasoning_parts=reasoning_parts) return "".join(text_parts), "".join(reasoning_parts) finally: self.close() diff --git a/agent/gemini_schema.py b/agent/gemini_schema.py index 9f9a956f9c..41b4a3e03f 100644 --- a/agent/gemini_schema.py +++ b/agent/gemini_schema.py @@ -5,37 +5,30 @@ from __future__ import annotations import math from typing import Any, Dict -# Gemini's ``FunctionDeclaration.parameters`` accepts the ``Schema`` object, -# only a subset of OpenAPI 3.0 / JSON Schema; everything else is stripped. +# Gemini's ``FunctionDeclaration.parameters`` accepts only a subset of OpenAPI 3.0 / +# JSON Schema (the ``Schema`` object); everything else is stripped. _GEMINI_SCHEMA_ALLOWED_KEYS = { - "type", "format", "title", "description", "nullable", "enum", - "maxItems", "minItems", "properties", "required", "minProperties", - "maxProperties", "minLength", "maxLength", "pattern", "example", - "anyOf", "propertyOrdering", "default", "items", "minimum", "maximum", + "type", "format", "title", "description", "nullable", "enum", "maxItems", "minItems", "properties", "required", + "minProperties", "maxProperties", "minLength", "maxLength", "pattern", "example", "anyOf", "propertyOrdering", + "default", "items", "minimum", "maximum", } def _stringify_enum_value(item: Any) -> Any: """Gemini-safe string for a scalar enum entry, or None to drop it.""" - if isinstance(item, str): - return item if isinstance(item, bool): return "true" if item else "false" if isinstance(item, (int, float)) and math.isfinite(item): return str(item) - return None + return item if isinstance(item, str) else None def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]: - """Return a Gemini-compatible copy of a tool parameter schema. - - Keeps only the documented Gemini subset (drops e.g. ``$schema`` / - ``additionalProperties``) and recursively sanitizes nested - ``properties`` / ``items`` / ``anyOf`` definitions. - """ + """Gemini-compatible copy of a tool parameter schema: keeps only the documented subset + (drops e.g. ``$schema`` / ``additionalProperties``) and recursively sanitizes nested + ``properties`` / ``items`` / ``anyOf``.""" if not isinstance(schema, dict): return {} - cleaned: Dict[str, Any] = {} for key, value in schema.items(): if key not in _GEMINI_SCHEMA_ALLOWED_KEYS: @@ -56,17 +49,15 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]: # 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"}: - stringified = list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None)) - if stringified: + if stringified := list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None)): cleaned["enum"] = stringified else: cleaned.pop("enum", None) - # Gemini validates ``required`` strictly against the same node's - # ``properties`` (HTTP 400 "property is not defined"), and one bad tool - # schema fails the ENTIRE request. MCP servers routinely emit ``required`` - # without ``properties``, so keep only names that exist here; the tool - # handler still validates required fields at execution time. + # Gemini validates ``required`` strictly against the same node's ``properties`` (HTTP 400 + # "property is not defined") and one bad tool schema fails the ENTIRE request. MCP servers + # routinely emit ``required`` without ``properties``, so keep only names that exist here; + # the tool handler still validates required fields at execution time. required_val = cleaned.get("required") if isinstance(required_val, list): props_val = cleaned.get("properties")