How to Debug Event Flow Using log_tree() in bubus

Call bus.log_tree() on any EventBus instance to generate a formatted string showing the complete event-handler hierarchy, timing data, and error states.

The bubus library provides structured event-driven architecture through its EventBus implementation. When applications grow complex, tracing how events trigger handlers and spawn child events becomes critical for debugging. The log_tree() method offers a read-only, pretty-printed visualization of the entire event flow directly from the bus instance.

Understanding the log_tree() Architecture

The tree visualization is generated through a chain of tightly-coupled functions across three core files.

Entry Point in service.py

The public API resides in bubus/service.py at lines 1433-1437. The EventBus class exposes log_tree() as an instance method:

def log_tree(self) -> str:
    """Print a nicely formatted tree view of all events in the history."""
    from bubus.logging import log_eventbus_tree
    return log_eventbus_tree(self)

This method is strictly read-only and returns the formatted string without mutating the bus state.

Tree Construction in logging.py

The core logic lives in bubus/logging.py between lines 166-208 inside the log_eventbus_tree() function. This helper performs four critical steps:

1. Maps parent-child relationships using a defaultdict keyed by event_parent_id
2. Sorts children by creation time to ensure deterministic output
3. Identifies root events where event_parent_id is None or self-referencing
4. Recursively renders each root via log_event_tree()

Event Node Rendering

Individual event nodes are formatted through log_event_tree(), which handles:

• Event lines: Displaying the event type, ID suffix, and status icon (📣)
• Handler results: Iterating event.event_results to show completion status (✅), errors (❌), timeouts (⏳), or pending states (🔜)
• Timing columns: Rendering duration as (⏳ 2s/5s) showing elapsed versus timeout limits
• Child recursion: Walking nested events with proper ASCII tree branches (├──, └──)

Each BaseEvent instance delegates to this renderer via event_log_tree() defined in bubus/models.py at lines 759-669.

Practical Usage Examples

Basic Debugging Workflow

The simplest way to inspect event flow is calling log_tree() after dispatch:

from bubus import EventBus, BaseEvent

class MyEvent(BaseEvent[int]):
    value: int = 42

bus = EventBus(name="DemoBus")
event = MyEvent()
bus.dispatch(event)

# Generate the visualization

print(bus.log_tree())

This outputs a hierarchical view showing the root MyEvent, any registered handlers, and child events spawned during processing.

Unit Test Integration

The repository includes comprehensive tests in tests/test_log_history_tree.py that demonstrate validation patterns. You can assert on specific tree fragments to verify handler execution:

def test_event_flow(capsys):
    bus = EventBus(name="TestBus")
    # ... dispatch events and trigger handlers ...

    tree = bus.log_tree()
    
    assert "✅ TestBus.my_handler#" in tree
    assert "RootEvent#" in tree
    assert "❌" not in tree  # Verify no errors occurred

This approach catches regressions in event propagation without parsing complex internal state.

Production Logging

Since log_tree() returns a plain string, you can integrate it with standard logging frameworks:

import logging
logger = logging.getLogger("myapp")

# After critical operations

logger.info("Current event flow:\n%s", bus.log_tree())

The internal bubus logger also emits a warning header when generating trees, ensuring visibility in standard output during debugging sessions.

Interpreting the Output Format

The tree uses visual conventions to convey execution state:

• 📣 – Event emission line showing the event type and unique ID suffix
• ✅ – Handler completed successfully with timing (⏳ elapsed/total)
• ❌ – Handler failed with exception details
• ⏳ – Handler currently running or timed out
• 🔜 – Handler pending execution
• Indentation – Child events spawned by handlers appear as indented branches using ├── and └── characters

Timing information displays as (⏳ 2s/5s) where the first number is elapsed seconds and the second is the timeout limit.

Summary

• log_tree() provides a read-only, string-based visualization of all events processed by an EventBus instance.
• The implementation spans bubus/service.py (entry point), bubus/logging.py (tree construction), and bubus/models.py (event delegation).
• The output displays root events, handler results with timing data, error states, and child event hierarchies using ASCII tree characters.
• Use print(bus.log_tree()) for debugging, assert on tree fragments in unit tests, or pass the result to standard logging frameworks for production monitoring.

Frequently Asked Questions

What does log_tree() return?

log_tree() returns a formatted multiline string containing the ASCII tree representation of the event history. It does not print to stdout directly, allowing you to log the output, write it to files, or assert against it in tests.

Can I use log_tree() while events are still processing?

Yes, log_tree() is read-only and safe to call at any time. However, the output represents a snapshot of the current state; handlers still running will display with the ⏳ icon, and events not yet processed will not appear until they enter the bus history.

How do I interpret timeout indicators in the tree output?

Timeout indicators appear as (⏳ 2s/5s) next to handler results, where the first number shows elapsed seconds and the second shows the configured timeout limit. A ❌ icon combined with this timing indicates the handler exceeded its timeout limit.

Where can I find examples of log_tree() usage in the codebase?

The repository includes working examples in tests/test_log_history_tree.py, which demonstrates asserting on tree fragments, and in README.md around line 260, which shows basic printing usage. The source implementation resides in bubus/service.py (lines 1433-1437), bubus/logging.py (lines 166-208), and bubus/models.py (lines 759-669).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}