fix: preserve native Gemini union constraints

Complete the type-array normalization salvaged from #55643: stringify mixed
union enum metadata, preserve existing anyOf constraints, and keep array
items and object properties/required on the corresponding typed branches.

Exercise real native request serialization over loopback and Google SDK
validation with a scalar control; no live Google credentials were available.
This commit is contained in:
Teknium
2026-09-07 21:04:29 -07:00
parent 6a04ea67c0
commit bbcf1ee180
4 changed files with 217 additions and 20 deletions

View File

@@ -16,6 +16,12 @@ _GEMINI_SCHEMA_ALLOWED_KEYS = {
}
_GEMINI_STRUCTURAL_KEYS = {
"array": {"items", "minItems", "maxItems"},
"object": {"properties", "required", "minProperties", "maxProperties", "propertyOrdering"},
}
def _stringify_enum_value(item: Any) -> Any:
"""Gemini-safe string for a scalar enum entry, or None to drop it."""
if isinstance(item, bool):
@@ -25,6 +31,30 @@ def _stringify_enum_value(item: Any) -> Any:
return item if isinstance(item, str) else None
def _normalize_gemini_type_array(type_array: list, cleaned: Dict[str, Any]) -> None:
"""Keep union alternatives and their branch-local structural constraints."""
derived: Dict[str, Any] = {}
_normalize_type_array(type_array, derived)
if "anyOf" in derived:
constraints = {"anyOf": cleaned["anyOf"]} if "anyOf" in cleaned else {}
# Gemini requires items/properties on the typed branch itself, not
# its typeless parent. Keep required paired with those properties.
structural = {key: cleaned.pop(key) for keys in _GEMINI_STRUCTURAL_KEYS.values()
for key in keys if key in cleaned}
cleaned["anyOf"] = [
sanitize_gemini_schema({**branch, **constraints, **{
key: value for key, value in structural.items()
if key in _GEMINI_STRUCTURAL_KEYS.get(branch["type"], ())
}}) for branch in derived["anyOf"]
]
else:
cleaned["type"] = derived["type"]
if derived.get("nullable"):
# Derived from "null" in the array. Set AFTER the loop so it beats an input
# ``nullable: false`` regardless of which key the producer emitted first.
cleaned["nullable"] = True
def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
"""Gemini-compatible copy of a tool parameter schema: keeps only the documented subset
(drops e.g. ``$schema`` / ``additionalProperties``) and recursively sanitizes nested
@@ -32,7 +62,6 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
if not isinstance(schema, dict):
return {}
cleaned: Dict[str, Any] = {}
type_array: Any = None
for key, value in schema.items():
if key not in _GEMINI_SCHEMA_ALLOWED_KEYS:
continue
@@ -44,33 +73,21 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
elif key == "anyOf":
if isinstance(value, list):
cleaned[key] = [sanitize_gemini_schema(item) for item in value if isinstance(item, dict)]
elif key == "type" and isinstance(value, list):
type_array = value # normalized after the loop, so the derived ``nullable`` wins
else:
cleaned[key] = value
# JSON Schema allows an array ``type`` (``["string", "null"]``, ``["string", "integer"]``).
# Gemini's ``Schema`` accepts only a single string, and the enum check below would evaluate
# ``[...] in {...}`` -> TypeError: unhashable type: 'list', aborting translation of the WHOLE
# tool catalog. Reuse the sanitizer's normalization so a multi-type array becomes an ``anyOf``
# of single-type branches (no branch dropped) rather than keeping only the first.
if type_array is not None:
derived: Dict[str, Any] = {}
_normalize_type_array(type_array, derived)
if "anyOf" in derived:
cleaned["anyOf"] = derived["anyOf"]
else:
cleaned["type"] = derived["type"]
if derived.get("nullable"):
# Derived from "null" in the array. Set AFTER the loop so it beats an input
# ``nullable: false`` regardless of which key the producer emitted first.
cleaned["nullable"] = True
type_array = cleaned.get("type")
if isinstance(type_array, list):
cleaned.pop("type")
_normalize_gemini_type_array(type_array, cleaned)
# 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. dict.fromkeys = ordered dedupe.
enum_val = cleaned.get("enum")
if isinstance(enum_val, list) and cleaned.get("type") in {"integer", "number", "boolean"}:
if isinstance(enum_val, list) and (
isinstance(type_array, list) or cleaned.get("type") in {"integer", "number", "boolean"}
):
if stringified := list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None)):
cleaned["enum"] = stringified
else:

View File

@@ -0,0 +1,120 @@
"""Offline A/B native adapter + Google SDK schema validation (no API calls).
Run with httpx and google-genai==1.47.0 installed:
python evals/gemini_type_array_probe.py --base origin/main
"""
from __future__ import annotations
import argparse
import copy
import importlib
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import types
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class WireCapture(BaseHTTPRequestHandler):
requests = []
def do_POST(self):
size = int(self.headers["Content-Length"])
self.requests.append(json.loads(self.rfile.read(size)))
# Intentionally stop at the wire boundary; never fake a model response.
self.send_response(418)
self.end_headers()
self.wfile.write(b"Local schema capture only")
def log_message(self, *_):
pass
def wire_translate(adapter, tools):
server = ThreadingHTTPServer(("127.0.0.1", 0), WireCapture)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with adapter.GeminiNativeClient(
api_key="local-probe-not-a-secret",
base_url=f"http://127.0.0.1:{server.server_port}/v1beta",
) as client:
try:
client.chat.completions.create(
model="gemini-2.5-flash", tools=tools,
messages=[{"role": "user", "content": "Schema validation probe"}],
)
except adapter.GeminiAPIError as exc:
assert exc.status_code == 418
return WireCapture.requests.pop()["tools"]
finally:
server.shutdown()
server.server_close()
thread.join()
def probe(label, translate):
from google.genai.types import FunctionDeclaration
cases = {
"nullable": {"type": ["number", "null"]},
"nullable_enum": {"type": ["integer", "null"], "enum": [1, 2]},
"union_enum": {"type": ["string", "integer"], "enum": ["one", 2]},
"nested_items": {"type": "array", "items": {"type": ["string", "null"]}},
"nested_anyof": {"anyOf": [{"type": ["string", "integer"]}, {"type": "boolean"}]},
"constrained_union": {"type": ["string", "integer"], "anyOf": [{"enum": ["one"]}, {"enum": ["two"]}]},
"structural_union": {"type": ["array", "object", "string"], "items": {"type": "integer"}, "properties": {"name": {"type": "string"}}, "required": ["name"]},
"scalar_control": {"type": "number"},
}
results = {}
for name, node in cases.items():
parameters = {"type": "object", "properties": {"value": node}}
original = copy.deepcopy(parameters)
try:
wire = translate([{"type": "function", "function": {"name": "probe", "parameters": parameters}}])
declaration = wire[0]["functionDeclarations"][0]
# Validate the actual adapter declaration, not a reconstructed schema.
FunctionDeclaration.model_validate(json.loads(json.dumps(declaration)))
results[name] = "accepted"
except (TypeError, ValueError) as exc:
results[name] = f"{type(exc).__name__}: {str(exc).splitlines()[0]}"
assert parameters == original, name
print(json.dumps({"label": label, "results": results}, indent=2))
return results
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", default="origin/main")
args = parser.parse_args()
root = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(root))
for name in list(sys.modules):
if name == "agent" or name.startswith(("agent.", "tools", "hermes")):
del sys.modules[name]
with tempfile.TemporaryDirectory(prefix="gemini-schema-home-") as home:
os.environ["HERMES_HOME"] = home
adapter = importlib.import_module("agent.gemini_native_adapter")
baseline = types.ModuleType("baseline_gemini_schema")
source = subprocess.check_output(
["git", "show", f"{args.base}:agent/gemini_schema.py"], cwd=root, text=True,
)
exec(compile(source, f"{args.base}:agent/gemini_schema.py", "exec"), baseline.__dict__)
fixed = adapter.sanitize_gemini_tool_parameters
adapter.sanitize_gemini_tool_parameters = baseline.sanitize_gemini_tool_parameters
before = probe(args.base, lambda tools: wire_translate(adapter, tools))
adapter.sanitize_gemini_tool_parameters = fixed
after = probe("worktree", lambda tools: wire_translate(adapter, tools))
assert before["scalar_control"] == "accepted"
assert all(result != "accepted" for name, result in before.items() if name != "scalar_control")
assert all(result == "accepted" for result in after.values())
print(f"PASS: {len(before) - 1} failing base cases repaired; scalar control and inputs preserved.")
print("LOCAL SDK VALIDATION ONLY: no Google API call or Windows Desktop exercise.")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,55 @@
"""Native tool translation must preserve union alternatives at every depth."""
import copy
import pytest
from agent.gemini_native_adapter import _translate_tools_to_gemini
@pytest.mark.parametrize("types", [["string", "null"], ["integer", "null"], ["string", "integer"], ["string", "integer", "boolean", "null"], ["array", "object", "string"], ["null"], [], "string"])
@pytest.mark.parametrize("location", ["properties", "items", "anyOf"])
@pytest.mark.parametrize("nullable_first", [False, True])
@pytest.mark.parametrize("constrained", [False, True])
def test_native_type_arrays_preserve_alternatives(types, location, nullable_first, constrained):
node = {"nullable": False, "type": types} if nullable_first else {"type": types, "nullable": False}
node["description"] = "Keep this guidance"
if "array" in types:
node.update(items={"type": "integer"}, minItems=1)
if "object" in types:
node.update(properties={"name": {"type": "string"}}, required=["name", "undefined"])
if constrained:
node["anyOf"] = [{"enum": ["one"]}, {"enum": ["two"]}]
wrapper = {"properties": {"value": node}} if location == "properties" else {location: node if location == "items" else [node]}
params = {"type": "object", "properties": {"nested": wrapper}}
original = copy.deepcopy(params)
tools = [{"type": "function", "function": {"name": "probe", "parameters": params}}]
out = _translate_tools_to_gemini(tools)[0]["functionDeclarations"][0]["parameters"]["properties"]["nested"][location]
out = out["value"] if location == "properties" else out[0] if location == "anyOf" else out
expected = {t for t in types if t != "null"} if isinstance(types, list) else {types}
branches = out["anyOf"] if "type" not in out else [out]
actual = {branch["type"] for branch in branches}
for branch in branches:
if branch["type"] == "array":
assert branch["items"] == node["items"]
assert branch["minItems"] == node["minItems"]
if branch["type"] == "object" and "object" in types:
assert branch["properties"] == node["properties"]
assert branch["required"] == ["name"]
if "array" in types and branch["type"] != "array":
assert "items" not in branch
if constrained:
assert all(branch["anyOf"] == node["anyOf"] for branch in branches)
assert actual == (expected or {"null" if "null" in types else "object"})
assert not isinstance(out.get("type"), list)
if isinstance(types, list) and "null" in types and expected:
assert out["nullable"] is True
assert out["description"] == node["description"]
assert params == original
@pytest.mark.parametrize("types,enum", [(["integer", "null"], [1, 2]), (["string", "integer"], ["one", 2]), (["boolean", "string"], [True, "other"])])
def test_native_type_array_enums_remain_wire_strings(types, enum):
params = {"type": "object", "properties": {"value": {"type": types, "enum": enum}}}
tools = [{"type": "function", "function": {"name": "probe", "parameters": params}}]
node = _translate_tools_to_gemini(tools)[0]["functionDeclarations"][0]["parameters"]["properties"]["value"]
assert node["enum"] == [str(v).lower() if isinstance(v, bool) else str(v) for v in enum]

View File

@@ -76,6 +76,11 @@ Hermes detects this endpoint and creates its native Gemini adapter. Internally,
- tool results → Gemini `functionResponse` parts
- streaming responses → OpenAI-shaped stream chunks for the Hermes loop
Tool parameter type arrays such as `"type": ["number", "null"]` are translated
into Gemini's scalar type plus `nullable` form. Multi-type unions keep every
alternative through `anyOf`, including nested properties and array items. This
happens automatically; no MCP server or provider configuration change is needed.
:::note Gemini 3 thought signatures
For Gemini 3 tool use, Hermes preserves the `thoughtSignature` values attached to function-call parts and replays them on the next tool turn. That covers the validation-critical path for multi-step agent workflows.