[Live API] Vertex gemini-3.8-live: input_transcription returns model text instead of caller audio after mid-session send_client_content

"""Minimal reproduction: Live API input_transcription contains model-style text after
mid-session send_client_content (gemini-3.8-live on Vertex AI; not on the Gemini Developer API).

WHAT IT DOES
  1. Opens a Live session (AUDIO out, input + output transcription on, automatic activity
     detection with a 200 ms end-of-speech silence).
  2. Sends background instructions mid-session as several client_content messages
     (role="user", turn_complete=False), the documented way to add context to a running session.
  3. Streams six short caller utterances (16 kHz PCM WAV, attached) in real time, each
     followed by silence while the model answers.
  4. Prints, per caller turn, what input_transcription returned next to what the caller
     actually said, and flags words the caller never said.

EXPECTED  input_transcription matches the caller audio.
ACTUAL    (Vertex AI, gemini-3.8-live) on some turns input_transcription contains sentences
          the caller never said: an example line from the injected instructions, or new
          assistant-style text. The model's spoken answer to the same turn is correct.

USAGE
  pip install "google-genai>=2.24.0"
  gcloud auth application-default login                      # for --surface vertex
  set GEMINI_API_KEY=...                                      # for --surface api
  python case1_input_transcription_repro.py --surface vertex --project YOUR_PROJECT --runs 3
  python case1_input_transcription_repro.py --surface api --runs 3
  python case1_input_transcription_repro.py --surface vertex --no-inject --project YOUR_PROJECT   # control

  --role user|model|system|system-joined   role of the injected messages (default user);
                                           system-joined sends all of them as ONE role=system message
  --silence MS                             silence_duration_ms (default 200)
  --audio-dir defaults to ./case1_audio (c1.wav ... c6.wav).
"""
from __future__ import annotations

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

from google import genai
from google.genai import types

SYSTEM = ("You are Hannah, a friendly phone assistant for Javier, a real-estate agent. "
          "Keep replies to one or two sentences.")

# Fictitious background instructions, sent mid-session. Each string is one client_content message.
_HEADER = ("[instructions part {n}] Background instructions for the assistant, not words from the "
           "caller. Do not reply to this message or read it aloud.")
