Call tools during a live conversation

"""Record current Live tool runs; optionally include the historical model negative control."""
import argparse
import asyncio
import json
from datetime import datetime, timezone
from pathlib import Path

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from google import genai
from google.genai import types

from live_harness import RecordingGemini, environment, MODEL, PACE, read_pcm, say_audio, say_text, spoken, talk, wait_for_turn
from live_tools import TOOLS_MODEL, build_agent

HERE = Path(__file__).parent
# Explicit historical negative control, not the runnable agent default.
HISTORICAL_MODEL = "gemini-2.5-flash-native-audio-preview-12-2025"
from google.adk.agents.run_config import RunConfig

RUN_CONFIG = RunConfig(response_modalities=["AUDIO"])
LOOKUP_CONFIG = RunConfig(response_modalities=["AUDIO"], realtime_input_config=types.RealtimeInputConfig(
    automatic_activity_detection=types.AutomaticActivityDetection(disabled=True)))


def read_page(name: str) -> dict:
    """Return the text of one docs page.

    Args:
      name: The page file name, for example "tools.md".
    """
    return {"name": name, "text": (HERE / "docs" / name).read_text()}


async def run(name, turns, agent_factory=build_agent, model_id=TOOLS_MODEL, timeout=90, run_config=RUN_CONFIG):
    model = RecordingGemini(model=model_id, connections=[])
    runner = Runner(app_name="voice_docs", agent=agent_factory(model), session_service=InMemorySessionService())
    session = await runner.session_service.create_session(app_name="voice_docs", user_id="dev")
    entry = {"name": name}
    try:
        rows = await talk(runner, session.id, run_config, turns, timeout=timeout)
    except asyncio.TimeoutError as error:
        rows, entry["error"] = error.rows, f"TimeoutError: no further turn within {timeout} s"
    except Exception as error:
        message = str(error).splitlines()[0].replace(str(HERE) + "/", "")  # keep local paths out of the record
        rows, entry["error"] = [], f"{type(error).__name__}: {message[:240]}"
    entry.update(events=rows, answer=spoken(rows), connections=model.connections)
    calls = [c["name"] for r in rows for c in r.get("calls", [])]
    print(f"[{name}] calls={calls} said={entry['answer'][:200]!r}" + (f" error={entry['error']}" if "error" in entry else ""))
    return entry


def push_to_talk(pcm):
    """Send one finite question with the same explicit activity signals as the media lesson."""
    async def send(queue):
        queue.send_activity_start()
        for start in range(0, len(pcm), 3200):
            queue.send_realtime(types.Blob(mime_type="audio/pcm;rate=16000", data=pcm[start:start + 3200]))
            await asyncio.sleep(PACE * 0.1)
        queue.send_activity_end()
    return send


def page_reader(model):
    return Agent(name="docs_helper", model=model, tools=[read_page],
                 instruction="You are the docs assistant. Use read_page to read pages. Answer in one short sentence.")


async def raw_tool_call(model_id, timeout=25):
    """Without ADK: which server messages follow a tool call on this model."""
    declaration = types.FunctionDeclaration(name="search_docs", description="Search the docs pages.", parameters=types.Schema(
        type="OBJECT", properties={"query": types.Schema(type="STRING")}, required=["query"]))
    config = types.LiveConnectConfig(response_modalities=["AUDIO"], system_instruction="Use search_docs to look facts up.",
                                     tools=[types.Tool(function_declarations=[declaration])])
    seen = []
    async with genai.Client().aio.live.connect(model=model_id, config=config) as live:
        await live.send_client_content(turns=types.Content(role="user", parts=[types.Part(text=LOOKUP)]))

        async def receive():
            async for message in live.receive():
                if message.tool_call:
                    seen.append("tool_call")
                if message.server_content and message.server_content.turn_complete:
                    seen.append("turn_complete")
                    return
        try:
            await asyncio.wait_for(receive(), timeout)
        except asyncio.TimeoutError:
            seen.append(f"nothing more in {timeout} s")
    print(f"[raw Live API, {model_id}] {seen}")
    return {"model": model_id, "messages": seen}


LOOKUP = "How long does a Live API connection stay open? Look it up."


