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 "strings"
14 "sync"
15 "time"
16
17 tea "charm.land/bubbletea/v2"
18 "charm.land/fantasy"
19 "charm.land/lipgloss/v2"
20 "github.com/charmbracelet/catwalk/pkg/catwalk"
21 "github.com/charmbracelet/crush/internal/agent"
22 "github.com/charmbracelet/crush/internal/agent/tools/mcp"
23 "github.com/charmbracelet/crush/internal/config"
24 "github.com/charmbracelet/crush/internal/csync"
25 "github.com/charmbracelet/crush/internal/db"
26 "github.com/charmbracelet/crush/internal/format"
27 "github.com/charmbracelet/crush/internal/history"
28 "github.com/charmbracelet/crush/internal/log"
29 "github.com/charmbracelet/crush/internal/lsp"
30 "github.com/charmbracelet/crush/internal/message"
31 "github.com/charmbracelet/crush/internal/permission"
32 "github.com/charmbracelet/crush/internal/pubsub"
33 "github.com/charmbracelet/crush/internal/session"
34 "github.com/charmbracelet/crush/internal/shell"
35 "github.com/charmbracelet/crush/internal/tui/components/anim"
36 "github.com/charmbracelet/crush/internal/tui/styles"
37 "github.com/charmbracelet/crush/internal/update"
38 "github.com/charmbracelet/crush/internal/version"
39 "github.com/charmbracelet/x/ansi"
40 "github.com/charmbracelet/x/exp/charmtone"
41 "github.com/charmbracelet/x/term"
42)
43
44// UpdateAvailableMsg is sent when a new version is available.
45type UpdateAvailableMsg struct {
46 CurrentVersion string
47 LatestVersion string
48 IsDevelopment bool
49}
50
51type App struct {
52 Sessions session.Service
53 Messages message.Service
54 History history.Service
55 Permissions permission.Service
56
57 AgentCoordinator agent.Coordinator
58
59 LSPClients *csync.Map[string, *lsp.Client]
60
61 config *config.Config
62
63 serviceEventsWG *sync.WaitGroup
64 eventsCtx context.Context
65 events chan tea.Msg
66 tuiWG *sync.WaitGroup
67
68 // global context and cleanup functions
69 globalCtx context.Context
70 cleanupFuncs []func() error
71}
72
73// New initializes a new application instance.
74func New(ctx context.Context, conn *sql.DB, cfg *config.Config) (*App, error) {
75 q := db.New(conn)
76 sessions := session.NewService(q, conn)
77 messages := message.NewService(q)
78 files := history.NewService(q, conn)
79 skipPermissionsRequests := cfg.Permissions != nil && cfg.Permissions.SkipRequests
80 var allowedTools []string
81 if cfg.Permissions != nil && cfg.Permissions.AllowedTools != nil {
82 allowedTools = cfg.Permissions.AllowedTools
83 }
84
85 app := &App{
86 Sessions: sessions,
87 Messages: messages,
88 History: files,
89 Permissions: permission.NewPermissionService(cfg.WorkingDir(), skipPermissionsRequests, allowedTools),
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, largeModel, smallModel string, quiet bool) error {
136 slog.Info("Running in non-interactive mode")
137
138 ctx, cancel := context.WithCancel(ctx)
139 defer cancel()
140
141 if largeModel != "" || smallModel != "" {
142 if err := app.overrideModelsForNonInteractive(ctx, largeModel, smallModel); err != nil {
143 return fmt.Errorf("failed to override models: %w", err)
144 }
145 }
146
147 var (
148 spinner *format.Spinner
149 stdoutTTY bool
150 stderrTTY bool
151 stdinTTY bool
152 )
153
154 if f, ok := output.(*os.File); ok {
155 stdoutTTY = term.IsTerminal(f.Fd())
156 }
157 stderrTTY = term.IsTerminal(os.Stderr.Fd())
158 stdinTTY = term.IsTerminal(os.Stdin.Fd())
159
160 if !quiet && stderrTTY {
161 t := styles.CurrentTheme()
162
163 // Detect background color to set the appropriate color for the
164 // spinner's 'Generating...' text. Without this, that text would be
165 // unreadable in light terminals.
166 hasDarkBG := true
167 if f, ok := output.(*os.File); ok && stdinTTY && stdoutTTY {
168 hasDarkBG = lipgloss.HasDarkBackground(os.Stdin, f)
169 }
170 defaultFG := lipgloss.LightDark(hasDarkBG)(charmtone.Pepper, t.FgBase)
171
172 spinner = format.NewSpinner(ctx, cancel, anim.Settings{
173 Size: 10,
174 Label: "Generating",
175 LabelColor: defaultFG,
176 GradColorA: t.Primary,
177 GradColorB: t.Secondary,
178 CycleColors: true,
179 })
180 spinner.Start()
181 }
182
183 // Helper function to stop spinner once.
184 stopSpinner := func() {
185 if !quiet && spinner != nil {
186 spinner.Stop()
187 spinner = nil
188 }
189 }
190
191 // Wait for MCP initialization to complete before reading MCP tools.
192 if err := mcp.WaitForInit(ctx); err != nil {
193 return fmt.Errorf("failed to wait for MCP initialization: %w", err)
194 }
195
196 // force update of agent models before running so mcp tools are loaded
197 app.AgentCoordinator.UpdateModels(ctx)
198
199 defer stopSpinner()
200
201 const maxPromptLengthForTitle = 100
202 const titlePrefix = "Non-interactive: "
203 var titleSuffix string
204
205 if len(prompt) > maxPromptLengthForTitle {
206 titleSuffix = prompt[:maxPromptLengthForTitle] + "..."
207 } else {
208 titleSuffix = prompt
209 }
210 title := titlePrefix + titleSuffix
211
212 sess, err := app.Sessions.Create(ctx, title)
213 if err != nil {
214 return fmt.Errorf("failed to create session for non-interactive mode: %w", err)
215 }
216 slog.Info("Created session for non-interactive run", "session_id", sess.ID)
217
218 // Automatically approve all permission requests for this non-interactive
219 // session.
220 app.Permissions.AutoApproveSession(sess.ID)
221
222 type response struct {
223 result *fantasy.AgentResult
224 err error
225 }
226 done := make(chan response, 1)
227
228 go func(ctx context.Context, sessionID, prompt string) {
229 result, err := app.AgentCoordinator.Run(ctx, sess.ID, prompt)
230 if err != nil {
231 done <- response{
232 err: fmt.Errorf("failed to start agent processing stream: %w", err),
233 }
234 }
235 done <- response{
236 result: result,
237 }
238 }(ctx, sess.ID, prompt)
239
240 messageEvents := app.Messages.Subscribe(ctx)
241 messageReadBytes := make(map[string]int)
242
243 defer func() {
244 if stderrTTY {
245 _, _ = fmt.Fprintf(os.Stderr, ansi.ResetProgressBar)
246 }
247
248 // Always print a newline at the end. If output is a TTY this will
249 // prevent the prompt from overwriting the last line of output.
250 _, _ = fmt.Fprintln(output)
251 }()
252
253 for {
254 if stderrTTY {
255 // HACK: Reinitialize the terminal progress bar on every iteration
256 // so it doesn't get hidden by the terminal due to inactivity.
257 _, _ = fmt.Fprintf(os.Stderr, ansi.SetIndeterminateProgressBar)
258 }
259
260 select {
261 case result := <-done:
262 stopSpinner()
263 if result.err != nil {
264 if errors.Is(result.err, context.Canceled) || errors.Is(result.err, agent.ErrRequestCancelled) {
265 slog.Info("Non-interactive: agent processing cancelled", "session_id", sess.ID)
266 return nil
267 }
268 return fmt.Errorf("agent processing failed: %w", result.err)
269 }
270 return nil
271
272 case event := <-messageEvents:
273 msg := event.Payload
274 if msg.SessionID == sess.ID && msg.Role == message.Assistant && len(msg.Parts) > 0 {
275 stopSpinner()
276
277 content := msg.Content().String()
278 readBytes := messageReadBytes[msg.ID]
279
280 if len(content) < readBytes {
281 slog.Error("Non-interactive: message content is shorter than read bytes", "message_length", len(content), "read_bytes", readBytes)
282 return fmt.Errorf("message content is shorter than read bytes: %d < %d", len(content), readBytes)
283 }
284
285 part := content[readBytes:]
286 // Trim leading whitespace. Sometimes the LLM includes leading
287 // formatting and intentation, which we don't want here.
288 if readBytes == 0 {
289 part = strings.TrimLeft(part, " \t")
290 }
291 fmt.Fprint(output, part)
292 messageReadBytes[msg.ID] = len(content)
293 }
294
295 case <-ctx.Done():
296 stopSpinner()
297 return ctx.Err()
298 }
299 }
300}
301
302func (app *App) UpdateAgentModel(ctx context.Context) error {
303 if app.AgentCoordinator == nil {
304 return fmt.Errorf("agent configuration is missing")
305 }
306 return app.AgentCoordinator.UpdateModels(ctx)
307}
308
309// overrideModelsForNonInteractive parses the model strings and temporarily
310// overrides the model configurations, then rebuilds the agent.
311// Format: "model-name" (searches all providers) or "provider/model-name".
312// Model matching is case-insensitive.
313// If largeModel is provided but smallModel is not, the small model defaults to
314// the provider's default small model.
315func (app *App) overrideModelsForNonInteractive(ctx context.Context, largeModel, smallModel string) error {
316 providers := app.config.Providers.Copy()
317
318 largeMatches, smallMatches, err := findModels(providers, largeModel, smallModel)
319 if err != nil {
320 return err
321 }
322
323 var largeProviderID string
324
325 // Override large model.
326 if largeModel != "" {
327 found, err := validateMatches(largeMatches, largeModel, "large")
328 if err != nil {
329 return err
330 }
331 largeProviderID = found.provider
332 slog.Info("Overriding large model for non-interactive run", "provider", found.provider, "model", found.modelID)
333 app.config.Models[config.SelectedModelTypeLarge] = config.SelectedModel{
334 Provider: found.provider,
335 Model: found.modelID,
336 }
337 }
338
339 // Override small model.
340 switch {
341 case smallModel != "":
342 found, err := validateMatches(smallMatches, smallModel, "small")
343 if err != nil {
344 return err
345 }
346 slog.Info("Overriding small model for non-interactive run", "provider", found.provider, "model", found.modelID)
347 app.config.Models[config.SelectedModelTypeSmall] = config.SelectedModel{
348 Provider: found.provider,
349 Model: found.modelID,
350 }
351
352 case largeModel != "":
353 // No small model specified, but large model was - use provider's default.
354 smallCfg := app.GetDefaultSmallModel(largeProviderID)
355 app.config.Models[config.SelectedModelTypeSmall] = smallCfg
356 }
357
358 return app.AgentCoordinator.UpdateModels(ctx)
359}
360
361// GetDefaultSmallModel returns the default small model for the given
362// provider. Falls back to the large model if no default is found.
363func (app *App) GetDefaultSmallModel(providerID string) config.SelectedModel {
364 cfg := app.config
365 largeModelCfg := cfg.Models[config.SelectedModelTypeLarge]
366
367 // Find the provider in the known providers list to get its default small model.
368 knownProviders, _ := config.Providers(cfg)
369 var knownProvider *catwalk.Provider
370 for _, p := range knownProviders {
371 if string(p.ID) == providerID {
372 knownProvider = &p
373 break
374 }
375 }
376
377 // For unknown/local providers, use the large model as small.
378 if knownProvider == nil {
379 slog.Warn("Using large model as small model for unknown provider", "provider", providerID, "model", largeModelCfg.Model)
380 return largeModelCfg
381 }
382
383 defaultSmallModelID := knownProvider.DefaultSmallModelID
384 model := cfg.GetModel(providerID, defaultSmallModelID)
385 if model == nil {
386 slog.Warn("Default small model not found, using large model", "provider", providerID, "model", largeModelCfg.Model)
387 return largeModelCfg
388 }
389
390 slog.Info("Using provider default small model", "provider", providerID, "model", defaultSmallModelID)
391 return config.SelectedModel{
392 Provider: providerID,
393 Model: defaultSmallModelID,
394 MaxTokens: model.DefaultMaxTokens,
395 ReasoningEffort: model.DefaultReasoningEffort,
396 }
397}
398
399func (app *App) setupEvents() {
400 ctx, cancel := context.WithCancel(app.globalCtx)
401 app.eventsCtx = ctx
402 setupSubscriber(ctx, app.serviceEventsWG, "sessions", app.Sessions.Subscribe, app.events)
403 setupSubscriber(ctx, app.serviceEventsWG, "messages", app.Messages.Subscribe, app.events)
404 setupSubscriber(ctx, app.serviceEventsWG, "permissions", app.Permissions.Subscribe, app.events)
405 setupSubscriber(ctx, app.serviceEventsWG, "permissions-notifications", app.Permissions.SubscribeNotifications, app.events)
406 setupSubscriber(ctx, app.serviceEventsWG, "history", app.History.Subscribe, app.events)
407 setupSubscriber(ctx, app.serviceEventsWG, "mcp", mcp.SubscribeEvents, app.events)
408 setupSubscriber(ctx, app.serviceEventsWG, "lsp", SubscribeLSPEvents, app.events)
409 cleanupFunc := func() error {
410 cancel()
411 app.serviceEventsWG.Wait()
412 return nil
413 }
414 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
415}
416
417func setupSubscriber[T any](
418 ctx context.Context,
419 wg *sync.WaitGroup,
420 name string,
421 subscriber func(context.Context) <-chan pubsub.Event[T],
422 outputCh chan<- tea.Msg,
423) {
424 wg.Go(func() {
425 subCh := subscriber(ctx)
426 for {
427 select {
428 case event, ok := <-subCh:
429 if !ok {
430 slog.Debug("subscription channel closed", "name", name)
431 return
432 }
433 var msg tea.Msg = event
434 select {
435 case outputCh <- msg:
436 case <-time.After(2 * time.Second):
437 slog.Warn("message dropped due to slow consumer", "name", name)
438 case <-ctx.Done():
439 slog.Debug("subscription cancelled", "name", name)
440 return
441 }
442 case <-ctx.Done():
443 slog.Debug("subscription cancelled", "name", name)
444 return
445 }
446 }
447 })
448}
449
450func (app *App) InitCoderAgent(ctx context.Context) error {
451 coderAgentCfg := app.config.Agents[config.AgentCoder]
452 if coderAgentCfg.ID == "" {
453 return fmt.Errorf("coder agent configuration is missing")
454 }
455 var err error
456 app.AgentCoordinator, err = agent.NewCoordinator(
457 ctx,
458 app.config,
459 app.Sessions,
460 app.Messages,
461 app.Permissions,
462 app.History,
463 app.LSPClients,
464 )
465 if err != nil {
466 slog.Error("Failed to create coder agent", "err", err)
467 return err
468 }
469 return nil
470}
471
472// Subscribe sends events to the TUI as tea.Msgs.
473func (app *App) Subscribe(program *tea.Program) {
474 defer log.RecoverPanic("app.Subscribe", func() {
475 slog.Info("TUI subscription panic: attempting graceful shutdown")
476 program.Quit()
477 })
478
479 app.tuiWG.Add(1)
480 tuiCtx, tuiCancel := context.WithCancel(app.globalCtx)
481 app.cleanupFuncs = append(app.cleanupFuncs, func() error {
482 slog.Debug("Cancelling TUI message handler")
483 tuiCancel()
484 app.tuiWG.Wait()
485 return nil
486 })
487 defer app.tuiWG.Done()
488
489 for {
490 select {
491 case <-tuiCtx.Done():
492 slog.Debug("TUI message handler shutting down")
493 return
494 case msg, ok := <-app.events:
495 if !ok {
496 slog.Debug("TUI message channel closed")
497 return
498 }
499 program.Send(msg)
500 }
501 }
502}
503
504// Shutdown performs a graceful shutdown of the application.
505func (app *App) Shutdown() {
506 start := time.Now()
507 defer func() { slog.Info("Shutdown took " + time.Since(start).String()) }()
508
509 // First, cancel all agents and wait for them to finish. This must complete
510 // before closing the DB so agents can finish writing their state.
511 if app.AgentCoordinator != nil {
512 app.AgentCoordinator.CancelAll()
513 }
514
515 // Now run remaining cleanup tasks in parallel.
516 var wg sync.WaitGroup
517
518 // Kill all background shells.
519 wg.Go(func() {
520 shell.GetBackgroundShellManager().KillAll()
521 })
522
523 // Shutdown all LSP clients.
524 shutdownCtx, cancel := context.WithTimeout(app.globalCtx, 5*time.Second)
525 defer cancel()
526 for name, client := range app.LSPClients.Seq2() {
527 wg.Go(func() {
528 if err := client.Close(shutdownCtx); err != nil &&
529 !errors.Is(err, io.EOF) &&
530 !errors.Is(err, context.Canceled) &&
531 err.Error() != "signal: killed" {
532 slog.Warn("Failed to shutdown LSP client", "name", name, "error", err)
533 }
534 })
535 }
536
537 // Call all cleanup functions.
538 for _, cleanup := range app.cleanupFuncs {
539 if cleanup != nil {
540 wg.Go(func() {
541 if err := cleanup(); err != nil {
542 slog.Error("Failed to cleanup app properly on shutdown", "error", err)
543 }
544 })
545 }
546 }
547 wg.Wait()
548}
549
550// checkForUpdates checks for available updates.
551func (app *App) checkForUpdates(ctx context.Context) {
552 checkCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
553 defer cancel()
554
555 info, err := update.Check(checkCtx, version.Version, update.Default)
556 if err != nil || !info.Available() {
557 return
558 }
559 app.events <- UpdateAvailableMsg{
560 CurrentVersion: info.Current,
561 LatestVersion: info.Latest,
562 IsDevelopment: info.IsDevelopment(),
563 }
564}