_BODIES = [
    "Rules for questions you cannot answer.\n\n1. Hand off to Javier only for real-estate questions you "
    "have no facts for, or anything on the NEVER list below. Do not hand off small talk, the weather, the "
    "time of day, simple directions or general knowledge; answer those normally and then return to the "
    "property. The caller should feel they are talking with a helpful person, not a machine that refuses "
    "everything.",
    "When a hand-off is needed, vary the wording every time and never use the same line twice in a row. "
    "Example lines, to be used as a feel rather than a script:",
    '- "Well, I\'m not a licensed agent, so I\'ll leave that one to Javier."\n'
    '- "That\'s really a question for Javier, and I can pass it along."\n'
    '- "I\'d rather have Javier walk you through that himself."\n'
    '- "Javier knows that part much better than I do."\n'
    '- "Let me flag that one for Javier."\n'
    '- "That\'s outside what I can speak to, but Javier can."\n'
    '- "Bueno, eso es una pregunta para Javier."',
    "Say these in whatever language the caller is using. The goal is a warm hand-off in your own voice. "
    "Do not tack a request for the caller's name and number onto every hand-off; collecting contact "
    "details is a separate step described further below.",
    "2. NEVER:\n- rate, rank or compare schools, or quote test scores\n- give mortgage, tax, legal or "
    "investment advice\n- say what a property is worth, whether it is a good deal, or how it compares to "
    "other homes\n- say what a property may be used for, what it could earn, or anything about zoning\n"
    "- invent features that you were not given\n- claim to be a licensed agent; you are Hannah, Javier's "
    "assistant\n- promise a showing time; Javier confirms times himself\n- describe the neighborhood or "
    "the kind of people who live there, in any words at all; if asked, hand off to Javier",
    "3. If you are ever unsure, hand off to Javier using the examples above. Saying Javier will follow up "
    "is always a safe answer.\n\nSeveral questions in a row that you cannot answer:\nWhen the caller has "
    "asked two or three things you had to hand off, stop handing them off one by one. Instead, gather the "
    "open questions so Javier has the full list, and collect contact details once at the end.",
    "The steps:\na. For the first one or two hand-offs, just hand off and keep the conversation going.\n"
    "b. After two or three, ask once: \"Is there anything else you'd like me to ask Javier when he calls "
    "you back?\"\nc. While they add more, keep it open: \"Anything else?\"\nd. When they are done, ask for "
    "their name and the best number and time to call, once. Do not read this word for word.",
    "If the caller asks only one question you cannot answer, do not start this list. Just hand off that "
    "one item.\n\n4. If the caller is upset, stay calm, acknowledge it, and flag it for Javier.\n5. When "
    "you take a phone number, read it back one digit at a time.",
    "6. If the caller asks whether the call is recorded, answer truthfully in the first person: you are "
    "not recording. Only say this when asked. Do not explain at length and do not claim to stop a "
    "recording.\n7. If the caller gives you their name, use it and never ask for it again.\n\nClose every "
    "call warmly: \"Thanks for calling. Javier will be in touch.\"",
    "Listing questions.\n\n8. Answer questions about a home only from facts you were given for that home. "
    "If a detail is not in your facts (a pool, the lot size, the year it was built, the HOA fee, the "
    "schools), do not guess and do not ask the caller to look it up. Say you do not have that detail and "
    "hand it to Javier. A caller who hears a wrong fact from you will not trust anything else you say.",
    "9. When the caller names an area rather than an address, ask one short question to find the home they "
    "mean, such as the street or the price range. Do not list several homes at once. If you still cannot "
    "tell which home they mean after one question, take the details they gave you and let Javier sort it "
    "out. Never read out more than three facts in one turn; offer to go on instead.",
    "10. Say prices the way a person would say them aloud: \"about one point two million\", not a string of "
    "digits. Say square footage rounded to the nearest hundred. Never quote a monthly payment, an interest "
    "rate, a tax amount or an estimate of what the home would rent for, even if the caller asks you to "
    "\"just ballpark it\". Those are hand-offs to Javier.",
    "Showing requests.\n\n11. If the caller wants to see a home, ask which days generally work and whether "
    "mornings or afternoons are better. Do not offer specific times and do not say a time is available; "
    "you cannot see Javier's calendar. Tell them Javier will confirm a time. If they ask whether they can "
    "just stop by, say Javier will set that up with them.",
    "12. If the caller says they are working with another agent, thank them, take their agent's name if "
    "they offer it, and tell them Javier will coordinate with their agent. Do not ask why they called "
    "instead of their own agent, and do not comment on the other agent in any way.",
    "Messages.\n\n13. When the caller wants Javier to call back, collect the caller's name, the best number "
    "and the reason for the call, in that order, one question at a time. If the caller already gave any of "
    "these, do not ask again. Read the number back one digit at a time, then the name, and ask if both are "
    "right. If they correct anything, read back only the corrected part.",
    "14. Never promise when Javier will call. Say he will reach out as soon as he can. If the caller says "
    "it is urgent, say you will mark it urgent for him. Do not say Javier is busy, away, on vacation or "
    "with another client; you do not know where he is.",
    "Conversation style.\n\n15. Ask one question at a time and wait for the answer. Keep each turn short, "
    "one or two sentences. Do not repeat the caller's question back to them before you answer it. If you "
    "did not catch what they said, ask them to say it again rather than guessing. Background noise or a "
    "single unclear word is not a reason to change the subject or the language.",
    "16. If the caller says goodbye, thank them and end on a warm note; do not start a new topic. If they "
    "go quiet for a while, ask once whether there is anything else you can help with. If the caller asks "
    "whether you are a person or an AI, say plainly that you are an AI assistant for Javier.",
]
INSTRUCTIONS = [_HEADER.format(n=i + 1) + "\n\n" + b for i, b in enumerate(_BODIES)]

# (clip, what the caller says). Number words and digits both count as "said" for turn 6.
CALLER = [
    ("c1", "Good morning."),
    ("c2", "I was interested in a property over in Foothill Ranch."),
    ("c3", "Does it have a pool?"),
    ("c4", "What are the assigned schools?"),
    ("c5", "What is the neighborhood like?"),
    ("c6", "Can you have him call me back? My name is Sam, and the number is nine four nine, five five "
           "five, zero one two three. 949 555 0123 949-555-0123"),
]
SILENCE_AFTER_EACH_S = 14.0
_WORDS = re.compile(r"[a-z']+")


def config(silence_ms: int) -> types.LiveConnectConfig:
    return types.LiveConnectConfig(
        response_modalities=["AUDIO"],
        system_instruction=SYSTEM,
        input_audio_transcription=types.AudioTranscriptionConfig(),
        output_audio_transcription=types.AudioTranscriptionConfig(),
        realtime_input_config=types.RealtimeInputConfig(
            activity_handling=types.ActivityHandling.START_OF_ACTIVITY_INTERRUPTS,
            automatic_activity_detection=types.AutomaticActivityDetection(
                disabled=False,
                start_of_speech_sensitivity=types.StartSensitivity.START_SENSITIVITY_HIGH,
                end_of_speech_sensitivity=types.EndSensitivity.END_SENSITIVITY_HIGH,
                silence_duration_ms=silence_ms,
                prefix_padding_ms=100,
            ),
        ),
    )


def read_pcm(path: Path) -> bytes:
    with wave.open(str(path), "rb") as w:
        assert (w.getframerate(), w.getnchannels(), w.getsampwidth()) == (16000, 1, 2), path
        return w.readframes(w.getnframes())


def unspoken_words(heard: str, said: str) -> list[str]:
    spoken = set(_WORDS.findall(said.lower()))
    return [w for w in _WORDS.findall(heard.lower()) if w not in spoken]


