tui: introduce subscription package for external event sources

Add a subscription package that provides patterns for converting
external event sources (channels, etc.) into the Bubble Tea message
flow, following the Elm Architecture subscription pattern.

Features:
- FromChannel: Convert channel reads to tea.Cmd/tea.Msg
- FromChannelWithClose: With optional close handler
- ChannelSubscription: Reusable wrapper with Listen() method

Updated the theme file watcher to use ChannelSubscription, making
the external event source explicit and demonstrating the pattern.

In Elm, subscriptions are declared at the top level and the runtime
manages them. This package provides similar patterns for Go/Bubble Tea,
making external event handling explicit and consistent.

Assisted-By: cagent

Author: dgageot

pkg/tui/subscription/subscription.go

@@ -0,0 +1,95 @@
+// Package subscription provides patterns for external event sources in the TUI.
+//
+// In the Elm Architecture, subscriptions declare interest in external events
+// (time, WebSocket messages, etc.). This package provides similar patterns
+// for Go/Bubble Tea, making external event sources explicit and manageable.
+//
+// # Pattern Overview
+//
+// A subscription converts an external event source (channel, timer, etc.)
+// into a tea.Cmd that returns tea.Msg values. The TUI's Update function
+// then processes these messages like any other.
+//
+// # Example Usage
+//
+//	// In your model
+//	type model struct {
+//	    eventCh chan ExternalEvent
+//	}
+//
+//	// Create a listener that converts channel events to messages
+//	func (m *model) listenForEvents() tea.Cmd {
+//	    return subscription.FromChannel(m.eventCh, func(e ExternalEvent) tea.Msg {
+//	        return MyEventMsg{Event: e}
+//	    })
+//	}
+//
+//	// In Init, start listening
+//	func (m *model) Init() tea.Cmd {
+//	    return m.listenForEvents()
+//	}
+//
+//	// In Update, handle the message and re-subscribe
+//	func (m *model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
+//	    switch msg := msg.(type) {
+//	    case MyEventMsg:
+//	        // Handle the event
+//	        // Re-subscribe to continue listening
+//	        return m, m.listenForEvents()
+//	    }
+//	}
+package subscription
+
+import tea "charm.land/bubbletea/v2"
+
+// FromChannel creates a tea.Cmd that waits for a value from the channel
+// and converts it to a tea.Msg using the provided function.
+//
+// The returned Cmd blocks until a value is received or the channel is closed.
+// When the channel is closed, it returns nil (no message).
+//
+// To continue listening after receiving a message, call this function again
+// in your Update handler (re-subscription pattern).
+func FromChannel[T any](ch <-chan T, toMsg func(T) tea.Msg) tea.Cmd {
+	return func() tea.Msg {
+		val, ok := <-ch
+		if !ok {
+			return nil // Channel closed
+		}
+		return toMsg(val)
+	}
+}
+
+// FromChannelWithClose is like FromChannel but also calls onClose when the
+// channel is closed, allowing cleanup or final messages.
+func FromChannelWithClose[T any](ch <-chan T, toMsg func(T) tea.Msg, onClose func() tea.Msg) tea.Cmd {
+	return func() tea.Msg {
+		val, ok := <-ch
+		if !ok {
+			if onClose != nil {
+				return onClose()
+			}
+			return nil
+		}
+		return toMsg(val)
+	}
+}
+
+// ChannelSubscription wraps a channel with helper methods for the
+// re-subscription pattern common in Bubble Tea.
+type ChannelSubscription[T any] struct {
+	ch    <-chan T
+	toMsg func(T) tea.Msg
+}
+
+// NewChannelSubscription creates a subscription for the given channel.
+func NewChannelSubscription[T any](ch <-chan T, toMsg func(T) tea.Msg) *ChannelSubscription[T] {
+	return &ChannelSubscription[T]{ch: ch, toMsg: toMsg}
+}
+
+// Listen returns a Cmd that waits for the next value from the channel.
+// Call this in Init() to start listening, and again in Update() after
+// handling each message to continue listening.
+func (s *ChannelSubscription[T]) Listen() tea.Cmd {
+	return FromChannel(s.ch, s.toMsg)
+}

pkg/tui/tui.go

@@ -28,6 +28,7 @@ import (
 	"github.com/docker/cagent/pkg/tui/page/chat"
 	"github.com/docker/cagent/pkg/tui/service"
 	"github.com/docker/cagent/pkg/tui/styles"
+	"github.com/docker/cagent/pkg/tui/subscription"
 )
 
 // appModel represents the main application model
