fix(agents): skip live after callbacks when invocation ends

[jaywang172]

Link to Issue or Description of Change

Fixes #7080

Companion documentation: google/adk-docs#2216
(should land with or after this runtime fix).

When a live agent sets end_invocation, its after-agent callbacks still run
and can emit another response or modify state. The async entry point already
skips those callbacks in the same situation.

Add the same post-execution check to BaseAgent.run_live(). The change is three
runtime lines and covers both plugin and agent-defined after callbacks. Normal
completion and the runner-level after-run hook retain their existing behavior.

Testing Plan

• Regression before the fix: 1 failed, 3 passed; only ended live execution fails.
• Regression after the fix: all four async/live × normal/ended cases pass.
  Full tox matrix (Python 3.10 completed sequentially; remaining versions ran
  with tox -e py311,py312,py313,py314 -p 4):

Python	Passed	Skipped	Xfailed	Xpassed	Failed
3.10	14,256	87	27	2	0
3.11	14,263	86	27	2	0
3.12	14,256	87	27	2	0
3.13	14,256	87	27	2	0
3.14 (uv-managed 3.14.6)	14,256	87	27	2	0

All five full-suite runs passed. The initial Homebrew 3.14 run had two
allowlisted-import failures caused by its startup sitecustomize module;
both reproduced on unmodified main. Recreating the environment with
tox -r -e py314 -x testenv.uv_python_preference=only-managed used Python
3.14.6 and passed the complete suite. No tests, allowlists, or repository
configuration were changed to suppress the failures.

A local ignored uv.lock was generated because the repository's tox runner
uses uv sync --locked; no dependency files are included in this PR.

• Focused regression tests also pass on Python 3.13 and 3.14.
• pre-commit run --files src/google/adk/agents/base_agent.py tests/unittests/agents/test_base_agent.py: passed.
• git diff --check: passed.
• uv build: passed. Installed the resulting wheel into a clean Python 3.11
  environment and ran the same Runner smoke test outside the source checkout;
  both cases passed.
• Targeted mypy check reports the same two pre-existing _run_callbacks
  argument-type errors on both the unchanged baseline and this patch; no new
  diagnostics. This check is not clean on the baseline.

Manual E2E: ran Runner.run_live() with a local BaseAgent, plugin, and
InMemorySessionService. Normal completion emitted the agent and after-agent
responses and invoked both after-agent hooks. With end_invocation=True, only
the agent response was emitted, both after-agent hooks were skipped, and
after_run_callback still ran. No model request or Google credentials were needed.

Reproducible Runner smoke test and output

Save the following as live_lifecycle_smoke.py, then run
PYTHONPATH=src python live_lifecycle_smoke.py from the checkout with ADK's
runtime dependencies installed.

import asyncio

from google.adk.agents.base_agent import BaseAgent
from google.adk.apps.app import App
from google.adk.events.event import Event
from google.adk.live.live_request_queue import LiveRequestQueue
from google.adk.plugins.base_plugin import BasePlugin
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types


async def probe(stop):
    calls = []

    class Agent(BaseAgent):
        async def _run_live_impl(self, ctx):
            yield Event(author=self.name, content=types.Content(parts=[types.Part(text='agent output')]))
            ctx.end_invocation = stop

    class Plugin(BasePlugin):
        async def after_agent_callback(self, *, agent, callback_context):
            calls.append('plugin.after_agent')

        async def after_run_callback(self, *, invocation_context):
            calls.append('plugin.after_run')

    def after(callback_context):
        calls.append('agent.after_agent')
        return types.Content(parts=[types.Part(text='after output')])

    service = InMemorySessionService()
    await service.create_session(app_name='app', user_id='u', session_id='s')
    runner = Runner(app=App(name='app', root_agent=Agent(name='agent', after_agent_callback=after), plugins=[Plugin(name='probe')]), session_service=service)
    try:
        events = [event async for event in runner.run_live(user_id='u', session_id='s', live_request_queue=LiveRequestQueue())]
        texts = [part.text for event in events if event.content for part in event.content.parts or [] if part.text]
        expected_calls = ['plugin.after_run'] if stop else ['plugin.after_agent', 'agent.after_agent', 'plugin.after_run']
        assert calls == expected_calls, calls
        assert texts == (['agent output'] if stop else ['agent output', 'after output']), texts
        print(f'end_invocation={stop}: output={texts}, callbacks={calls}')
    finally:
        await runner.close()


async def main():
    await probe(False)
    await probe(True)


asyncio.run(main())

end_invocation=False: output=['agent output', 'after output'], callbacks=['plugin.after_agent', 'agent.after_agent', 'plugin.after_run']
end_invocation=True: output=['agent output'], callbacks=['plugin.after_run']

Checklist

• Read CONTRIBUTING.md and performed a self-review.
• Added regression tests for the changed behavior and normal completion.
• Ran relevant existing tests, formatting, and compliance checks.
• Manually verified the live Runner path end to end.
• Prepared a companion callback-guide update in adk-docs.