fix(runners): honor before_run_callback early-exit on the node execution path

The node execution path (_run_node_async) invoked
plugin_manager.run_before_run_callback but discarded its return value,
so a plugin returning types.Content (the documented signal to halt the
run) was ignored and execution continued. The legacy path
(_exec_with_plugin) already honors this contract.

This affects every root that dispatches through the node path: a
Workflow root and a root LlmAgent (chat/task mode), which means
plugin-based guardrails (e.g. safety filters that block a turn in
before_run_callback) are silently bypassed for those shapes.

Fix mirrors the _exec_with_plugin early-exit contract: the returned
Content becomes the final response event (with RunConfig
custom_metadata applied), is appended to the session, and the run ends.
after_run callbacks and compaction still execute via the enclosing
finally, matching the success path.

Adapted from the stale PR #6032 by @garyzava (rebased onto current main
and extended with a root-LlmAgent regression test).

Fixes #6828

Author: danielpartatus-lab

src/google/adk/runners.py

@@ -673,11 +673,32 @@ async def _run() -> AsyncGenerator[Event, None]:
               yield user_event
 
           # Run before_run callbacks
-          await ic.plugin_manager.run_before_run_callback(invocation_context=ic)
+          early_exit_result = await ic.plugin_manager.run_before_run_callback(
+              invocation_context=ic
+          )
         except Exception as e:
           await _notify_run_error(ic.plugin_manager, ic, e)
           raise
 
+        # A Content returned by before_run halts the run and becomes the
+        # final response, mirroring the early-exit contract of
+        # _exec_with_plugin. Returning here still runs after_run and
+        # compaction via the enclosing finally, same as a successful run.
+        if isinstance(early_exit_result, types.Content):
+          early_exit_event = Event(
+              invocation_id=ic.invocation_id,
+              author='model',
+              content=early_exit_result,
+          )
+          _apply_run_config_custom_metadata(early_exit_event, ic.run_config)
+          if self._should_append_event(early_exit_event, is_live_call=False):
+            await self.session_service.append_event(
+                session=ic.session,
+                event=early_exit_event,
+            )
+          yield early_exit_event
+          return
+
         # 3. Start root node in background
         from .agents.context import Context
         from .workflow._dynamic_node_scheduler import DynamicNodeScheduler

tests/unittests/workflow/test_workflow_failures.py

@@ -21,8 +21,11 @@
 
 from google.adk import platform as adk_platform
 from google.adk.agents.context import Context
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.agents.llm_agent import Agent
 from google.adk.apps.app import App
 from google.adk.events.event import Event
+from google.adk.plugins.base_plugin import BasePlugin
 # Added for the moved test
 from google.adk.runners import Runner
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
@@ -1162,3 +1165,70 @@ async def failing_node_2(ctx: Context):
 
   with pytest.raises(ValueError, match='Fail 1'):
     await runner.run_async(testing_utils.get_user_content('start'))
+
+
+class _HaltingPlugin(BasePlugin):
+  """Plugin whose before_run_callback halts the run with a Content."""
+
+  def __init__(self):
+    super().__init__(name='halting_plugin')
+
+  async def before_run_callback(
+      self, *, invocation_context: InvocationContext
+  ) -> types.Content:
+    return types.Content(
+        role='model', parts=[types.Part(text='halted by plugin')]
+    )
+
+
+def _texts(events: list[Event]) -> list[str]:
+  return [
+      part.text
+      for event in events
+      if event.content and event.content.parts
+      for part in event.content.parts
+      if part.text
+  ]
+
+
+@pytest.mark.asyncio
+async def test_workflow_halts_when_before_run_callback_returns_content(
+    request: pytest.FixtureRequest,
+):
+  """Regression for #6013: a Content returned by before_run_callback must
+  halt the run with that content and skip node execution."""
+  node_a = TestingNode(name='NodeA', output='should not run')
+  graph = Graph(edges=[Edge(from_node=START, to_node=node_a)])
+  workflow = Workflow(name='halt_workflow', graph=graph)
+
+  app = App(
+      name=request.function.__name__,
+      root_agent=workflow,
+      plugins=[_HaltingPlugin()],
+  )
+  runner = testing_utils.InMemoryRunner(app=app)
+  events = await runner.run_async(testing_utils.get_user_content('start'))
+
+  assert node_a.received_inputs == []
+  assert 'halted by plugin' in _texts(events)
+
+
+@pytest.mark.asyncio
+async def test_llm_agent_root_halts_when_before_run_callback_returns_content(
+    request: pytest.FixtureRequest,
+):
+  """Same regression for the other node-path shape: a root LlmAgent. The
+  model must never be called on a halted run."""
+  mock_model = testing_utils.MockModel.create(responses=['should not run'])
+  agent = Agent(name='root_agent', model=mock_model)
+
+  app = App(
+      name=request.function.__name__,
+      root_agent=agent,
+      plugins=[_HaltingPlugin()],
+  )
+  runner = testing_utils.InMemoryRunner(app=app)
+  events = await runner.run_async(testing_utils.get_user_content('hello'))
+
+  assert not mock_model.requests
+  assert 'halted by plugin' in _texts(events)