@@ -48,9 +49,10 @@ type appModel struct {
 
 	transcriber *transcribe.Transcriber
 
-	themeWatcher         *styles.ThemeWatcher
-	themeWatcherEventCh  chan string // Channel for theme file change events (carries themeRef)
-	themeListenerStarted bool        // Guard to prevent multiple listeners
+	// External event subscriptions (Elm Architecture pattern)
+	themeWatcher      *styles.ThemeWatcher
+	themeSubscription *subscription.ChannelSubscription[string] // Listens for theme file changes
+	themeSubStarted   bool                                      // Guard against multiple subscriptions
 
 	// keyboardEnhancements stores the last keyboard enhancements message from the terminal.
 	// This is reapplied to new chat/editor instances when sessions are switched.
@@ -119,21 +121,24 @@ func DefaultKeyMap() KeyMap {
 func New(ctx context.Context, a *app.App) tea.Model {
 	sessionState := service.NewSessionState(a.Session())
 
-	// Create a channel for theme file change events (carries themeRef)
+	// Create a channel for theme file change events
 	themeEventCh := make(chan string, 1)
 
 	t := &appModel{
-		keyMap:              DefaultKeyMap(),
-		dialog:              dialog.New(),
-		notification:        notification.New(),
-		completions:         completion.New(),
-		application:         a,
-		sessionState:        sessionState,
-		transcriber:         transcribe.New(os.Getenv("OPENAI_API_KEY")), // TODO(dga): should use envProvider
-		themeWatcherEventCh: themeEventCh,
+		keyMap:       DefaultKeyMap(),
+		dialog:       dialog.New(),
+		notification: notification.New(),
+		completions:  completion.New(),
+		application:  a,
+		sessionState: sessionState,
+		transcriber:  transcribe.New(os.Getenv("OPENAI_API_KEY")), // TODO(dga): should use envProvider
+		// Set up theme subscription using the subscription package
+		themeSubscription: subscription.NewChannelSubscription(themeEventCh, func(themeRef string) tea.Msg {
+			return messages.ThemeFileChangedMsg{ThemeRef: themeRef}
+		}),
 	}
 
-	// Create theme watcher with callback that sends themeRef to channel
+	// Create theme watcher with callback that sends to the subscription channel
 	t.themeWatcher = styles.NewThemeWatcher(func(themeRef string) {
 		// Non-blocking send to the event channel
 		select {
@@ -170,27 +175,15 @@ func (a *appModel) Init() tea.Cmd {
 		a.application.SendFirstMessage(),
 	}
 
-	// Start theme file listener only once (guard against Init being called multiple times)
-	if !a.themeListenerStarted {
-		a.themeListenerStarted = true
-		cmds = append(cmds, a.listenForThemeFileChanges())
+	// Start theme subscription only once (guard against Init being called multiple times)
+	if !a.themeSubStarted {
+		a.themeSubStarted = true
+		cmds = append(cmds, a.themeSubscription.Listen())
 	}
 
 	return tea.Sequence(cmds...)
 }
 
-// listenForThemeFileChanges returns a command that listens for theme file change events
-// and sends them as messages to the TUI.
-func (a *appModel) listenForThemeFileChanges() tea.Cmd {
-	return func() tea.Msg {
-		themeRef, ok := <-a.themeWatcherEventCh
-		if !ok {
-			return nil // Channel closed
-		}
-		return messages.ThemeFileChangedMsg{ThemeRef: themeRef}
-	}
-}
-
 // Help returns help information
 func (a *appModel) Help() help.KeyMap {
 	return core.NewSimpleHelp(a.Bindings())
@@ -422,14 +415,14 @@ func (a *appModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 		if err != nil {
 			// Failed to load - show error but keep current theme
 			return a, tea.Batch(
-				a.listenForThemeFileChanges(),
+				a.themeSubscription.Listen(), // Re-subscribe to continue listening
 				notification.ErrorCmd(fmt.Sprintf("Failed to hot-reload theme: %v", err)),
 			)
 		}
 		styles.ApplyTheme(theme)
 		// Continue listening for more changes and emit ThemeChangedMsg for cache invalidation
 		return a, tea.Batch(
-			a.listenForThemeFileChanges(),
+			a.themeSubscription.Listen(), // Re-subscribe to continue listening
 			notification.SuccessCmd("Theme hot-reloaded"),
 			core.CmdHandler(messages.ThemeChangedMsg{}),
 		)