[Live API] Function declaration size degrades speech output (name pronunciation): is there a documented budget or design pattern?

"""Minimal reproduction: gemini-3.8-live renders a short Spanish first name with an English onset in
a scripted greeting when ~1.7 KB of unused function declarations are added to a ~1,750-character
system instruction. The same instruction without the declarations says it correctly.

WHAT IT DOES
  1. Opens a Live session (AUDIO out, output transcription on) with a short fictitious system
     instruction: Hannah, a phone assistant for a real-estate agent named NAME.
  2. Condition (flags combine; none = control):
       --tool-bytes N       neutral filler function declarations (one STRING parameter each,
                            ~157 bytes each) until their JSON reaches N bytes; never called
       --prompt-chars N     neutral padding appended to the system instruction up to N chars
       --name-mentions K    K neutral sentences naming the agent, spread through that padding
  3. Triggers the greeting with one client_content message (role="user", turn_complete=True)
     asking the model to say a fixed greeting word for word. The greeting says NAME three times.
  4. Saves each session's model audio (24 kHz WAV) and output_transcription to --out-dir.
  5. Optional --grade: sends each WAV 3 times to a Gemini text+audio model that lists every first
     name it hears, written as it SOUNDS (it is never told the spelling). A read is GOOD only if
     every listed name matches --good-re; any other name is BAD; no name is ABSENT. Majority of 3.

EXPECTED  NAME is pronounced the same way in every condition (as it is in the control).
ACTUAL    measured with THIS script (2026-10-12, one Spanish first name, sound-graded, majority of 3;
          output_transcription showed the correct spelling throughout). Sessions with a bad rendering:
            --prompt-chars 1756                    (no tools)    API 3.8 0/20   Vertex 3.8 0/20   API 3.1 0/10
            --prompt-chars 1756 --tool-bytes 1700  (1729 B)      API 3.8 10/20  Vertex 3.8 5/20   API 3.1 0/10
          The same declarations with the bare 111-char instruction gave only 2/20 (API) and 1/20
          (Vertex): the effect needs both a longer system instruction and the declarations.
          A ~15,000-character instruction alone gave 5/20 on API 3.8.
          A bad rendering keeps the name's letters but swaps the Spanish onset for an English one
          (in the grader's syllables, a J-, D- or Y- onset, or the nearest English name).

USAGE
  pip install "google-genai>=2.24.0"
  gcloud auth application-default login                       # for --surface vertex
  set GEMINI_API_KEY=...                                       # for --surface api and for --grade
  python case2_name_rendering_repro.py --surface api --name NAME --prompt-chars 1756 --runs 20 --out-dir out/a_p1          # control
  python case2_name_rendering_repro.py --surface api --name NAME --prompt-chars 1756 --tool-bytes 1700 --runs 20 --out-dir out/a_p2
  python case2_name_rendering_repro.py --surface vertex --project YOUR_PROJECT --name NAME --prompt-chars 1756 --tool-bytes 1700 --runs 20 --out-dir out/v_p2
  add:  --grade --good-re "<regex>"     to sound-grade; <regex> matches the CORRECT sounded-out form
        of NAME as the grader writes it (simple English syllables, stressed syllable in capitals)

  --name is required: the name under test is supplied at run time.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import time
import wave
from collections import Counter
from pathlib import Path

from google import genai
from google.genai import types

SYSTEM = ("You are Hannah, a friendly phone assistant for {name}, a real-estate agent. "
          "Keep replies to one or two sentences.")
GREETING = ("Hi, this is Hannah, {name}'s assistant. {name} can't come to the phone right now, but I can "
            "take a message. Would you like {name} to call you back?")
CUE = ("The caller just connected and has not spoken yet. Say this greeting now, exactly, word for word, "
       "then stop and listen: \"{greeting}\"")
PADDING_SENTENCES = [
    "Background notes: the office keeps ordinary business hours, there is visitor parking by the lobby, "
    "the team shares one calendar, and paperwork is filed electronically.",
    "These notes change nothing about how you speak with callers.",
]
# --name-mentions: neutral sentences that mention the agent's name, used in turn.
NAME_LINES = [
    "{name} works with buyers and sellers across the area.",
    "Messages for {name} are reviewed at the end of each day.",
    "{name} returns calls as soon as possible.",
    "Callers often ask for {name} directly.",
    "{name} keeps the shared calendar up to date.",
]
PHON_Q = ("List every person's first name spoken in this audio, except 'Hannah', in the order spoken. "
          "Write each one only as it SOUNDS, in simple English syllables with the stressed syllable in "
          "capitals (for example: ka-REE-na). One per line, nothing else. If there are none, write NONE.")


def filler_tools(target_bytes: int) -> tuple[list[types.Tool], int, int]:
    decls, total = [], 0
    while total < target_bytes:
        d = types.FunctionDeclaration(
            name=f"record_caller_detail_{len(decls) + 1}", description="Record one caller detail.",
            parameters=types.Schema(type="OBJECT", properties={"value": types.Schema(type="STRING")}))
        decls.append(d)
        total += len(json.dumps(d.model_dump(exclude_none=True), ensure_ascii=False))
    return ([types.Tool(function_declarations=decls)] if decls else []), total, len(decls)


def system_instruction(name: str, prompt_chars: int, mentions: int = 0) -> str:
    """SYSTEM, then neutral padding up to prompt_chars. With mentions=K, K neutral sentences that
    name the agent are spread evenly through the padding (they are never truncated)."""
    base = SYSTEM.format(name=name)
    if not prompt_chars and not mentions:
        return base
    named = [NAME_LINES[i % len(NAME_LINES)].format(name=name) for i in range(mentions)]
    room = max(0, prompt_chars - len(base) - 2)
    neutral, size, i = [], sum(len(s) + 1 for s in named), 0
    while size < room:
        s = PADDING_SENTENCES[i % len(PADDING_SENTENCES)]
        neutral.append(s)
        size += len(s) + 1
        i += 1
    pieces = list(neutral)
    step = len(neutral) / (len(named) + 1) if named else 0
    for k, s in reversed(list(enumerate(named, 1))):  # insert from the back so earlier slots stay put
        pieces.insert(round(k * step), s)
    pad = " ".join(pieces)
    if neutral:  # the last piece is always neutral, so this cut never touches a named sentence
        pad = pad[:room]
    return base + "\n\n" + pad


def config(system: str, tools: list[types.Tool]) -> types.LiveConnectConfig:
    return types.LiveConnectConfig(
        response_modalities=["AUDIO"],
        system_instruction=system,
        tools=tools or None,
        temperature=0.7,
        speech_config=types.SpeechConfig(voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Leda"))),
        input_audio_transcription=types.AudioTranscriptionConfig(),
        output_audio_transcription=types.AudioTranscriptionConfig(),
    )


async def run_session(client: genai.Client, model: str, cfg: types.LiveConnectConfig, cue: str,
                      wav_path: Path) -> dict:
    audio, said, session_id = bytearray(), [], None
    started = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    async with client.aio.live.connect(model=model, config=cfg) as session:
        done = asyncio.Event()

        async def receive() -> None:
            nonlocal session_id
            while not done.is_set():
                async for msg in session.receive():
                    if msg.setup_complete and msg.setup_complete.session_id:
                        session_id = msg.setup_complete.session_id
                    sc = msg.server_content
                    if not sc:
                        continue
                    if sc.model_turn:
                        for p in sc.model_turn.parts or []:
                            if p.inline_data and p.inline_data.data:
                                audio.extend(p.inline_data.data)
                    if sc.output_transcription and sc.output_transcription.text:
                        said.append(sc.output_transcription.text)
                    if sc.turn_complete:
                        done.set()
                        break

        receiver = asyncio.create_task(receive())
        await session.send_client_content(turns=[types.Content(role="user", parts=[types.Part(text=cue)])],
                                          turn_complete=True)
        timed_out = False
        try:
            await asyncio.wait_for(done.wait(), timeout=20)
        except asyncio.TimeoutError:
            timed_out = True
        receiver.cancel()
    with wave.open(str(wav_path), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(24000)
        w.writeframes(bytes(audio))
    return {"started_utc": started, "session_id": session_id, "wav": wav_path.name,
            "audio_s": round(len(audio) / 48000, 1), "output_transcription": "".join(said).strip(),
            "timed_out": timed_out}


def classify(read: list[str], good_re: str, ignore_re: str) -> str:
    names = [n for n in read if n.upper() != "NONE" and not (ignore_re and re.search(ignore_re, n, re.I))]
    if not names:
        return "ABSENT"
    return "GOOD" if all(re.search(good_re, n, re.I) for n in names) else "BAD"


async def grade(gclient: genai.Client, gmodel: str, wav_path: Path, good_re: str, ignore_re: str) -> dict:
    data, reads = wav_path.read_bytes(), []
    for _ in range(3):
        for attempt in range(4):
            try:
                r = await gclient.aio.models.generate_content(
                    model=gmodel, contents=[types.Part.from_bytes(data=data, mime_type="audio/wav"), PHON_Q])
                break
            except Exception as e:  # noqa: BLE001  (retry only "model overloaded")
                if "503" not in str(e) or attempt == 3:
                    raise
                await asyncio.sleep(20 * (attempt + 1))
        reads.append([ln.strip() for ln in (r.text or "").splitlines() if ln.strip()])
    labels = [classify(rd, good_re, ignore_re) for rd in reads]
    top, n = Counter(labels).most_common(1)[0]
    return {"reads": [" / ".join(rd) for rd in reads], "labels": labels,
            "majority": top if n >= 2 else "NO_MAJORITY", "split": len(set(labels)) > 1}


async def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--surface", choices=["vertex", "api"], required=True)
    ap.add_argument("--model", default="gemini-3.8-live")
    ap.add_argument("--project", default=os.getenv("GOOGLE_CLOUD_PROJECT"))
    ap.add_argument("--location", default="us")
    ap.add_argument("--name", required=True, help="the first name under test (run-time only)")
    ap.add_argument("--tool-bytes", type=int, default=0)
    ap.add_argument("--prompt-chars", type=int, default=0)
    ap.add_argument("--name-mentions", type=int, default=0,
                    help="neutral sentences naming the agent, spread through the padding")
    ap.add_argument("--runs", type=int, default=20)
    ap.add_argument("--concurrency", type=int, default=4)
    ap.add_argument("--out-dir", type=Path, required=True)
    ap.add_argument("--grade", action="store_true", help="sound-grade each WAV (needs GEMINI_API_KEY)")
    ap.add_argument("--grade-model", default="gemini-3.8-flash")
    ap.add_argument("--good-re", default=None, help="--grade: regex for the CORRECT sounded-out form")
    ap.add_argument("--ignore-re", default="", help="--grade: listed tokens that are not the name")
    a = ap.parse_args()
    if a.grade and not a.good_re:
        ap.error("--grade needs --good-re")

    if a.surface == "vertex":
        if not a.project:
            ap.error("--project (or GOOGLE_CLOUD_PROJECT) is required for --surface vertex")
        client = genai.Client(vertexai=True, project=a.project, location=a.location)
    else:
        client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    gclient = genai.Client(api_key=os.environ["GEMINI_API_KEY"]) if a.grade else None

    system = system_instruction(a.name, a.prompt_chars, a.name_mentions)
    tools, tool_bytes, tool_count = filler_tools(a.tool_bytes) if a.tool_bytes else ([], 0, 0)
    cue = CUE.format(greeting=GREETING.format(name=a.name))
    mentions = len(re.findall(rf"\b{re.escape(a.name)}\b", system))
    parts = ([f"prompt-chars {len(system)}"] if a.prompt_chars else []) + \
            ([f"name x{mentions} in system"] if a.name_mentions else []) + \
            ([f"tool-bytes {tool_bytes} ({tool_count} declarations)"] if a.tool_bytes else [])
    cond = " + ".join(parts) or "control"
    label = f"{a.surface}:{a.model} {cond}"
    a.out_dir.mkdir(parents=True, exist_ok=True)
    sem = asyncio.Semaphore(a.concurrency)

    async def one(n: int) -> dict:
        async with sem:
            try:
                r = await run_session(client, a.model, config(system, tools), cue, a.out_dir / f"session_{n:02d}.wav")
            except Exception as e:  # noqa: BLE001
                return {"run": n, "error": f"{type(e).__name__}: {str(e)[:300]}"}
            if a.grade and r["audio_s"] > 0.5:
                try:
                    r["grade"] = await grade(gclient, a.grade_model, a.out_dir / r["wav"], a.good_re, a.ignore_re)
                except Exception as e:  # noqa: BLE001
                    r["grade_error"] = f"{type(e).__name__}: {str(e)[:200]}"
            r["run"] = n
            return r

    results = await asyncio.gather(*(one(n) for n in range(1, a.runs + 1)))
    print(f"google-genai {genai.__version__}  {label}  system_instruction={len(system)} chars")
    for r in results:
        if "error" in r:
            print(f"  session {r['run']:2d}: ERROR {r['error']}")
            continue
        g = r.get("grade", {})
        print(f"  session {r['run']:2d}: {g.get('majority', '-'):11s} {g.get('reads', '')}  "
              f"audio={r['audio_s']}s{' TIMEOUT' if r['timed_out'] else ''}  transcript: {r['output_transcription'][:90]}")
    tally = Counter(r.get("grade", {}).get("majority", "UNGRADED") for r in results if "error" not in r)
    errors = sum("error" in r for r in results)
    splits = sum(r.get("grade", {}).get("split", False) for r in results)
    print(f"\nSUMMARY {label}: BAD {tally['BAD']} / GOOD {tally['GOOD']} / ABSENT {tally['ABSENT']} / "
          f"NO_MAJORITY {tally['NO_MAJORITY']} / UNGRADED {tally['UNGRADED']}  (split votes {splits}, errors {errors})")
    (a.out_dir / "results.json").write_text(json.dumps({
        "surface": a.surface, "model": a.model, "condition": cond, "system_instruction_chars": len(system),
        "name_mentions_in_system": mentions,
        "tool_bytes": tool_bytes, "tool_count": tool_count, "google_genai": genai.__version__,
        "grade_model": a.grade_model if a.grade else None, "good_re": a.good_re, "ignore_re": a.ignore_re,
        "tally": dict(tally), "split_votes": splits, "errors": errors, "sessions": results},
        indent=2, ensure_ascii=False), encoding="utf-8")


if __name__ == "__main__":
    asyncio.run(main())