async def main(include_historical_models=False):
    started_at = datetime.now(timezone.utc).isoformat()
    runs = [
        await run("blocking lookup", [push_to_talk(read_pcm(HERE / "lookup.wav"))], run_config=LOOKUP_CONFIG),
        await run("streaming link check, then stop", [say_text("Check the links in the docs pages."),
                                                      wait_for_turn(), say_text("Never mind, stop the link check.")], timeout=60),
        await run("scheduled export on current Live model",
                  [say_text("Start the page index export. While it runs, tell me the input sample rate."), wait_for_turn()], timeout=60),
        await run("tool raises an exception", [say_text("Read the page deploy.md.")], agent_factory=page_reader, timeout=40),
    ]
    raw = [await raw_tool_call(TOOLS_MODEL)]
    if include_historical_models:
        runs.append(await run("tools on Gemini 2.5 Flash Live", [say_text(LOOKUP)], model_id=HISTORICAL_MODEL, timeout=40))
        raw.append(await raw_tool_call(HISTORICAL_MODEL))
    (HERE / "tools-run.json").write_text(json.dumps({"recorded_at": started_at, "environment": environment() | {"tools_model": TOOLS_MODEL},
                                                     "runs": runs, "raw_tool_call": raw}, indent=1) + "\n")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--include-historical-models", action="store_true",
                        help="Also rerun the explicitly historical Gemini 2.5 negative control")
    asyncio.run(main(parser.parse_args().include_historical_models))

"""Replay the recorded tool calls through ADK's run_live loop; ADK runs the real tools. No network."""
import asyncio
import copy
import json
import logging
import unittest
from pathlib import Path

from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import FunctionTool
from google.genai import types

import live_harness
import live_tools
from live_harness import ReplayLive, read_pcm, say_audio, say_text, talk, wait_for_turn
from live_tools import build_agent, search_docs
from record import RUN_CONFIG, LOOKUP_CONFIG, page_reader, push_to_talk

HERE = Path(__file__).parent
RECORD = json.loads((HERE / "tools-run-historical.json").read_text())
RUNS = {run["name"]: run for run in RECORD["runs"]}
CURRENT = json.loads((HERE / "tools-run.json").read_text())
CURRENT_RUNS = {run["name"]: run for run in CURRENT["runs"]}
BEFORE_ACTIVITY = json.loads((HERE / "tools-run-before-activity-signals.json").read_text())
live_harness.PACE = 0
live_tools.SLOW_S = 0.05
logging.disable(logging.CRITICAL)


def converse(name, turns, agent_factory=build_agent, timeout=10, runs=RUNS, run_config=RUN_CONFIG):
    model = ReplayLive(model=runs[name]["connections"][0]["model"],
                       connections=copy.deepcopy(runs[name]["connections"]), opened=[])

    async def run():
        runner = Runner(app_name="voice_docs", agent=agent_factory(model), session_service=InMemorySessionService())
        session = await runner.session_service.create_session(app_name="voice_docs", user_id="dev")
        return await talk(runner, session.id, run_config, turns, timeout=timeout)
    return asyncio.run(run()), model


def tool_results(model, name):
    return [part.function_response.response for _, connection in model.opened for kind, item in connection.received
            if kind == "content" for part in item.parts or [] if part.function_response and part.function_response.name == name]


def finished_lines(run):
    return [r["output_transcription"]["text"] for r in run["events"]
            if "output_transcription" in r and not r.get("partial")]


def calls(run):
    return [call["name"] for row in run["events"] for call in row.get("calls", [])]