async def run_session(client: genai.Client, model: str, audio_dir: Path, inject: bool, role: str,
                      silence_ms: int) -> dict:
    heard: dict[int, list[str]] = {}
    model_said: dict[int, list[str]] = {}
    turn = -1
    session_id = None
    started = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

    async with client.aio.live.connect(model=model, config=config(silence_ms)) as session:
        async def receive() -> None:
            nonlocal session_id
            while True:
                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.input_transcription and sc.input_transcription.text:
                        heard.setdefault(turn, []).append(sc.input_transcription.text)
                    if sc.output_transcription and sc.output_transcription.text:
                        model_said.setdefault(turn, []).append(sc.output_transcription.text)
                    if sc.turn_complete:
                        break

        receiver = asyncio.create_task(receive())
        if inject:
            # system-joined: every instruction in ONE role=system message (separate role=system
            # messages appeared to replace each other in our tests).
            msgs = [SYSTEM + "\n\n" + "\n\n".join(INSTRUCTIONS)] if role == "system-joined" else INSTRUCTIONS
            send_role = "system" if role == "system-joined" else role
            for text in msgs:
                await session.send_client_content(
                    turns=[types.Content(role=send_role, parts=[types.Part(text=text)])], turn_complete=False)
        await asyncio.sleep(1.5)

        for idx, (clip, _) in enumerate(CALLER):
            turn = idx
            pcm = read_pcm(audio_dir / f"{clip}.wav")
            chunks = [pcm[i:i + 3200] for i in range(0, len(pcm), 3200)]       # 100 ms
            chunks += [b"\x00\x00" * 1600] * int(SILENCE_AFTER_EACH_S * 10)    # 100 ms silence frames
            for c in chunks:                                                     # real time
                await session.send_realtime_input(audio=types.Blob(data=c, mime_type="audio/pcm;rate=16000"))
                await asyncio.sleep(0.1)
        await asyncio.sleep(0.5)
        receiver.cancel()

    turns = []
    for idx, (clip, said) in enumerate(CALLER):
        h = "".join(heard.get(idx, [])).strip()
        extra = unspoken_words(h, said)
        turns.append({"clip": clip, "caller_said": said.split(" 949")[0], "input_transcription": h,
                      "model_said": "".join(model_said.get(idx, [])).strip(),
                      "unspoken_words": extra, "contaminated": len(extra) >= 3})
    return {"started_utc": started, "session_id": session_id, "turns": turns,
            "contaminated_turns": sum(t["contaminated"] for t in turns)}


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("--runs", type=int, default=3)
    ap.add_argument("--no-inject", action="store_true", help="control: skip the mid-session instructions")
    ap.add_argument("--role", default="user", choices=["user", "model", "system", "system-joined"],
                    help="role of the injected instruction messages (default user)")
    ap.add_argument("--silence", type=int, default=200, help="silence_duration_ms (default 200)")
    ap.add_argument("--audio-dir", type=Path, default=Path(__file__).resolve().parent / "case1_audio")
    ap.add_argument("--out", type=Path, default=None, help="write results JSON here")
    a = ap.parse_args()

    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"])

    results = await asyncio.gather(*(run_session(client, a.model, a.audio_dir, not a.no_inject, a.role, a.silence)
                                     for _ in range(a.runs)), return_exceptions=True)

    label = f"{a.surface}:{a.model}{' (no injection)' if a.no_inject else f' (injected as role={a.role})'} silence={a.silence}ms"
    print(f"google-genai {genai.__version__}  {label}")
    for n, r in enumerate(results, 1):
        if isinstance(r, BaseException):
            print(f"\nsession {n}: ERROR {r!r}")
            continue
        print(f"\nsession {n} ({r['started_utc']}{', session_id=' + r['session_id'] if r['session_id'] else ''}): {r['contaminated_turns']} of {len(r['turns'])} turns contaminated")
        for t in r["turns"]:
            mark = "CONTAMINATED" if t["contaminated"] else "ok"
            print(f"  [{mark}] {t['clip']}  caller said:  {t['caller_said']}")
            print(f"  {' ' * (len(mark) + 2)} {' ' * len(t['clip'])}  transcription: {t['input_transcription']}")
            if t["contaminated"]:
                print(f"  {' ' * (len(mark) + 2)} {' ' * len(t['clip'])}  model said:    {t['model_said']}")
    ok = [r for r in results if isinstance(r, dict)]
    print(f"\nSUMMARY {label}: {sum(r['contaminated_turns'] > 0 for r in ok)} of {len(ok)} sessions contaminated")
    if a.out:
        a.out.write_text(json.dumps({"surface": a.surface, "model": a.model, "inject": not a.no_inject, "role": a.role, "silence_ms": a.silence,
                                     "google_genai": genai.__version__,
                                     "sessions": [r if isinstance(r, dict) else repr(r) for r in results]},
                                    indent=2, ensure_ascii=False), encoding="utf-8")


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