Observer System — Overview

The Observer System in UFO Galaxy implements an event-driven architecture that enables real-time monitoring, visualization, and coordination of constellation execution. It provides a decoupled, extensible mechanism for components to react to system events without tight coupling.

The system implements the classic Observer Pattern (also known as Publish-Subscribe), enabling loose coupling between event producers and consumers. This allows the system to be extended with new observers without modifying existing code.

🎯 Purpose and Design Goals

The observer system serves several critical functions in the Galaxy framework:

1. Real-time Monitoring — Track task execution, constellation lifecycle, and system events
2. Visualization — Provide live updates for DAG topology and execution progress
3. Metrics Collection — Gather performance statistics and execution data
4. Synchronization — Coordinate between agent modifications and orchestrator execution
5. Agent Output Handling — Display agent responses and actions in real-time

🏗️ Architecture Overview

The observer system consists of three main layers:

Architecture Layers:

📊 Core Components

Event System Core

The foundation of the observer system consists of:

For detailed documentation of the event system core components, see the Event System Core documentation.

Observer Implementations

Five specialized observers handle different aspects of system monitoring:

🔄 Event Flow

The following diagram illustrates how events flow through the system:

The event flow demonstrates how a single action (task completion) triggers multiple observers, each performing their specialized function without interfering with others.

📋 Event Types

The system defines the following event types:

Task Events

Track individual task execution lifecycle:

Constellation Events

Monitor constellation-level operations:

Agent Events

Display agent interactions:

Device Events

Monitor device status (used by client):

🚀 Usage Example

Here's a complete example showing how observers are initialized and used in a Galaxy session:

from galaxy.core.events import get_event_bus, EventType
from galaxy.session.observers import (
    ConstellationProgressObserver,
    SessionMetricsObserver,
    DAGVisualizationObserver,
    ConstellationModificationSynchronizer,
    AgentOutputObserver
)

# Get the global event bus
event_bus = get_event_bus()

# 1. Create progress observer for agent coordination
progress_observer = ConstellationProgressObserver(agent=constellation_agent)
event_bus.subscribe(progress_observer)

# 2. Create metrics observer for performance tracking
metrics_observer = SessionMetricsObserver(
    session_id="my_session",
    logger=logger
)
event_bus.subscribe(metrics_observer)

# 3. Create visualization observer for real-time display
viz_observer = DAGVisualizationObserver(enable_visualization=True)
event_bus.subscribe(viz_observer)

# 4. Create synchronizer to prevent race conditions
synchronizer = ConstellationModificationSynchronizer(
    orchestrator=orchestrator,
    logger=logger
)
event_bus.subscribe(synchronizer)

# 5. Create agent output observer for displaying interactions
agent_output_observer = AgentOutputObserver(presenter_type="rich")
event_bus.subscribe(agent_output_observer)

# Execute constellation
await orchestrator.execute_constellation(constellation)

# Retrieve collected metrics
metrics = metrics_observer.get_metrics()
print(f"Tasks completed: {metrics['completed_tasks']}")
print(f"Total execution time: {metrics['total_execution_time']:.2f}s")
print(f"Modifications: {metrics['constellation_modifications']}")

🔑 Key Benefits

1. Decoupling

Events decouple components — publishers don't need to know about observers:

• Orchestrator publishes task events without knowing who's listening
• Agent publishes modification events without coordinating with orchestrator
• New observers can be added without changing existing code

2. Extensibility

Add custom observers for new functionality:

from galaxy.core.events import IEventObserver, Event, EventType

class CustomMetricsObserver(IEventObserver):
    """Custom observer for domain-specific metrics."""

    def __init__(self):
        self.custom_metrics = {}

    async def on_event(self, event: Event) -> None:
        if event.event_type == EventType.TASK_COMPLETED:
            # Collect custom metrics
            task_type = event.data.get("task_type")
            if task_type not in self.custom_metrics:
                self.custom_metrics[task_type] = []

            self.custom_metrics[task_type].append({
                "duration": event.data.get("execution_time"),
                "result": event.result
            })

# Subscribe to specific events
event_bus = get_event_bus()
custom_observer = CustomMetricsObserver()
event_bus.subscribe(custom_observer, {EventType.TASK_COMPLETED})

3. Concurrent Execution

All observers are notified concurrently using asyncio.gather():

• No observer blocks another
• Exceptions in one observer don't affect others
• Efficient parallel processing

4. Type-Safe Event Handling

Specialized event classes provide type safety:

async def on_event(self, event: Event) -> None:
    if isinstance(event, TaskEvent):
        # TaskEvent-specific handling
        task_id = event.task_id  # Type-safe access
        status = event.status

    elif isinstance(event, ConstellationEvent):
        # ConstellationEvent-specific handling
        constellation_id = event.constellation_id
        state = event.constellation_state

📚 Component Documentation

Explore detailed documentation for each observer:

• Session Metrics Observer — Performance metrics and statistics collection
• Event System Core — Event bus, event types, and interfaces

🔗 Related Documentation

• Constellation Orchestrator — Event publishers for task execution
• Constellation Agent — Event publishers for agent operations
• Performance Metrics — How metrics are collected and analyzed
• Event-Driven Coordination — Deep dive into event system architecture

💡 Best Practices

Observer Lifecycle Management

Properly manage observer subscriptions to prevent memory leaks:

# Subscribe observers
observers = [progress_observer, metrics_observer, viz_observer]
for observer in observers:
    event_bus.subscribe(observer)

try:
    # Execute constellation
    await orchestrator.execute_constellation(constellation)
finally:
    # Clean up observers
    for observer in observers:
        event_bus.unsubscribe(observer)

Event-Specific Subscription

Subscribe only to relevant events for efficiency:

# Instead of subscribing to all events
event_bus.subscribe(observer)  # Receives ALL events

# Subscribe to specific event types
event_bus.subscribe(observer, {
    EventType.TASK_COMPLETED,
    EventType.TASK_FAILED,
    EventType.CONSTELLATION_MODIFIED
})

Error Handling in Observers

Always handle exceptions gracefully:

async def on_event(self, event: Event) -> None:
    try:
        # Process event
        await self._handle_event(event)
    except Exception as e:
        self.logger.error(f"Error processing event: {e}")
        # Don't re-raise - let other observers continue

🎓 Summary

The Observer System provides a robust, event-driven foundation for monitoring and coordinating Galaxy's constellation execution:

• Event Bus acts as central message broker
• 5 specialized observers handle different aspects of monitoring
• Loose coupling enables extensibility and maintainability
• Concurrent execution ensures efficient event processing
• Type-safe events provide clear contracts and error prevention

For implementation details of specific observers, refer to the individual component documentation pages linked above.