From e99aa075788fe0067e632fee88d6a002c20ab3be Mon Sep 17 00:00:00 2001 From: jcfs Date: Sat, 10 Jan 2026 17:15:48 +0000 Subject: [PATCH] fix: publish session notifications to event bus for WebSocket streaming - Add publishSessionNotification() to lifecycle manager - Convert agentctl SessionNotification to event bus message - Enables real-time streaming of agent output to WebSocket clients --- .../internal/agent/lifecycle/manager.go | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index a27408b8a5..394df7d5db 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -414,6 +414,36 @@ func (m *Manager) handleSessionNotification(instance *AgentInstance, notificatio zap.String("instance_id", instance.ID), zap.Int("num_entries", len(update.Plan.Entries))) } + + // Publish session notification to event bus for WebSocket streaming + m.publishSessionNotification(instance, notification) +} + +// publishSessionNotification publishes a session notification to the event bus +func (m *Manager) publishSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) { + if m.eventBus == nil { + return + } + + // Build ACP message data from the session notification + data := map[string]interface{}{ + "type": "session/update", + "timestamp": time.Now(), + "agent_id": instance.ID, + "task_id": instance.TaskID, + "session_id": string(notification.SessionId), + "data": notification.Update, + } + + event := bus.NewEvent(events.ACPMessage, "agent-manager", data) + subject := events.BuildACPSubject(instance.TaskID) + + if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { + m.logger.Error("failed to publish session notification", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } } // updateInstanceProgress updates an instance's progress