before_run_callback early-exit is silently ignored on the node execution path (root LlmAgent and Workflow) — reopen of #6013

[Dratatus]

Describe the bug
Runner has two execution paths. The legacy path (_exec_with_plugin) honors the documented before_run_callback contract: a plugin returning types.Content halts the run and that content becomes the final response. The node execution path (_run_node_async) calls plugin_manager.run_before_run_callback(...) but discards the return value, so the run continues as if the plugin had returned None.

This was reported in #6013 (confirmed by adk-bot as high priority) but that issue was closed by the stale bot on 2026-07-04 without a fix, and the fix PR #6032 has been waiting for human review since June and no longer merges cleanly.

Why this matters
Every root that dispatches through the node path is affected: a Workflow root and, since 2.7.0, any root LlmAgent. In practice, plugin-based guardrails (safety filters that block a turn in before_run_callback, e.g. Model Armor-style input screening) are silently bypassed for the most common application shape. We hit this while hardening a production healthcare analytics assistant: the blocked turn reaches the model anyway.

To Reproduce
Minimal repro on google-adk 2.7.0 (also reproduces on current main):

import asyncio
from google.adk.agents.llm_agent import Agent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.apps.app import App
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

class HaltPlugin(BasePlugin):
    def __init__(self):
        super().__init__(name="halt")
    async def before_run_callback(self, *, invocation_context: InvocationContext):
        return types.Content(role="model", parts=[types.Part(text="halted by plugin")])

async def main():
    agent = Agent(name="root_agent", model="gemini-2.5-flash")
    ss = InMemorySessionService()
    runner = Runner(app=App(name="repro", root_agent=agent, plugins=[HaltPlugin()]),
                    session_service=ss)
    session = await ss.create_session(app_name="repro", user_id="u")
    async for event in runner.run_async(
        user_id="u", session_id=session.id,
        new_message=types.Content(role="user", parts=[types.Part(text="hi")]),
    ):
        print(event.author, event.content)

asyncio.run(main())

Expected behavior
The run halts with the plugin's content as the final response and the model is never called — the behavior _exec_with_plugin already implements, and what the plugin docs promise ("Returning a value to halt execution of the runner and ends the runner with that event").

Actual behavior
The plugin's content is discarded and the model generates a normal response.

Desktop:

• OS: macOS 15
• Python version: 3.12
• ADK version: 2.7.0 (also reproduced on main @ 4599a52)

Additional context
We have prepared a PR that rebases the stale fix #6032 onto current main (with credit to its author) and extends it with a root-LlmAgent regression test. Both new tests fail without the fix and pass with it; the full tests/unittests/test_runners.py and tests/unittests/workflow/ suites pass (738 passed). Happy to iterate on review feedback.

[Dratatus]

Closing for now — will be resubmitted together with the fix by the colleague who did the original investigation.