1// Package app wires together services, coordinates agents, and manages
2// application lifecycle.
3package app
4
5import (
6 "context"
7 "database/sql"
8 "errors"
9 "fmt"
10 "io"
11 "log/slog"
12 "os"
13 "sync"
14 "time"
15
16 tea "charm.land/bubbletea/v2"
17 "charm.land/fantasy"
18 "charm.land/lipgloss/v2"
19 "git.secluded.site/crush/internal/agent"
20 "git.secluded.site/crush/internal/agent/tools/mcp"
21 "git.secluded.site/crush/internal/config"
22 "git.secluded.site/crush/internal/csync"
23 "git.secluded.site/crush/internal/db"
24 "git.secluded.site/crush/internal/format"
25 "git.secluded.site/crush/internal/history"
26 "git.secluded.site/crush/internal/log"
27 "git.secluded.site/crush/internal/lsp"
28 "git.secluded.site/crush/internal/message"
29 "git.secluded.site/crush/internal/notification"
30 "git.secluded.site/crush/internal/permission"
31 "git.secluded.site/crush/internal/pubsub"
32 "git.secluded.site/crush/internal/session"
33 "git.secluded.site/crush/internal/shell"
34 "git.secluded.site/crush/internal/term"
35 "git.secluded.site/crush/internal/tui/components/anim"
36 "git.secluded.site/crush/internal/tui/styles"
37 "git.secluded.site/crush/internal/update"
38 "git.secluded.site/crush/internal/version"
39 "github.com/charmbracelet/x/ansi"
40 "github.com/charmbracelet/x/exp/charmtone"
41)
42
43type App struct {
44 Sessions session.Service
45 Messages message.Service
46 History history.Service
47 Permissions permission.Service
48
49 AgentCoordinator agent.Coordinator
50 Notifier *notification.Notifier
51
52 LSPClients *csync.Map[string, *lsp.Client]
53
54 config *config.Config
55
56 serviceEventsWG *sync.WaitGroup
57 eventsCtx context.Context
58 events chan tea.Msg
59 tuiWG *sync.WaitGroup
60
61 // global context and cleanup functions
62 globalCtx context.Context
63 cleanupFuncs []func() error
64}
65
66// New initializes a new applcation instance.
67func New(ctx context.Context, conn *sql.DB, cfg *config.Config) (*App, error) {
68 q := db.New(conn)
69 sessions := session.NewService(q)
70 messages := message.NewService(q)
71 files := history.NewService(q, conn)
72 skipPermissionsRequests := cfg.Permissions != nil && cfg.Permissions.SkipRequests
73 allowedTools := []string{}
74 if cfg.Permissions != nil && cfg.Permissions.AllowedTools != nil {
75 allowedTools = cfg.Permissions.AllowedTools
76 }
77
78 // Enable notifications by default, disable if configured.
79 notificationsEnabled := cfg.Options == nil || !cfg.Options.DisableNotifications
80
81 notifier := notification.New(notificationsEnabled)
82 permissions := permission.NewPermissionService(ctx, cfg.WorkingDir(), skipPermissionsRequests, allowedTools, notifier)
83
84 app := &App{
85 Sessions: sessions,
86 Messages: messages,
87 History: files,
88 Permissions: permissions,
89 Notifier: notifier,
90 LSPClients: csync.NewMap[string, *lsp.Client](),
91
92 globalCtx: ctx,
93
94 config: cfg,
95
96 events: make(chan tea.Msg, 100),
97 serviceEventsWG: &sync.WaitGroup{},
98 tuiWG: &sync.WaitGroup{},
99 }
100
101 app.setupEvents()
102
103 // Initialize LSP clients in the background.
104 app.initLSPClients(ctx)
105
106 // Check for updates in the background.
107 go app.checkForUpdates(ctx)
108
109 go func() {
110 slog.Info("Initializing MCP clients")
111 mcp.Initialize(ctx, app.Permissions, cfg)
112 }()
113
114 // cleanup database upon app shutdown
115 app.cleanupFuncs = append(app.cleanupFuncs, conn.Close, mcp.Close)
116
117 // TODO: remove the concept of agent config, most likely.
118 if !cfg.IsConfigured() {
119 slog.Warn("No agent configuration found")
120 return app, nil
121 }
122 if err := app.InitCoderAgent(ctx); err != nil {
123 return nil, fmt.Errorf("failed to initialize coder agent: %w", err)
124 }
125 return app, nil
126}
127
128// Config returns the application configuration.
129func (app *App) Config() *config.Config {
130 return app.config
131}
132
133// RunNonInteractive runs the application in non-interactive mode with the
134// given prompt, printing to stdout.
135func (app *App) RunNonInteractive(ctx context.Context, output io.Writer, prompt string, quiet bool) error {
136 slog.Info("Running in non-interactive mode")
137
138 ctx, cancel := context.WithCancel(ctx)
139 defer cancel()
140
141 var spinner *format.Spinner
142 if !quiet {
143 t := styles.CurrentTheme()
144
145 // Detect background color to set the appropriate color for the
146 // spinner's 'Generating...' text. Without this, that text would be
147 // unreadable in light terminals.
148 hasDarkBG := true
149 if f, ok := output.(*os.File); ok {
150 hasDarkBG = lipgloss.HasDarkBackground(os.Stdin, f)
151 }
152 defaultFG := lipgloss.LightDark(hasDarkBG)(charmtone.Pepper, t.FgBase)
153
154 spinner = format.NewSpinner(ctx, cancel, anim.Settings{
155 Size: 10,
156 Label: "Generating",
157 LabelColor: defaultFG,
158 GradColorA: t.Primary,
159 GradColorB: t.Secondary,
160 CycleColors: true,
161 })
162 spinner.Start()
163 }
164
165 // Helper function to stop spinner once.
166 stopSpinner := func() {
167 if !quiet && spinner != nil {
168 spinner.Stop()
169 spinner = nil
170 }
171 }
172 defer stopSpinner()
173
174 const maxPromptLengthForTitle = 100
175 const titlePrefix = "Non-interactive: "
176 var titleSuffix string
177
178 if len(prompt) > maxPromptLengthForTitle {
179 titleSuffix = prompt[:maxPromptLengthForTitle] + "..."
180 } else {
181 titleSuffix = prompt
182 }
183 title := titlePrefix + titleSuffix
184
185 sess, err := app.Sessions.Create(ctx, title)
186 if err != nil {
187 return fmt.Errorf("failed to create session for non-interactive mode: %w", err)
188 }
189 slog.Info("Created session for non-interactive run", "session_id", sess.ID)
190
191 // Automatically approve all permission requests for this non-interactive
192 // session.
193 app.Permissions.AutoApproveSession(sess.ID)
194
195 type response struct {
196 result *fantasy.AgentResult
197 err error
198 }
199 done := make(chan response, 1)
200
201 go func(ctx context.Context, sessionID, prompt string) {
202 result, err := app.AgentCoordinator.Run(ctx, sess.ID, prompt)
203 if err != nil {
204 done <- response{
205 err: fmt.Errorf("failed to start agent processing stream: %w", err),
206 }
207 }
208 done <- response{
209 result: result,
210 }
211 }(ctx, sess.ID, prompt)
212
213 messageEvents := app.Messages.Subscribe(ctx)
214 messageReadBytes := make(map[string]int)
215 supportsProgressBar := term.SupportsProgressBar()
216
217 defer func() {
218 if supportsProgressBar {
219 _, _ = fmt.Fprintf(os.Stderr, ansi.ResetProgressBar)
220 }
221
222 // Always print a newline at the end. If output is a TTY this will
223 // prevent the prompt from overwriting the last line of output.
224 _, _ = fmt.Fprintln(output)
225 }()
226
227 for {
228 if supportsProgressBar {
229 // HACK: Reinitialize the terminal progress bar on every iteration so
230 // it doesn't get hidden by the terminal due to inactivity.
231 _, _ = fmt.Fprintf(os.Stderr, ansi.SetIndeterminateProgressBar)
232 }
233
234 select {
235 case result := <-done:
236 stopSpinner()
237 if result.err != nil {
238 if errors.Is(result.err, context.Canceled) || errors.Is(result.err, agent.ErrRequestCancelled) {
239 slog.Info("Non-interactive: agent processing cancelled", "session_id", sess.ID)
240 return nil
241 }
242 return fmt.Errorf("agent processing failed: %w", result.err)
243 }
244 return nil
245
246 case event := <-messageEvents:
247 msg := event.Payload
248 if msg.SessionID == sess.ID && msg.Role == message.Assistant && len(msg.Parts) > 0 {
249 stopSpinner()
250
251 content := msg.Content().String()
252 readBytes := messageReadBytes[msg.ID]
253
254 if len(content) < readBytes {
255 slog.Error("Non-interactive: message content is shorter than read bytes", "message_length", len(content), "read_bytes", readBytes)
256 return fmt.Errorf("message content is shorter than read bytes: %d < %d", len(content), readBytes)
257 }
258
259 part := content[readBytes:]
260 fmt.Fprint(output, part)
261 messageReadBytes[msg.ID] = len(content)
262 }
263
264 case <-ctx.Done():
265 stopSpinner()
266 return ctx.Err()
267 }
268 }
269}
270
271func (app *App) UpdateAgentModel(ctx context.Context) error {
272 return app.AgentCoordinator.UpdateModels(ctx)
273}
274
275func (app *App) setupEvents() {
276 ctx, cancel := context.WithCancel(app.globalCtx)
277 app.eventsCtx = ctx
278 setupSubscriber(ctx, app.serviceEventsWG, "sessions", app.Sessions.Subscribe, app.events)
279 setupSubscriber(ctx, app.serviceEventsWG, "messages", app.Messages.Subscribe, app.events)
280 setupSubscriber(ctx, app.serviceEventsWG, "permissions", app.Permissions.Subscribe, app.events)
281 setupSubscriber(ctx, app.serviceEventsWG, "permissions-notifications", app.Permissions.SubscribeNotifications, app.events)
282 setupSubscriber(ctx, app.serviceEventsWG, "history", app.History.Subscribe, app.events)
283 setupSubscriber(ctx, app.serviceEventsWG, "mcp", mcp.SubscribeEvents, app.events)
284 setupSubscriber(ctx, app.serviceEventsWG, "lsp", SubscribeLSPEvents, app.events)
285 cleanupFunc := func() error {
286 cancel()
287 app.serviceEventsWG.Wait()
288 return nil
289 }
290 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
291}
292
293func setupSubscriber[T any](
294 ctx context.Context,
295 wg *sync.WaitGroup,
296 name string,
297 subscriber func(context.Context) <-chan pubsub.Event[T],
298 outputCh chan<- tea.Msg,
299) {
300 wg.Go(func() {
301 subCh := subscriber(ctx)
302 for {
303 select {
304 case event, ok := <-subCh:
305 if !ok {
306 slog.Debug("subscription channel closed", "name", name)
307 return
308 }
309 var msg tea.Msg = event
310 select {
311 case outputCh <- msg:
312 case <-time.After(2 * time.Second):
313 slog.Warn("message dropped due to slow consumer", "name", name)
314 case <-ctx.Done():
315 slog.Debug("subscription cancelled", "name", name)
316 return
317 }
318 case <-ctx.Done():
319 slog.Debug("subscription cancelled", "name", name)
320 return
321 }
322 }
323 })
324}
325
326func (app *App) InitCoderAgent(ctx context.Context) error {
327 coderAgentCfg := app.config.Agents[config.AgentCoder]
328 if coderAgentCfg.ID == "" {
329 return fmt.Errorf("coder agent configuration is missing")
330 }
331 var err error
332 app.AgentCoordinator, err = agent.NewCoordinator(
333 ctx,
334 app.config,
335 app.Sessions,
336 app.Messages,
337 app.Permissions,
338 app.History,
339 app.LSPClients,
340 app.Notifier,
341 )
342 if err != nil {
343 slog.Error("Failed to create coder agent", "err", err)
344 return err
345 }
346 return nil
347}
348
349// Subscribe sends events to the TUI as tea.Msgs.
350func (app *App) Subscribe(program *tea.Program) {
351 defer log.RecoverPanic("app.Subscribe", func() {
352 slog.Info("TUI subscription panic: attempting graceful shutdown")
353 program.Quit()
354 })
355
356 app.tuiWG.Add(1)
357 tuiCtx, tuiCancel := context.WithCancel(app.globalCtx)
358 app.cleanupFuncs = append(app.cleanupFuncs, func() error {
359 slog.Debug("Cancelling TUI message handler")
360 tuiCancel()
361 app.tuiWG.Wait()
362 return nil
363 })
364 defer app.tuiWG.Done()
365
366 for {
367 select {
368 case <-tuiCtx.Done():
369 slog.Debug("TUI message handler shutting down")
370 return
371 case msg, ok := <-app.events:
372 if !ok {
373 slog.Debug("TUI message channel closed")
374 return
375 }
376 program.Send(msg)
377 }
378 }
379}
380
381// Shutdown performs a graceful shutdown of the application.
382func (app *App) Shutdown() {
383 if app.AgentCoordinator != nil {
384 app.AgentCoordinator.CancelAll()
385 }
386
387 // Kill all background shells.
388 shell.GetBackgroundShellManager().KillAll()
389
390 // Shutdown all LSP clients.
391 for name, client := range app.LSPClients.Seq2() {
392 shutdownCtx, cancel := context.WithTimeout(app.globalCtx, 5*time.Second)
393 if err := client.Close(shutdownCtx); err != nil {
394 slog.Error("Failed to shutdown LSP client", "name", name, "error", err)
395 }
396 cancel()
397 }
398
399 // Call call cleanup functions.
400 for _, cleanup := range app.cleanupFuncs {
401 if cleanup != nil {
402 if err := cleanup(); err != nil {
403 slog.Error("Failed to cleanup app properly on shutdown", "error", err)
404 }
405 }
406 }
407}
408
409// checkForUpdates checks for available updates.
410func (app *App) checkForUpdates(ctx context.Context) {
411 checkCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
412 defer cancel()
413
414 info, err := update.Check(checkCtx, version.Version, update.Default)
415 if err != nil || !info.Available() {
416 return
417 }
418 app.events <- pubsub.UpdateAvailableMsg{
419 CurrentVersion: info.Current,
420 LatestVersion: info.Latest,
421 IsDevelopment: info.IsDevelopment(),
422 }
423}