diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index d8c15b288d..91b8473925 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -358,6 +358,40 @@ func main() { }) log.Info("Historical logs provider configured for task subscriptions (using messages and git status)") + // Set up session data provider for session subscriptions + // Sends initial git status when a client subscribes to a session + gateway.Hub.SetSessionDataProvider(func(ctx context.Context, sessionID string) ([]*ws.Message, error) { + session, err := taskRepo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, nil // Session not found, no data to send + } + + var result []*ws.Message + + // Send git status if available in session metadata + if session.Metadata != nil { + if gitStatus, ok := session.Metadata["git_status"]; ok { + gitStatusData, ok := gitStatus.(map[string]interface{}) + if !ok { + gitStatusData = map[string]interface{}{"data": gitStatus} + } + // Ensure session_id is included + gitStatusData["session_id"] = sessionID + if session.TaskID != "" { + gitStatusData["task_id"] = session.TaskID + } + + gitStatusNotification, err := ws.NewNotification("session.git.status", gitStatusData) + if err == nil { + result = append(result, gitStatusNotification) + } + } + } + + return result, nil + }) + log.Info("Session data provider configured for session subscriptions (git status)") + // NOTE: Pending permissions are now stored as messages and retrieved via the historical logs provider. // The separate pending permissions provider is no longer needed. diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index 58fd02c5db..60f4c92a54 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -223,6 +223,9 @@ func (c *Client) handleSessionSubscribe(msg *ws.Message) { "session_id": req.SessionID, }) c.sendMessage(resp) + + // Send initial session data (e.g., git status) if available + c.sendSessionData(req.SessionID) } func (c *Client) handleUserUnsubscribe(msg *ws.Message) { @@ -272,6 +275,31 @@ func (c *Client) sendHistoricalLogs(taskID string) { } } +// sendSessionData sends initial session data (e.g., git status) to the client +func (c *Client) sendSessionData(sessionID string) { + ctx := context.Background() + data, err := c.hub.GetSessionData(ctx, sessionID) + if err != nil { + c.logger.Error("Failed to get session data", + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + + if len(data) == 0 { + return + } + + c.logger.Debug("Sending session data", + zap.String("session_id", sessionID), + zap.Int("count", len(data))) + + // Send each piece of session data as a notification + for _, msg := range data { + c.sendMessage(msg) + } +} + // handleUnsubscribe handles task.unsubscribe action func (c *Client) handleUnsubscribe(msg *ws.Message) { var req SubscribeRequest diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index 2f7c52cc91..c05c0548e3 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -14,6 +14,9 @@ import ( // HistoricalLogsProvider is a function that retrieves historical logs for a task type HistoricalLogsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error) +// SessionDataProvider is a function that retrieves initial data for a session subscription (e.g., git status) +type SessionDataProvider func(ctx context.Context, sessionID string) ([]*ws.Message, error) + // Hub manages all WebSocket client connections type Hub struct { // All registered clients @@ -39,6 +42,9 @@ type Hub struct { // Optional provider for historical logs on subscription historicalLogsProvider HistoricalLogsProvider + // Optional provider for session data on subscription (e.g., git status) + sessionDataProvider SessionDataProvider + mu sync.RWMutex logger *logger.Logger } @@ -385,4 +391,17 @@ func (h *Hub) GetHistoricalLogs(ctx context.Context, taskID string) ([]*ws.Messa return nil, nil } return h.historicalLogsProvider(ctx, taskID) +} + +// SetSessionDataProvider sets the provider for session data on subscription +func (h *Hub) SetSessionDataProvider(provider SessionDataProvider) { + h.sessionDataProvider = provider +} + +// GetSessionData retrieves session data (e.g., git status) if a provider is set +func (h *Hub) GetSessionData(ctx context.Context, sessionID string) ([]*ws.Message, error) { + if h.sessionDataProvider == nil { + return nil, nil + } + return h.sessionDataProvider(ctx, sessionID) } \ No newline at end of file