root.go 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252
  1. package cmd
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "sync"
  7. "time"
  8. tea "github.com/charmbracelet/bubbletea"
  9. "github.com/kujtimiihoxha/termai/internal/app"
  10. "github.com/kujtimiihoxha/termai/internal/config"
  11. "github.com/kujtimiihoxha/termai/internal/db"
  12. "github.com/kujtimiihoxha/termai/internal/llm/agent"
  13. "github.com/kujtimiihoxha/termai/internal/logging"
  14. "github.com/kujtimiihoxha/termai/internal/pubsub"
  15. "github.com/kujtimiihoxha/termai/internal/tui"
  16. zone "github.com/lrstanley/bubblezone"
  17. "github.com/spf13/cobra"
  18. )
  19. var rootCmd = &cobra.Command{
  20. Use: "OpenCode",
  21. Short: "A terminal ai assistant",
  22. Long: `A terminal ai assistant`,
  23. RunE: func(cmd *cobra.Command, args []string) error {
  24. // If the help flag is set, show the help message
  25. if cmd.Flag("help").Changed {
  26. cmd.Help()
  27. return nil
  28. }
  29. // Load the config
  30. debug, _ := cmd.Flags().GetBool("debug")
  31. cwd, _ := cmd.Flags().GetString("cwd")
  32. if cwd != "" {
  33. err := os.Chdir(cwd)
  34. if err != nil {
  35. return fmt.Errorf("failed to change directory: %v", err)
  36. }
  37. }
  38. if cwd == "" {
  39. c, err := os.Getwd()
  40. if err != nil {
  41. return fmt.Errorf("failed to get current working directory: %v", err)
  42. }
  43. cwd = c
  44. }
  45. _, err := config.Load(cwd, debug)
  46. if err != nil {
  47. return err
  48. }
  49. // Connect DB, this will also run migrations
  50. conn, err := db.Connect()
  51. if err != nil {
  52. return err
  53. }
  54. // Create main context for the application
  55. ctx, cancel := context.WithCancel(context.Background())
  56. defer cancel()
  57. app, err := app.New(ctx, conn)
  58. if err != nil {
  59. logging.Error("Failed to create app: %v", err)
  60. return err
  61. }
  62. // Set up the TUI
  63. zone.NewGlobal()
  64. program := tea.NewProgram(
  65. tui.New(app),
  66. tea.WithAltScreen(),
  67. tea.WithMouseCellMotion(),
  68. )
  69. // Initialize MCP tools in the background
  70. initMCPTools(ctx, app)
  71. // Setup the subscriptions, this will send services events to the TUI
  72. ch, cancelSubs := setupSubscriptions(app)
  73. // Create a context for the TUI message handler
  74. tuiCtx, tuiCancel := context.WithCancel(ctx)
  75. var tuiWg sync.WaitGroup
  76. tuiWg.Add(1)
  77. // Set up message handling for the TUI
  78. go func() {
  79. defer tuiWg.Done()
  80. defer logging.RecoverPanic("TUI-message-handler", func() {
  81. attemptTUIRecovery(program)
  82. })
  83. for {
  84. select {
  85. case <-tuiCtx.Done():
  86. logging.Info("TUI message handler shutting down")
  87. return
  88. case msg, ok := <-ch:
  89. if !ok {
  90. logging.Info("TUI message channel closed")
  91. return
  92. }
  93. program.Send(msg)
  94. }
  95. }
  96. }()
  97. // Cleanup function for when the program exits
  98. cleanup := func() {
  99. // Shutdown the app
  100. app.Shutdown()
  101. // Cancel subscriptions first
  102. cancelSubs()
  103. // Then cancel TUI message handler
  104. tuiCancel()
  105. // Wait for TUI message handler to finish
  106. tuiWg.Wait()
  107. logging.Info("All goroutines cleaned up")
  108. }
  109. // Run the TUI
  110. result, err := program.Run()
  111. cleanup()
  112. if err != nil {
  113. logging.Error("TUI error: %v", err)
  114. return fmt.Errorf("TUI error: %v", err)
  115. }
  116. logging.Info("TUI exited with result: %v", result)
  117. return nil
  118. },
  119. }
  120. // attemptTUIRecovery tries to recover the TUI after a panic
  121. func attemptTUIRecovery(program *tea.Program) {
  122. logging.Info("Attempting to recover TUI after panic")
  123. // We could try to restart the TUI or gracefully exit
  124. // For now, we'll just quit the program to avoid further issues
  125. program.Quit()
  126. }
  127. func initMCPTools(ctx context.Context, app *app.App) {
  128. go func() {
  129. defer logging.RecoverPanic("MCP-goroutine", nil)
  130. // Create a context with timeout for the initial MCP tools fetch
  131. ctxWithTimeout, cancel := context.WithTimeout(ctx, 30*time.Second)
  132. defer cancel()
  133. // Set this up once with proper error handling
  134. agent.GetMcpTools(ctxWithTimeout, app.Permissions)
  135. logging.Info("MCP message handling goroutine exiting")
  136. }()
  137. }
  138. func setupSubscriber[T any](
  139. ctx context.Context,
  140. wg *sync.WaitGroup,
  141. name string,
  142. subscriber func(context.Context) <-chan pubsub.Event[T],
  143. outputCh chan<- tea.Msg,
  144. ) {
  145. wg.Add(1)
  146. go func() {
  147. defer wg.Done()
  148. defer logging.RecoverPanic(fmt.Sprintf("subscription-%s", name), nil)
  149. for {
  150. select {
  151. case event, ok := <-subscriber(ctx):
  152. if !ok {
  153. logging.Info("%s subscription channel closed", name)
  154. return
  155. }
  156. // Convert generic event to tea.Msg if needed
  157. var msg tea.Msg = event
  158. // Non-blocking send with timeout to prevent deadlocks
  159. select {
  160. case outputCh <- msg:
  161. case <-time.After(500 * time.Millisecond):
  162. logging.Warn("%s message dropped due to slow consumer", name)
  163. case <-ctx.Done():
  164. logging.Info("%s subscription cancelled", name)
  165. return
  166. }
  167. case <-ctx.Done():
  168. logging.Info("%s subscription cancelled", name)
  169. return
  170. }
  171. }
  172. }()
  173. }
  174. func setupSubscriptions(app *app.App) (chan tea.Msg, func()) {
  175. ch := make(chan tea.Msg, 100)
  176. // Add a buffer to prevent blocking
  177. wg := sync.WaitGroup{}
  178. ctx, cancel := context.WithCancel(context.Background())
  179. // Setup each subscription using the helper
  180. setupSubscriber(ctx, &wg, "logging", logging.Subscribe, ch)
  181. setupSubscriber(ctx, &wg, "sessions", app.Sessions.Subscribe, ch)
  182. setupSubscriber(ctx, &wg, "messages", app.Messages.Subscribe, ch)
  183. setupSubscriber(ctx, &wg, "permissions", app.Permissions.Subscribe, ch)
  184. // Return channel and a cleanup function
  185. cleanupFunc := func() {
  186. logging.Info("Cancelling all subscriptions")
  187. cancel() // Signal all goroutines to stop
  188. // Wait with a timeout for all goroutines to complete
  189. waitCh := make(chan struct{})
  190. go func() {
  191. defer logging.RecoverPanic("subscription-cleanup", nil)
  192. wg.Wait()
  193. close(waitCh)
  194. }()
  195. select {
  196. case <-waitCh:
  197. logging.Info("All subscription goroutines completed successfully")
  198. case <-time.After(5 * time.Second):
  199. logging.Warn("Timed out waiting for some subscription goroutines to complete")
  200. }
  201. close(ch) // Safe to close after all writers are done or timed out
  202. }
  203. return ch, cleanupFunc
  204. }
  205. func Execute() {
  206. err := rootCmd.Execute()
  207. if err != nil {
  208. os.Exit(1)
  209. }
  210. }
  211. func init() {
  212. rootCmd.Flags().BoolP("help", "h", false, "Help")
  213. rootCmd.Flags().BoolP("debug", "d", false, "Debug")
  214. rootCmd.Flags().StringP("cwd", "c", "", "Current working directory")
  215. }