diff --git a/contributing/samples/runner_debug_example/README.md b/contributing/samples/runner_debug_example/README.md
new file mode 100644
index 0000000000..c8a07b4202
--- /dev/null
+++ b/contributing/samples/runner_debug_example/README.md
@@ -0,0 +1,217 @@
+# Runner Debug Helper Example
+
+This example demonstrates the `run_debug()` helper method that simplifies agent interaction for debugging and experimentation in ADK.
+
+## Overview
+
+The `run_debug()` method reduces agent interaction boilerplate from 7-8 lines to just 2 lines, making it ideal for:
+
+- Quick debugging sessions
+- Jupyter notebooks
+- REPL experimentation
+- Writing examples
+- Initial agent development
+
+## Files Included
+
+- `agent.py` - Agent with 2 tools: weather and calculate
+- `main.py` - 8 examples demonstrating all features
+- `README.md` - This documentation
+
+## Setup
+
+### Prerequisites
+
+Set your Google API key:
+
+```bash
+export GOOGLE_API_KEY="your-api-key"
+```
+
+### Running the Example
+
+```bash
+python -m contributing.samples.runner_debug_example.main
+```
+
+## Features Demonstrated
+
+1. **Minimal Usage**: Simple 2-line agent interaction
+2. **Multiple Queries**: Processing multiple queries in sequence
+3. **Session Persistence**: Maintaining conversation context
+4. **Separate Sessions**: Managing multiple user sessions
+5. **Tool Calls**: Displaying tool invocations and results
+6. **Event Capture**: Collecting events for programmatic inspection
+7. **Advanced Configuration**: Using RunConfig for custom settings
+8. **Comparison**: Before/after boilerplate reduction
+
+## Part Types Supported
+
+The `run_debug()` method properly displays all ADK part types:
+
+| Part Type | Display Format | Use Case |
+|-----------|---------------|----------|
+| `text` | `agent > {text}` | Regular text responses |
+| `function_call` | `agent > [Calling tool: {name}({args})]` | Tool invocations |
+| `function_response` | `agent > [Tool result: {response}]` | Tool results |
+| `executable_code` | `agent > [Executing {language} code...]` | Code blocks |
+| `code_execution_result` | `agent > [Code output: {output}]` | Code execution results |
+| `inline_data` | `agent > [Inline data: {mime_type}]` | Images, files, etc. |
+| `file_data` | `agent > [File: {uri}]` | File references |
+
+## Tools Available in Example
+
+The example agent includes 2 tools to demonstrate tool handling:
+
+1. **`get_weather(city)`** - Returns mock weather data for major cities
+2. **`calculate(expression)`** - Evaluates mathematical expressions safely
+
+## Key Benefits
+
+### Before (7-8 lines)
+
+```python
+from google.adk.sessions import InMemorySessionService
+from google.genai import types
+
+APP_NAME = "default"
+USER_ID = "default"
+session_service = InMemorySessionService()
+runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
+session = await session_service.create_session(
+    app_name=APP_NAME, user_id=USER_ID, session_id="default"
+)
+content = types.Content(role="user", parts=[types.Part.from_text("Hi")])
+async for event in runner.run_async(
+    user_id=USER_ID, session_id=session.id, new_message=content
+):
+    if event.content and event.content.parts:
+        print(event.content.parts[0].text)
+```
+
+### After (2 lines)
+
+```python
+runner = InMemoryRunner(agent=agent)
+await runner.run_debug("Hi")
+```
+
+## API Reference
+
+```python
+async def run_debug(
+    self,
+    user_queries: str | list[str] | None = None,
+    *,
+    user_id: str = 'default',
+    session_name: str = 'default',
+    print_output: bool = True,
+    return_events: bool = False,
+    verbose: bool = False,
+    run_config: Optional[RunConfig] = None,
+) -> Optional[List[Event]]:
+```
+
+### Parameters
+
+- `user_queries`: Single query string or list of queries
+- `user_id`: User identifier for session tracking
+- `session_name`: Session identifier for conversation continuity
+- `print_output`: Whether to print responses to console
+- `return_events`: Whether to return event list
+- `verbose`: Whether to show detailed tool calls and responses (default: False)
+- `run_config`: Optional advanced configuration
+
+### Usage Examples
+
+```python
+# Minimal usage
+runner = InMemoryRunner(agent=agent)
+await runner.run_debug("What's the weather?")
+
+# Multiple queries
+await runner.run_debug(["Query 1", "Query 2", "Query 3"])
+
+# Custom session
+await runner.run_debug(
+    "Hello",
+    user_id="alice",
+    session_name="debug_session"
+)
+
+# Capture events without printing
+events = await runner.run_debug(
+    "Process this",
+    print_output=False,
+    return_events=True
+)
+
+# Show tool calls with verbose mode
+await runner.run_debug(
+    "What's the weather?",
+    verbose=True  # Shows [Calling tool: ...] and [Tool result: ...]
+)
+
+# With custom configuration
+from google.adk.agents.run_config import RunConfig
+config = RunConfig(support_cfc=False)
+await runner.run_debug("Query", run_config=config)
+```
+
+## Troubleshooting
+
+### Common Issues and Solutions
+
+1. **Tool calls not showing in output**
+   - **Issue**: Tool invocations and responses are not displayed
+   - **Solution**: Set `verbose=True` to see detailed tool interactions:
+
+     ```python
+     await runner.run_debug("Query", verbose=True)
+     ```
+
+2. **Import errors when running tests**
+   - **Issue**: `ModuleNotFoundError: No module named 'google.adk'`
+   - **Solution**: Ensure you're using the virtual environment:
+
+     ```bash
+     source .venv/bin/activate
+     python -m pytest tests/
+     ```
+
+3. **Session state not persisting between calls**
+   - **Issue**: Agent doesn't remember previous interactions
+   - **Solution**: Use the same `user_id` and `session_name` across calls:
+
+     ```python
+     await runner.run_debug("First query", user_id="alice", session_name="debug")
+     await runner.run_debug("Follow-up", user_id="alice", session_name="debug")
+     ```
+
+4. **Output truncation issues**
+   - **Issue**: Long tool responses are truncated with "..."
+   - **Solution**: This is by design to keep debug output readable. For full responses, use:
+
+     ```python
+     events = await runner.run_debug("Query", print_output=False, return_events=True)
+     # Process events programmatically for full content
+     ```
+
+5. **API key errors**
+   - **Issue**: Authentication failures or missing API key
+   - **Solution**: Ensure your Google API key is set:
+
+     ```bash
+     export GOOGLE_API_KEY="your-api-key"
+     ```
+
+## Important Notes
+
+`run_debug()` is designed for debugging and experimentation only. For production use requiring:
+
+- Custom session/memory services (Spanner, Cloud SQL)
+- Fine-grained event processing
+- Error recovery and resumability
+- Performance optimization
+
+Use the standard `run_async()` method instead.
diff --git a/contributing/samples/runner_debug_example/__init__.py b/contributing/samples/runner_debug_example/__init__.py
new file mode 100644
index 0000000000..1ca56dac2b
--- /dev/null
+++ b/contributing/samples/runner_debug_example/__init__.py
@@ -0,0 +1,17 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Runner debug example demonstrating simplified agent interaction."""
+
+from . import agent
diff --git a/contributing/samples/runner_debug_example/agent.py b/contributing/samples/runner_debug_example/agent.py
new file mode 100644
index 0000000000..029f78b432
--- /dev/null
+++ b/contributing/samples/runner_debug_example/agent.py
@@ -0,0 +1,124 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Example agent for demonstrating run_debug helper method."""
+
+from google.adk import Agent
+from google.adk.tools.tool_context import ToolContext
+
+
+def get_weather(city: str, tool_context: ToolContext) -> str:
+  """Get weather information for a city.
+
+  Args:
+      city: Name of the city to get weather for.
+      tool_context: Tool context for session state.
+
+  Returns:
+      Weather information as a string.
+  """
+  # Store query history in session state
+  if "weather_queries" not in tool_context.state:
+    tool_context.state["weather_queries"] = []
+  tool_context.state["weather_queries"].append(city)
+
+  # Mock weather data for demonstration
+  weather_data = {
+      "San Francisco": "Foggy, 15°C (59°F)",
+      "New York": "Sunny, 22°C (72°F)",
+      "London": "Rainy, 12°C (54°F)",
+      "Tokyo": "Clear, 25°C (77°F)",
+      "Paris": "Cloudy, 18°C (64°F)",
+  }
+
+  return weather_data.get(
+      city, f"Weather data not available for {city}. Try a major city."
+  )
+
+
+def calculate(expression: str) -> str:
+  """Safely evaluate a mathematical expression.
+
+  This tool demonstrates how function calls are displayed in run_debug().
+
+  Args:
+      expression: Mathematical expression to evaluate.
+
+  Returns:
+      Result of the calculation as a string.
+  """
+  import ast
+  import operator
+
+  # Supported operators for safe evaluation
+  operators = {
+      ast.Add: operator.add,
+      ast.Sub: operator.sub,
+      ast.Mult: operator.mul,
+      ast.Div: operator.truediv,
+      ast.Pow: operator.pow,
+      ast.USub: operator.neg,
+  }
+
+  def _eval(node):
+    """Recursively evaluate an AST node."""
+    if isinstance(node, ast.Expression):
+      return _eval(node.body)
+    elif isinstance(node, ast.Constant):  # Python 3.8+
+      return node.value
+    elif isinstance(node, ast.Num):  # For older Python versions
+      return node.n
+    elif isinstance(node, ast.BinOp):
+      op = operators.get(type(node.op))
+      if op:
+        return op(_eval(node.left), _eval(node.right))
+      else:
+        raise ValueError(f"Unsupported operation: {type(node.op).__name__}")
+    elif isinstance(node, ast.UnaryOp):
+      op = operators.get(type(node.op))
+      if op:
+        return op(_eval(node.operand))
+      else:
+        raise ValueError(f"Unsupported operation: {type(node.op).__name__}")
+    else:
+      raise ValueError(f"Unsupported expression type: {type(node).__name__}")
+
+  try:
+    # Parse the expression into an AST
+    tree = ast.parse(expression, mode="eval")
+    # Safely evaluate the AST
+    result = _eval(tree)
+    return f"Result: {result}"
+  except (SyntaxError, ValueError) as e:
+    return f"Error: {str(e)}"
+  except ZeroDivisionError:
+    return "Error: Division by zero"
+  except Exception as e:
+    return f"Error: {str(e)}"
+
+
+root_agent = Agent(
+    model="gemini-2.5-flash-lite",
+    name="agent",
+    description="A helpful assistant demonstrating run_debug() helper method",
+    instruction="""You are a helpful assistant that can:
+    1. Provide weather information for major cities
+    2. Perform mathematical calculations
+    3. Remember previous queries in the conversation
+
+    When users ask about weather, use the get_weather tool.
+    When users ask for calculations, use the calculate tool.
+    Be friendly and conversational.""",
+    tools=[get_weather, calculate],
+)
diff --git a/contributing/samples/runner_debug_example/main.py b/contributing/samples/runner_debug_example/main.py
new file mode 100644
index 0000000000..e8ee9f88e2
--- /dev/null
+++ b/contributing/samples/runner_debug_example/main.py
@@ -0,0 +1,259 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Demonstrates the run_debug() helper method for simplified agent interaction."""
+
+import asyncio
+
+from google.adk.runners import InMemoryRunner
+
+from . import agent
+
+
+async def example_minimal():
+  """Minimal usage - just 2 lines for debugging."""
+  print("------------------------------------")
+  print("Example 1: Minimal Debug Usage")
+  print("------------------------------------")
+
+  # Create runner
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # Debug with just 2 lines
+  await runner.run_debug("What's the weather in San Francisco?")
+
+
+async def example_multiple_queries():
+  """Debug with multiple queries in sequence."""
+  print("\n------------------------------------")
+  print("Example 2: Multiple Queries")
+  print("------------------------------------")
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # Pass multiple queries as a list
+  await runner.run_debug([
+      "Hi there!",
+      "What's the weather in Tokyo?",
+      "How about New York?",
+      "Calculate 15 * 7 + 3",
+  ])
+
+
+async def example_conversation_persistence():
+  """Demonstrate conversation persistence during debugging."""
+  print("\n------------------------------------")
+  print("Example 3: Session Persistence")
+  print("------------------------------------")
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # First interaction
+  await runner.run_debug("Hi, I'm planning a trip to Europe")
+
+  # Second interaction - continues same session
+  await runner.run_debug("What's the weather in Paris?")
+
+  # Third interaction - agent remembers context
+  await runner.run_debug("And London?")
+
+  # Fourth interaction - referring to previous queries
+  await runner.run_debug("Which city had better weather?")
+
+
+async def example_separate_sessions():
+  """Debug with multiple separate sessions."""
+  print("\n------------------------------------")
+  print("Example 4: Separate Sessions")
+  print("------------------------------------")
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # Alice's session
+  print("\n-- Alice's session --")
+  await runner.run_debug(
+      "What's the weather in San Francisco?",
+      user_id="alice",
+      session_name="alice_debug",
+  )
+
+  # Bob's session (separate)
+  print("\n-- Bob's session --")
+  await runner.run_debug(
+      "Calculate 100 / 5", user_id="bob", session_name="bob_debug"
+  )
+
+  # Continue Alice's session
+  print("\n-- Back to Alice's session --")
+  await runner.run_debug(
+      "Should I bring an umbrella?",
+      user_id="alice",
+      session_name="alice_debug",
+  )
+
+
+async def example_with_tools():
+  """Demonstrate tool calls and responses with verbose flag."""
+  print("\n------------------------------------")
+  print("Example 5: Tool Calls (verbose flag)")
+  print("------------------------------------")
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  print("\n-- Default (verbose=False) - Clean output --")
+  # Without verbose: Only shows final agent responses
+  await runner.run_debug([
+      "What's the weather in Tokyo?",
+      "Calculate (42 * 3.14) + 10",
+  ])
+
+  print("\n-- With verbose=True - Detailed output --")
+  # With verbose: Shows tool calls as [Calling tool: ...] and [Tool result: ...]
+  await runner.run_debug(
+      [
+          "What's the weather in Paris?",
+          "Calculate 100 / 5",
+      ],
+      verbose=True,
+  )
+
+
+async def example_capture_events():
+  """Capture events for inspection during debugging."""
+  print("\n------------------------------------")
+  print("Example 6: Capture Events (No Print)")
+  print("------------------------------------")
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # Capture events without printing for inspection
+  events = await runner.run_debug(
+      ["Get weather for London", "Calculate 42 * 3.14"],
+      print_output=False,
+      return_events=True,
+  )
+
+  # Inspect the captured events
+  print(f"Captured {len(events)} events")
+  for i, event in enumerate(events):
+    if event.content and event.content.parts:
+      for part in event.content.parts:
+        if part.text:
+          print(f"  Event {i+1}: {event.author} - Text: {len(part.text)} chars")
+        elif hasattr(part, "function_call") and part.function_call:
+          print(
+              f"  Event {i+1}: {event.author} - Tool call:"
+              f" {part.function_call.name}"
+          )
+        elif hasattr(part, "function_response") and part.function_response:
+          print(f"  Event {i+1}: {event.author} - Tool response received")
+
+
+async def example_with_run_config():
+  """Demonstrate using RunConfig for advanced settings."""
+  print("\n------------------------------------")
+  print("Example 7: Advanced Configuration")
+  print("------------------------------------")
+
+  from google.adk.agents.run_config import RunConfig
+
+  runner = InMemoryRunner(agent=agent.root_agent)
+
+  # Custom configuration - RunConfig supports:
+  # - support_cfc: Control function calling behavior
+  # - response_modalities: Output modalities (for LIVE API)
+  # - speech_config: Speech settings (for LIVE API)
+  config = RunConfig(
+      support_cfc=False,  # Disable controlled function calling
+  )
+
+  await runner.run_debug(
+      "Explain what tools you have available", run_config=config
+  )
+
+
+async def example_comparison():
+  """Show before/after comparison of boilerplate reduction."""
+  print("\n------------------------------------")
+  print("Example 8: Before vs After Comparison")
+  print("------------------------------------")
+
+  print("\nBefore (7-8 lines of boilerplate):")
+  print("""
+  from google.adk.sessions import InMemorySessionService
+  from google.genai import types
+
+  APP_NAME = "default"
+  USER_ID = "default"
+  session_service = InMemorySessionService()
+  runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
+  session = await session_service.create_session(
+      app_name=APP_NAME, user_id=USER_ID, session_id="default"
+  )
+  content = types.Content(role="user", parts=[types.Part.from_text("Hi")])
+  async for event in runner.run_async(
+      user_id=USER_ID, session_id=session.id, new_message=content
+  ):
+      if event.content and event.content.parts:
+          print(event.content.parts[0].text)
+  """)
+
+  print("\nAfter (just 2 lines):")
+  print("""
+  runner = InMemoryRunner(agent=agent)
+  await runner.run_debug("Hi")
+  """)
+
+  print("\nThat's a 75% reduction in boilerplate.")
+
+
+async def main():
+  """Run all debug examples."""
+  print("ADK run_debug() Helper Method Examples")
+  print("=======================================")
+  print("Demonstrating all capabilities:\n")
+  print("1. Minimal usage (2 lines)")
+  print("2. Multiple queries")
+  print("3. Session persistence")
+  print("4. Separate sessions")
+  print("5. Tool calls")
+  print("6. Event capture")
+  print("7. Advanced configuration")
+  print("8. Before/after comparison")
+
+  await example_minimal()
+  await example_multiple_queries()
+  await example_conversation_persistence()
+  await example_separate_sessions()
+  await example_with_tools()
+  await example_capture_events()
+  await example_with_run_config()
+  await example_comparison()
+
+  print("\n=======================================")
+  print("All examples completed.")
+  print("\nHow different part types appear:")
+  print("  Text: agent > Hello world (always shown)")
+  print("\nWith verbose=True only:")
+  print("  Tool call: agent > [Calling tool: calculate({'expression': '2+2'})]")
+  print("  Tool result: agent > [Tool result: Result: 4]")
+  print("\nNote: When models have code execution enabled (verbose=True):")
+  print("  Code exec: agent > [Executing python code...]")
+  print("  Code output: agent > [Code output: Result: 42]")
+  print("  Inline data: agent > [Inline data: image/png]")
+  print("  File ref: agent > [File: gs://bucket/file.pdf]")
+
+
+if __name__ == "__main__":
+  asyncio.run(main())
diff --git a/src/google/adk/runners.py b/src/google/adk/runners.py
index 50197dda2f..c03192b3f8 100644
--- a/src/google/adk/runners.py
+++ b/src/google/adk/runners.py
@@ -46,6 +46,7 @@
 from .artifacts.in_memory_artifact_service import InMemoryArtifactService
 from .auth.credential_service.base_credential_service import BaseCredentialService
 from .code_executors.built_in_code_executor import BuiltInCodeExecutor
+from .errors.already_exists_error import AlreadyExistsError
 from .events.event import Event
 from .events.event import EventActions
 from .flows.llm_flows import contents
@@ -931,6 +932,215 @@ def _is_transferable_across_agent_tree(self, agent_to_run: BaseAgent) -> bool:
       agent = agent.parent_agent
     return True
 
+  # Constants for debug output truncation
+  _DEBUG_ARGS_MAX_LEN = 50  # Keep arg previews short for readability
+  _DEBUG_RESPONSE_MAX_LEN = 100  # Show more of response for context
+  _DEBUG_OUTPUT_MAX_LEN = 100  # Same as response for consistency
+
+  async def run_debug(
+      self,
+      user_queries: str | list[str] | None = None,
+      *,
+      user_id: str = 'default',
+      session_name: str = 'default',
+      print_output: bool = True,
+      return_events: bool = False,
+      verbose: bool = False,
+      run_config: RunConfig | None = None,
+  ) -> list[Event] | None:
+    """Debug helper for quick agent experimentation and testing.
+
+    This convenience method is designed for developers getting started with ADK
+    who want to quickly test agents without dealing with session management,
+    content formatting, or event streaming. It automatically handles common
+    boilerplate while hiding complexity.
+
+    IMPORTANT: This is for debugging and experimentation only. For production
+    use, please use the standard run_async() method which provides full control
+    over session management, event streaming, and error handling.
+
+    Args:
+        user_queries: Question(s) to ask the agent. Can be:
+            - None: Creates/retrieves session without sending messages
+            - Single string: "What is 2+2?"
+            - List of strings: ["Hello!", "What's my name?"]
+        user_id: User identifier. Defaults to "default".
+        session_name: Session identifier for conversation persistence.
+            Defaults to "default". Reuse the same name to continue a conversation.
+        print_output: If True, prints the conversation to stdout. Defaults to True.
+        return_events: If True, returns a list of all events. Defaults to False.
+        verbose: If True, shows detailed tool calls and responses. Defaults to False
+            for cleaner output showing only final agent responses.
+        run_config: Optional configuration for the agent execution.
+
+    Returns:
+        None by default when return_events=False (just prints output).
+        list[Event] if return_events=True, containing all events from all queries.
+        Returns empty list [] if return_events=True and no queries provided.
+
+    Raises:
+        ValueError: If session creation/retrieval fails.
+
+    Examples:
+        Quick debugging:
+        >>> runner = InMemoryRunner(agent=my_agent)
+        >>> await runner.run_debug("What is 2+2?")
+
+        Multiple queries in conversation:
+        >>> await runner.run_debug(["Hello!", "What's my name?"])
+
+        Continue a debug session:
+        >>> await runner.run_debug("What did we discuss?")  # Continues default session
+
+        Separate debug sessions:
+        >>> await runner.run_debug("Hi", user_id="alice", session_name="debug1")
+        >>> await runner.run_debug("Hi", user_id="bob", session_name="debug2")
+
+        Capture events for inspection:
+        >>> events = await runner.run_debug("Analyze this", return_events=True)
+        >>> for event in events:
+        ...     inspect_event(event)
+
+    Note:
+        For production applications requiring:
+        - Custom session/memory services (Spanner, Cloud SQL, etc.)
+        - Fine-grained event processing and streaming
+        - Error recovery and resumability
+        - Performance optimization
+        Please use run_async() with proper configuration.
+    """
+    # Display session identifier if printing enabled
+    if print_output:
+      print(f'\n ### Session: {session_name}')
+
+    # Attempt to create a new session or retrieve existing one
+    try:
+      session = await self.session_service.create_session(
+          app_name=self.app_name, user_id=user_id, session_id=session_name
+      )
+      # Debug logging follows print_output for consistency - when user requests
+      # silent mode, we suppress all output including debug logs
+      if print_output:
+        logger.debug(f'Created new session: {session_name}')
+    except AlreadyExistsError:
+      # Session exists, retrieve it
+      session = await self.session_service.get_session(
+          app_name=self.app_name, user_id=user_id, session_id=session_name
+      )
+      # Debug logging follows print_output for consistency - when user requests
+      # silent mode, we suppress all output including debug logs
+      if print_output:
+        logger.debug(f'Retrieved existing session: {session_name}')
+
+    # Validate session was created/retrieved successfully
+    if not session:
+      raise ValueError(
+          f"Failed to create or retrieve session '{session_name}' "
+          f"for user '{user_id}' in app '{self.app_name}'"
+      )
+
+    # Only allocate list when needed to save memory in common case
+    collected_events = [] if return_events else None
+
+    # Process queries if provided
+    if user_queries:
+      # Normalize input to list for uniform processing
+      if isinstance(user_queries, str):
+        user_queries = [user_queries]
+
+      # Process each query sequentially
+      for query in user_queries:
+        # Display user query if printing enabled
+        if print_output:
+          print(f'\nUser > {query}')
+
+        # Convert query string to Content format required by Runner
+        content = types.Content(
+            role='user', parts=[types.Part.from_text(text=query)]
+        )
+
+        # Stream agent responses
+        async for event in self.run_async(
+            user_id=user_id,
+            session_id=session.id,
+            new_message=content,
+            run_config=run_config,
+        ):
+          # Print response if enabled and event contains content
+          if print_output and event.content and event.content.parts:
+            for part in event.content.parts:
+              # Text parts are always shown regardless of verbose setting
+              # because they contain the actual agent responses users expect
+              # Filter out None strings that some models return as placeholders
+              text = getattr(part, 'text', None)
+              if text and text != 'None':
+                print(f'{event.author} > {text}')
+              # Non-text parts (tool calls, code, etc.) are hidden by default
+              # to reduce clutter and show only what matters: the final results
+              elif verbose:
+                # Tool invocations show the behind-the-scenes processing
+                func_call = getattr(part, 'function_call', None)
+                func_resp = getattr(part, 'function_response', None)
+                executable_code = getattr(part, 'executable_code', None)
+                code_result = getattr(part, 'code_execution_result', None)
+                inline_data = getattr(part, 'inline_data', None)
+                file_data = getattr(part, 'file_data', None)
+
+                if func_call:
+                  args_str = str(func_call.args)
+                  args_preview = (
+                      args_str[: self._DEBUG_ARGS_MAX_LEN] + '...'
+                      if len(args_str) > self._DEBUG_ARGS_MAX_LEN
+                      else args_str
+                  )
+                  print(
+                      f'{event.author} > [Calling tool:'
+                      f' {func_call.name}({args_preview})]'
+                  )
+                # Handle function response parts (tool results)
+                elif func_resp:
+                  resp_str = str(func_resp.response)
+                  resp_preview = (
+                      resp_str[: self._DEBUG_RESPONSE_MAX_LEN] + '...'
+                      if len(resp_str) > self._DEBUG_RESPONSE_MAX_LEN
+                      else resp_str
+                  )
+                  print(f'{event.author} > [Tool result: {resp_preview}]')
+                # Handle executable code parts
+                elif executable_code:
+                  lang = getattr(executable_code, 'language', 'code')
+                  print(f'{event.author} > [Executing {lang} code...]')
+                # Handle code execution result parts
+                elif code_result:
+                  output = getattr(code_result, 'output', 'result')
+                  output_str = str(output)
+                  output_preview = (
+                      output_str[: self._DEBUG_OUTPUT_MAX_LEN] + '...'
+                      if len(output_str) > self._DEBUG_OUTPUT_MAX_LEN
+                      else output_str
+                  )
+                  print(f'{event.author} > [Code output: {output_preview}]')
+                # Handle inline data (images, files)
+                elif inline_data:
+                  mime_type = getattr(inline_data, 'mime_type', 'data')
+                  print(f'{event.author} > [Inline data: {mime_type}]')
+                # Handle file data
+                elif file_data:
+                  uri = getattr(file_data, 'file_uri', 'file')
+                  print(f'{event.author} > [File: {uri}]')
+
+          # Collect events if requested
+          if return_events:
+            collected_events.append(event)
+
+    else:
+      # No queries provided, just session setup
+      if print_output:
+        print('Session ready. No queries provided.')
+
+    # Return collected events or None
+    return collected_events
+
   async def _setup_context_for_new_invocation(
       self,
       *,
diff --git a/tests/unittests/runners/test_runner_debug.py b/tests/unittests/runners/test_runner_debug.py
new file mode 100644
index 0000000000..461e52c813
--- /dev/null
+++ b/tests/unittests/runners/test_runner_debug.py
@@ -0,0 +1,1018 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for Runner.run_debug helper method."""
+
+from __future__ import annotations
+
+from unittest import mock
+
+from google.adk.agents import Agent
+from google.adk.agents.run_config import RunConfig
+from google.adk.errors.already_exists_error import AlreadyExistsError
+from google.adk.runners import InMemoryRunner
+from google.adk.runners import Runner
+from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.genai import types
+import pytest
+
+
+class TestRunDebug:
+  """Tests for Runner.run_debug method."""
+
+  @pytest.mark.asyncio
+  async def test_run_debug_single_query(self):
+    """Test run_debug with a single string query."""
+    # Setup
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="You are a helpful assistant.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    # Mock the runner's run_async to return controlled events
+    mock_event = mock.Mock()
+    mock_event.author = "test_agent"
+    mock_event.content = mock.Mock()
+    mock_event.content.parts = [mock.Mock(text="Hello! I can help you.")]
+
+    async def mock_run_async(*args, **kwargs):
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute
+      events = await runner.run_debug(
+          "Hello, how are you?", print_output=False, return_events=True
+      )
+
+      # Assertions
+      assert events is not None
+      assert len(events) == 1
+      assert events[0].author == "test_agent"
+      assert events[0].content.parts[0].text == "Hello! I can help you."
+
+      # Verify session was created with defaults
+      session = await runner.session_service.get_session(
+          app_name=runner.app_name, user_id="default", session_id="default"
+      )
+      assert session is not None
+
+  @pytest.mark.asyncio
+  async def test_run_debug_multiple_queries(self):
+    """Test run_debug with multiple queries in sequence."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="You are a test bot.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    # Mock responses for multiple queries
+    responses = ["First response", "Second response"]
+    call_count = 0
+
+    async def mock_run_async(*args, **kwargs):
+      nonlocal call_count
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text=responses[call_count])]
+      call_count += 1
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with multiple queries
+      events = await runner.run_debug(
+          ["First query", "Second query"],
+          print_output=False,
+          return_events=True,
+      )
+
+      # Assertions
+      assert events is not None
+      assert len(events) == 2
+      assert events[0].content.parts[0].text == "First response"
+      assert events[1].content.parts[0].text == "Second response"
+
+  @pytest.mark.asyncio
+  async def test_run_debug_empty_query(self):
+    """Test run_debug with no queries (session setup only)."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    # Execute with no queries
+    events = await runner.run_debug(
+        None, print_output=False, return_events=True
+    )
+
+    # With no queries, returns empty list when return_events=True
+    assert events == []
+
+    # But session should still be created
+    session = await runner.session_service.get_session(
+        app_name=runner.app_name, user_id="default", session_id="default"
+    )
+    assert session is not None
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_return_events(self):
+    """Test that return_events=True returns events, False returns None."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="Response")]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Test with return_events=True
+      events = await runner.run_debug(
+          "Query", print_output=False, return_events=True
+      )
+      assert events is not None
+      assert len(events) == 1
+
+      # Test with return_events=False (default)
+      events = await runner.run_debug("Query", print_output=False)
+      assert events is None
+
+  @pytest.mark.asyncio
+  async def test_run_debug_without_print_output(self, capsys):
+    """Test that print_output=False suppresses printing."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="This should not be printed")]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with print_output=False
+      await runner.run_debug("Test query", print_output=False)
+
+      # Check that nothing was printed
+      captured = capsys.readouterr()
+      assert "This should not be printed" not in captured.out
+      assert "User >" not in captured.out
+      assert "Session:" not in captured.out
+
+  @pytest.mark.asyncio
+  async def test_run_debug_custom_session_name(self):
+    """Test run_debug with custom session_name."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="Response")]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with custom session name
+      await runner.run_debug(
+          "Query", session_name="custom_debug_session", print_output=False
+      )
+
+      # Verify session was created with custom name
+      session = await runner.session_service.get_session(
+          app_name=runner.app_name,
+          user_id="default",
+          session_id="custom_debug_session",
+      )
+      assert session is not None
+      assert session.id == "custom_debug_session"
+
+  @pytest.mark.asyncio
+  async def test_run_debug_custom_user_id(self):
+    """Test run_debug with custom user_id."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="Response")]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with custom user_id
+      await runner.run_debug(
+          "Query", user_id="test_user_123", print_output=False
+      )
+
+      # Verify session was created with custom user_id
+      session = await runner.session_service.get_session(
+          app_name=runner.app_name,
+          user_id="test_user_123",
+          session_id="default",
+      )
+      assert session is not None
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_run_config(self):
+    """Test that run_config is properly passed through to run_async."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    run_config_used = None
+
+    async def mock_run_async(*args, **kwargs):
+      nonlocal run_config_used
+      run_config_used = kwargs.get("run_config")
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="Response")]
+      yield mock_event
+
+    with mock.patch.object(
+        runner, "run_async", side_effect=mock_run_async
+    ) as mock_method:
+      # Create a custom run_config
+      custom_config = RunConfig(support_cfc=True)
+
+      # Execute with custom run_config
+      await runner.run_debug(
+          "Query", run_config=custom_config, print_output=False
+      )
+
+      # Verify run_config was passed to run_async
+      assert mock_method.called
+      call_args = mock_method.call_args
+      assert call_args is not None
+      assert "run_config" in call_args.kwargs
+      assert call_args.kwargs["run_config"] == custom_config
+
+  @pytest.mark.asyncio
+  async def test_run_debug_session_persistence(self):
+    """Test that multiple calls to run_debug maintain conversation context."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Remember previous messages.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    call_count = 0
+    responses = ["First response", "Second response remembering first"]
+
+    async def mock_run_async(*args, **kwargs):
+      nonlocal call_count
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text=responses[call_count])]
+      call_count += 1
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # First call
+      events1 = await runner.run_debug(
+          "First message", print_output=False, return_events=True
+      )
+      assert events1[0].content.parts[0].text == "First response"
+
+      # Second call to same session
+      events2 = await runner.run_debug(
+          "Second message", print_output=False, return_events=True
+      )
+      assert (
+          events2[0].content.parts[0].text
+          == "Second response remembering first"
+      )
+
+      # Verify both calls used the same session
+      session = await runner.session_service.get_session(
+          app_name=runner.app_name, user_id="default", session_id="default"
+      )
+      assert session is not None
+
+  @pytest.mark.asyncio
+  async def test_run_debug_error_handling(self):
+    """Test error handling when session creation fails."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+
+    # Create a mock session service that fails then returns None
+    mock_session_service = mock.AsyncMock(spec=InMemorySessionService)
+    mock_session_service.create_session.side_effect = AlreadyExistsError(
+        "Already exists"
+    )
+    mock_session_service.get_session.return_value = None
+
+    # Create runner with mock session service
+    runner = Runner(
+        agent=agent,
+        app_name="test_app",
+        session_service=mock_session_service,
+    )
+
+    # Should raise ValueError when session can't be created or retrieved
+    with pytest.raises(ValueError) as exc_info:
+      await runner.run_debug("Test query", print_output=False)
+
+    assert "Failed to create or retrieve session" in str(exc_info.value)
+
+  @pytest.mark.asyncio
+  async def test_run_debug_filters_none_text(self):
+    """Test that run_debug filters out 'None' text and empty parts."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Yield events with various text values
+      events = [
+          mock.Mock(
+              author="test_agent",
+              content=mock.Mock(parts=[mock.Mock(text="Valid text")]),
+          ),
+          mock.Mock(
+              author="test_agent",
+              content=mock.Mock(parts=[mock.Mock(text="None")]),
+          ),  # Should be filtered
+          mock.Mock(
+              author="test_agent",
+              content=mock.Mock(parts=[mock.Mock(text="")]),
+          ),  # Should be filtered
+          mock.Mock(
+              author="test_agent",
+              content=mock.Mock(parts=[mock.Mock(text="Another valid")]),
+          ),
+      ]
+      for event in events:
+        yield event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute and capture output
+      events = await runner.run_debug(
+          "Query", print_output=False, return_events=True
+      )
+
+      # All 4 events should be returned (filtering is for printing only)
+      assert len(events) == 4
+
+      # But when printing, "None" and empty strings should be filtered
+      # This is tested implicitly by the implementation
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_existing_session(self):
+    """Test that run_debug retrieves existing session when AlreadyExistsError occurs."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    # First create a session
+    await runner.session_service.create_session(
+        app_name=runner.app_name,
+        user_id="default",
+        session_id="existing_session",
+    )
+
+    async def mock_run_async(*args, **kwargs):
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = [mock.Mock(text="Using existing session")]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with same session name (should retrieve existing)
+      events = await runner.run_debug(
+          "Query",
+          session_name="existing_session",
+          print_output=False,
+          return_events=True,
+      )
+
+      assert events is not None
+      assert len(events) == 1
+      assert events[0].content.parts[0].text == "Using existing session"
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_tool_calls(self, capsys):
+    """Test that run_debug properly handles and prints tool calls."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with tools.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # First event: tool call
+      mock_call_event = mock.Mock()
+      mock_call_event.author = "test_agent"
+      mock_call_event.content = mock.Mock()
+      mock_function_call = mock.Mock()
+      mock_function_call.name = "calculate"
+      mock_function_call.args = {"operation": "add", "a": 5, "b": 3}
+      mock_part_call = mock.Mock()
+      mock_part_call.text = None
+      mock_part_call.function_call = mock_function_call
+      mock_part_call.function_response = None
+      mock_call_event.content.parts = [mock_part_call]
+      yield mock_call_event
+
+      # Second event: tool response
+      mock_resp_event = mock.Mock()
+      mock_resp_event.author = "test_agent"
+      mock_resp_event.content = mock.Mock()
+      mock_function_response = mock.Mock()
+      mock_function_response.response = {"result": 8}
+      mock_part_resp = mock.Mock()
+      mock_part_resp.text = None
+      mock_part_resp.function_call = None
+      mock_part_resp.function_response = mock_function_response
+      mock_resp_event.content.parts = [mock_part_resp]
+      yield mock_resp_event
+
+      # Third event: final text response
+      mock_text_event = mock.Mock()
+      mock_text_event.author = "test_agent"
+      mock_text_event.content = mock.Mock()
+      mock_text_event.content.parts = [mock.Mock(text="The result is 8")]
+      yield mock_text_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      # Execute with print_output=True and verbose=True to see tool calls
+      events = await runner.run_debug(
+          "Calculate 5 + 3", print_output=True, return_events=True, verbose=True
+      )
+
+      # Check output was printed
+      captured = capsys.readouterr()
+      assert "[Calling tool: calculate" in captured.out
+      assert "[Tool result:" in captured.out
+      assert "The result is 8" in captured.out
+
+      # Check events were collected
+      assert len(events) == 3
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_executable_code(self, capsys):
+    """Test that run_debug properly handles executable code parts."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with code execution.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Event with executable code
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+
+      mock_exec_code = mock.Mock()
+      mock_exec_code.language = "python"
+      mock_exec_code.code = "print('Hello World')"
+
+      mock_part = mock.Mock()
+      mock_part.text = None
+      mock_part.function_call = None
+      mock_part.function_response = None
+      mock_part.executable_code = mock_exec_code
+      mock_part.code_execution_result = None
+      mock_part.inline_data = None
+      mock_part.file_data = None
+
+      mock_event.content.parts = [mock_part]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Run some code", print_output=True, return_events=True, verbose=True
+      )
+
+      captured = capsys.readouterr()
+      assert "[Executing python code...]" in captured.out
+      assert len(events) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_code_execution_result(self, capsys):
+    """Test that run_debug properly handles code execution result parts."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with code results.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Event with code execution result
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+
+      mock_result = mock.Mock()
+      mock_result.output = "Hello World\n42"
+
+      mock_part = mock.Mock()
+      mock_part.text = None
+      mock_part.function_call = None
+      mock_part.function_response = None
+      mock_part.executable_code = None
+      mock_part.code_execution_result = mock_result
+      mock_part.inline_data = None
+      mock_part.file_data = None
+
+      mock_event.content.parts = [mock_part]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Show code output",
+          print_output=True,
+          return_events=True,
+          verbose=True,
+      )
+
+      captured = capsys.readouterr()
+      assert "[Code output: Hello World\n42]" in captured.out
+      assert len(events) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_inline_data(self, capsys):
+    """Test that run_debug properly handles inline data parts."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with inline data.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Event with inline data (e.g., image)
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+
+      mock_inline = mock.Mock()
+      mock_inline.mime_type = "image/png"
+      mock_inline.data = b"fake_image_data"
+
+      mock_part = mock.Mock()
+      mock_part.text = None
+      mock_part.function_call = None
+      mock_part.function_response = None
+      mock_part.executable_code = None
+      mock_part.code_execution_result = None
+      mock_part.inline_data = mock_inline
+      mock_part.file_data = None
+
+      mock_event.content.parts = [mock_part]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Show image", print_output=True, return_events=True, verbose=True
+      )
+
+      captured = capsys.readouterr()
+      assert "[Inline data: image/png]" in captured.out
+      assert len(events) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_file_data(self, capsys):
+    """Test that run_debug properly handles file data parts."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with file data.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Event with file data
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+
+      mock_file = mock.Mock()
+      mock_file.file_uri = "gs://bucket/path/to/file.pdf"
+
+      mock_part = mock.Mock()
+      mock_part.text = None
+      mock_part.function_call = None
+      mock_part.function_response = None
+      mock_part.executable_code = None
+      mock_part.code_execution_result = None
+      mock_part.inline_data = None
+      mock_part.file_data = mock_file
+
+      mock_event.content.parts = [mock_part]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Reference file", print_output=True, return_events=True, verbose=True
+      )
+
+      captured = capsys.readouterr()
+      assert "[File: gs://bucket/path/to/file.pdf]" in captured.out
+      assert len(events) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_mixed_parts(self, capsys):
+    """Test that run_debug handles events with multiple part types."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with mixed parts.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Event with multiple part types
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+
+      # Text part
+      mock_text_part = mock.Mock()
+      mock_text_part.text = "Here's your result:"
+      mock_text_part.function_call = None
+      mock_text_part.function_response = None
+      mock_text_part.executable_code = None
+      mock_text_part.code_execution_result = None
+      mock_text_part.inline_data = None
+      mock_text_part.file_data = None
+
+      # Code execution part
+      mock_code_part = mock.Mock()
+      mock_code_part.text = None
+      mock_code_part.function_call = None
+      mock_code_part.function_response = None
+      mock_exec_code = mock.Mock()
+      mock_exec_code.language = "python"
+      mock_code_part.executable_code = mock_exec_code
+      mock_code_part.code_execution_result = None
+      mock_code_part.inline_data = None
+      mock_code_part.file_data = None
+
+      # Result part
+      mock_result_part = mock.Mock()
+      mock_result_part.text = None
+      mock_result_part.function_call = None
+      mock_result_part.function_response = None
+      mock_result_part.executable_code = None
+      mock_result = mock.Mock()
+      mock_result.output = "42"
+      mock_result_part.code_execution_result = mock_result
+      mock_result_part.inline_data = None
+      mock_result_part.file_data = None
+
+      mock_event.content.parts = [
+          mock_text_part,
+          mock_code_part,
+          mock_result_part,
+      ]
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Mixed response", print_output=True, return_events=True, verbose=True
+      )
+
+      captured = capsys.readouterr()
+      assert "Here's your result:" in captured.out
+      assert "[Executing python code...]" in captured.out
+      assert "[Code output: 42]" in captured.out
+      assert len(events) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_long_output_truncation(self, capsys):
+    """Test that run_debug properly truncates long outputs."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with long outputs.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Tool call with long args
+      mock_call_event = mock.Mock()
+      mock_call_event.author = "test_agent"
+      mock_call_event.content = mock.Mock()
+
+      mock_function_call = mock.Mock()
+      mock_function_call.name = "process"
+      # Create a long argument string
+      mock_function_call.args = {"data": "x" * 100}
+
+      mock_part_call = mock.Mock()
+      mock_part_call.text = None
+      mock_part_call.function_call = mock_function_call
+      mock_part_call.function_response = None
+      mock_part_call.executable_code = None
+      mock_part_call.code_execution_result = None
+      mock_part_call.inline_data = None
+      mock_part_call.file_data = None
+
+      mock_call_event.content.parts = [mock_part_call]
+      yield mock_call_event
+
+      # Tool response with long result
+      mock_resp_event = mock.Mock()
+      mock_resp_event.author = "test_agent"
+      mock_resp_event.content = mock.Mock()
+
+      mock_function_response = mock.Mock()
+      # Create a long response string
+      mock_function_response.response = {"result": "y" * 200}
+
+      mock_part_resp = mock.Mock()
+      mock_part_resp.text = None
+      mock_part_resp.function_call = None
+      mock_part_resp.function_response = mock_function_response
+      mock_part_resp.executable_code = None
+      mock_part_resp.code_execution_result = None
+      mock_part_resp.inline_data = None
+      mock_part_resp.file_data = None
+
+      mock_resp_event.content.parts = [mock_part_resp]
+      yield mock_resp_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Process data", print_output=True, return_events=True, verbose=True
+      )
+
+      captured = capsys.readouterr()
+      # Check that args are truncated at 50 chars
+      assert "..." in captured.out
+      assert "[Calling tool: process(" in captured.out
+      # Check that response is truncated at 100 chars
+      assert "[Tool result:" in captured.out
+      assert len(events) == 2
+
+  @pytest.mark.asyncio
+  async def test_run_debug_verbose_flag_false(self, capsys):
+    """Test that run_debug hides tool calls when verbose=False (default)."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with tools.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Tool call event
+      mock_call_event = mock.Mock()
+      mock_call_event.author = "test_agent"
+      mock_call_event.content = mock.Mock()
+
+      mock_function_call = mock.Mock()
+      mock_function_call.name = "get_weather"
+      mock_function_call.args = {"city": "Tokyo"}
+
+      mock_part_call = mock.Mock()
+      mock_part_call.text = None
+      mock_part_call.function_call = mock_function_call
+      mock_part_call.function_response = None
+      mock_part_call.executable_code = None
+      mock_part_call.code_execution_result = None
+      mock_part_call.inline_data = None
+      mock_part_call.file_data = None
+
+      mock_call_event.content.parts = [mock_part_call]
+      yield mock_call_event
+
+      # Tool response event
+      mock_resp_event = mock.Mock()
+      mock_resp_event.author = "test_agent"
+      mock_resp_event.content = mock.Mock()
+
+      mock_function_response = mock.Mock()
+      mock_function_response.response = {"weather": "Clear, 25°C"}
+
+      mock_part_resp = mock.Mock()
+      mock_part_resp.text = None
+      mock_part_resp.function_call = None
+      mock_part_resp.function_response = mock_function_response
+      mock_part_resp.executable_code = None
+      mock_part_resp.code_execution_result = None
+      mock_part_resp.inline_data = None
+      mock_part_resp.file_data = None
+
+      mock_resp_event.content.parts = [mock_part_resp]
+      yield mock_resp_event
+
+      # Final text response
+      mock_text_event = mock.Mock()
+      mock_text_event.author = "test_agent"
+      mock_text_event.content = mock.Mock()
+      mock_text_event.content.parts = [
+          mock.Mock(text="The weather in Tokyo is clear and 25°C.")
+      ]
+      yield mock_text_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "What's the weather?",
+          print_output=True,
+          return_events=True,
+          verbose=False,  # Default - should NOT show tool calls
+      )
+
+      captured = capsys.readouterr()
+      # Should NOT show tool call details
+      assert "[Calling tool:" not in captured.out
+      assert "[Tool result:" not in captured.out
+      # Should show final text response
+      assert "The weather in Tokyo is clear and 25°C." in captured.out
+      assert len(events) == 3
+
+  @pytest.mark.asyncio
+  async def test_run_debug_verbose_flag_true(self, capsys):
+    """Test that run_debug shows tool calls when verbose=True."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent with tools.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*args, **kwargs):
+      # Tool call event
+      mock_call_event = mock.Mock()
+      mock_call_event.author = "test_agent"
+      mock_call_event.content = mock.Mock()
+
+      mock_function_call = mock.Mock()
+      mock_function_call.name = "calculate"
+      mock_function_call.args = {"expression": "42 * 3.14"}
+
+      mock_part_call = mock.Mock()
+      mock_part_call.text = None
+      mock_part_call.function_call = mock_function_call
+      mock_part_call.function_response = None
+      mock_part_call.executable_code = None
+      mock_part_call.code_execution_result = None
+      mock_part_call.inline_data = None
+      mock_part_call.file_data = None
+
+      mock_call_event.content.parts = [mock_part_call]
+      yield mock_call_event
+
+      # Tool response event
+      mock_resp_event = mock.Mock()
+      mock_resp_event.author = "test_agent"
+      mock_resp_event.content = mock.Mock()
+
+      mock_function_response = mock.Mock()
+      mock_function_response.response = {"result": 131.88}
+
+      mock_part_resp = mock.Mock()
+      mock_part_resp.text = None
+      mock_part_resp.function_call = None
+      mock_part_resp.function_response = mock_function_response
+      mock_part_resp.executable_code = None
+      mock_part_resp.code_execution_result = None
+      mock_part_resp.inline_data = None
+      mock_part_resp.file_data = None
+
+      mock_resp_event.content.parts = [mock_part_resp]
+      yield mock_resp_event
+
+      # Final text response
+      mock_text_event = mock.Mock()
+      mock_text_event.author = "test_agent"
+      mock_text_event.content = mock.Mock()
+      mock_text_event.content.parts = [mock.Mock(text="The result is 131.88")]
+      yield mock_text_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Calculate 42 * 3.14",
+          print_output=True,
+          return_events=True,
+          verbose=True,  # Should show tool calls
+      )
+
+      captured = capsys.readouterr()
+      # Should show tool call details
+      assert (
+          "[Calling tool: calculate({'expression': '42 * 3.14'})]"
+          in captured.out
+      )
+      assert "[Tool result: {'result': 131.88}]" in captured.out
+      # Should also show final text response
+      assert "The result is 131.88" in captured.out
+      assert len(events) == 3
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_empty_parts_list(self, capsys):
+    """Test that run_debug handles events with empty parts list gracefully."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*_args, **_kwargs):
+      # Event with empty parts list
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = mock.Mock()
+      mock_event.content.parts = []  # Empty parts list
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Test query", print_output=True, return_events=True
+      )
+
+      captured = capsys.readouterr()
+      # Should handle gracefully without crashing
+      assert "User > Test query" in captured.out
+      assert len(events) == 1
+      # Should not print any agent response since parts is empty
+      assert "test_agent >" not in captured.out
+
+  @pytest.mark.asyncio
+  async def test_run_debug_with_none_event_content(self, capsys):
+    """Test that run_debug handles events with None content gracefully."""
+    agent = Agent(
+        name="test_agent",
+        model="gemini-2.5-flash-lite",
+        instruction="Test agent.",
+    )
+    runner = InMemoryRunner(agent=agent)
+
+    async def mock_run_async(*_args, **_kwargs):
+      # Event with None content
+      mock_event = mock.Mock()
+      mock_event.author = "test_agent"
+      mock_event.content = None  # None content
+      yield mock_event
+
+    with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
+      events = await runner.run_debug(
+          "Test query", print_output=True, return_events=True
+      )
+
+      captured = capsys.readouterr()
+      # Should handle gracefully without crashing
+      assert "User > Test query" in captured.out
+      assert len(events) == 1
+      # Should not print any agent response since content is None
+      assert "test_agent >" not in captured.out