class LiveTools(unittest.TestCase):
    def test_push_to_talk_marks_both_activity_boundaries(self):
        async def check():
            queue = LiveRequestQueue()
            await push_to_talk(bytes(320))(queue)
            requests = [await asyncio.wait_for(queue.get(), 0.1) for _ in range(3)]
            self.assertIsNotNone(requests[0].activity_start)
            self.assertEqual(requests[1].blob.mime_type, "audio/pcm;rate=16000")
            self.assertIsNotNone(requests[2].activity_end)
        asyncio.run(check())

    def test_activity_signals_complete_the_recorded_lookup(self):
        failed = BEFORE_ACTIVITY["runs"][0]
        self.assertEqual(BEFORE_ACTIVITY["environment"]["model"], "gemini-3.8-live")
        self.assertIn("TimeoutError", failed["error"])
        self.assertFalse(calls(failed))
        current = CURRENT_RUNS["blocking lookup"]
        self.assertEqual(CURRENT["environment"]["tools_model"], "gemini-3.8-live")
        self.assertNotIn("error", current)
        [connection] = current["connections"]
        controls = [e.get("mime_type") for e in connection["traffic"] if e.get("sent") == "realtime"]
        self.assertIn("ActivityStart", controls)
        self.assertIn("ActivityEnd", controls)
        self.assertTrue(connection["config"]["realtime_input_config"]["automatic_activity_detection"]["disabled"])
        behaviors = {tool["name"]: tool["behavior"] for tool in connection["tools"]}
        self.assertEqual(behaviors["search_docs"], "BLOCKING")
        _, model = converse("blocking lookup", [push_to_talk(read_pcm(HERE / "lookup.wav"))], runs=CURRENT_RUNS, run_config=LOOKUP_CONFIG)
        results = tool_results(model, "search_docs")
        self.assertTrue(any("10 minutes" in hit["line"] for result in results for hit in result["hits"]))

    def test_lookup_explicitly_declares_blocking_for_current_live_model(self):
        lookup = build_agent().tools[0]
        self.assertIsInstance(lookup, FunctionTool)
        self.assertEqual(lookup.name, "search_docs")
        self.assertEqual(lookup.behavior, types.Behavior.BLOCKING)

    def test_adk_runs_the_lookup_and_sends_the_result_back(self):
        _, model = converse("blocking lookup", [say_audio(read_pcm(HERE / "lookup.wav"))])
        [result] = tool_results(model, "search_docs")
        self.assertTrue(any("10 minutes" in hit["line"] for hit in result["hits"]))

    def test_search_reports_a_miss_instead_of_inventing_a_hit(self):
        self.assertEqual(search_docs("kubernetes helm chart")["hits"], [])

    def test_async_tools_are_declared_non_blocking(self):
        behaviors = {t["name"]: t["behavior"] for t in RUNS["blocking lookup"]["connections"][0]["tools"]}
        self.assertEqual(behaviors, {"search_docs": None, "export_page_index": "NON_BLOCKING",
                                     "check_links": "NON_BLOCKING", "stop_streaming": None})

    def test_each_streamed_update_goes_back_to_the_model(self):
        turns = [say_text("Check the links in the docs pages."), wait_for_turn(), say_text("Never mind, stop the link check.")]
        _, model = converse("streaming link check, then stop", turns)
        updates = [r["result"] for r in tool_results(model, "check_links") if "result" in r]
        self.assertTrue(updates[0].startswith("Checked "), updates)
        self.assertEqual(updates[-1], "Done: 3 pages checked.")

    def test_the_pending_status_is_spoken_as_its_own_turn(self):
        lines = finished_lines(RUNS["streaming link check, then stop"])
        self.assertEqual(lines[0].lower(), "async return")

    def test_streamed_updates_reach_the_model_but_are_not_narrated(self):
        run = RUNS["streaming link check, then stop"]
        sent = [e for e in run["connections"][0]["traffic"] if e.get("function_responses") == ["check_links"]]
        self.assertGreaterEqual(len(sent), 4)  # the pending status, then each yield
        before_stop = finished_lines(run)[:2]
        self.assertFalse(any("of 3" in line or "Checked" in line for line in before_stop), before_stop)

    def test_a_stop_after_the_stream_finished_finds_nothing(self):
        run = RUNS["streaming link check, then stop"]
        [stop] = [r["response"] for row in run["events"] for r in row.get("responses", []) if r["name"] == "stop_streaming"]
        self.assertIn("No active streaming function", stop["status"])

    def test_a_raising_tool_ends_the_whole_live_session(self):
        with self.assertRaises(FileNotFoundError):
            converse("tool raises an exception", [say_text("Read the page deploy.md.")], agent_factory=page_reader)
        self.assertTrue(RUNS["tool raises an exception"]["error"].startswith("FileNotFoundError"))

    def test_scheduled_result_on_gemini_3_1_is_never_announced(self):
        run = RUNS["scheduled export on Gemini 3.1 Flash Live"]
        self.assertIn("export_page_index", [r["name"] for row in run["events"] for r in row.get("responses", [])])
        self.assertIn("TimeoutError", run["error"])

    def test_gemini_2_5_tool_calls_never_reach_adk(self):
        self.assertEqual(calls(RUNS["tools on Gemini 2.5 Flash Live"]), [])
        raw = {r["model"]: r["messages"] for r in RECORD["raw_tool_call"]}
        self.assertEqual(raw[RECORD["environment"]["model"]][0], "tool_call")
        self.assertNotIn("turn_complete", raw[RECORD["environment"]["model"]])


if __name__ == "__main__":
    unittest.main()