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/catwalk/pkg/catwalk"
19 "charm.land/fantasy"
20 "charm.land/lipgloss/v2"
21 "github.com/charmbracelet/crush/internal/agent"
22 "github.com/charmbracelet/crush/internal/agent/notify"
23 "github.com/charmbracelet/crush/internal/agent/tools/mcp"
24 "github.com/charmbracelet/crush/internal/config"
25 "github.com/charmbracelet/crush/internal/db"
26 "github.com/charmbracelet/crush/internal/event"
27 "github.com/charmbracelet/crush/internal/filetracker"
28 "github.com/charmbracelet/crush/internal/format"
29 "github.com/charmbracelet/crush/internal/history"
30 "github.com/charmbracelet/crush/internal/log"
31 "github.com/charmbracelet/crush/internal/lsp"
32 "github.com/charmbracelet/crush/internal/message"
33 "github.com/charmbracelet/crush/internal/permission"
34 "github.com/charmbracelet/crush/internal/pubsub"
35 "github.com/charmbracelet/crush/internal/session"
36 "github.com/charmbracelet/crush/internal/shell"
37 "github.com/charmbracelet/crush/internal/ui/anim"
38 "github.com/charmbracelet/crush/internal/ui/styles"
39 "github.com/charmbracelet/crush/internal/update"
40 "github.com/charmbracelet/crush/internal/version"
41 "github.com/charmbracelet/x/ansi"
42 "github.com/charmbracelet/x/exp/charmtone"
43 "github.com/charmbracelet/x/term"
44)
45
46// UpdateAvailableMsg is sent when a new version is available.
47type UpdateAvailableMsg struct {
48 CurrentVersion string
49 LatestVersion string
50 IsDevelopment bool
51}
52
53type App struct {
54 Sessions session.Service
55 Messages message.Service
56 History history.Service
57 Permissions permission.Service
58 FileTracker filetracker.Service
59
60 AgentCoordinator agent.Coordinator
61
62 LSPManager *lsp.Manager
63
64 config *config.ConfigStore
65
66 serviceEventsWG *sync.WaitGroup
67 eventsCtx context.Context
68 events chan tea.Msg
69 tuiWG *sync.WaitGroup
70
71 // global context and cleanup functions
72 globalCtx context.Context
73 cleanupFuncs []func(context.Context) error
74 agentNotifications *pubsub.Broker[notify.Notification]
75}
76
77// New initializes a new application instance.
78func New(ctx context.Context, conn *sql.DB, store *config.ConfigStore) (*App, error) {
79 q := db.New(conn)
80 sessions := session.NewService(q, conn)
81 messages := message.NewService(q)
82 files := history.NewService(q, conn)
83 cfg := store.Config()
84 skipPermissionsRequests := store.Overrides().SkipPermissionRequests
85 var allowedTools []string
86 if cfg.Permissions != nil && cfg.Permissions.AllowedTools != nil {
87 allowedTools = cfg.Permissions.AllowedTools
88 }
89
90 app := &App{
91 Sessions: sessions,
92 Messages: messages,
93 History: files,
94 Permissions: permission.NewPermissionService(store.WorkingDir(), skipPermissionsRequests, allowedTools),
95 FileTracker: filetracker.NewService(q),
96 LSPManager: lsp.NewManager(store),
97
98 globalCtx: ctx,
99
100 config: store,
101
102 events: make(chan tea.Msg, 100),
103 serviceEventsWG: &sync.WaitGroup{},
104 tuiWG: &sync.WaitGroup{},
105 agentNotifications: pubsub.NewBroker[notify.Notification](),
106 }
107
108 app.setupEvents()
109
110 // Check for updates in the background.
111 go app.checkForUpdates(ctx)
112
113 go mcp.Initialize(ctx, app.Permissions, store)
114
115 // cleanup database upon app shutdown
116 app.cleanupFuncs = append(
117 app.cleanupFuncs,
118 func(context.Context) error { return conn.Close() },
119 func(ctx context.Context) error { return mcp.Close(ctx) },
120 )
121
122 // TODO: remove the concept of agent config, most likely.
123 if !cfg.IsConfigured() {
124 slog.Warn("No agent configuration found")
125 return app, nil
126 }
127 if err := app.InitCoderAgent(ctx); err != nil {
128 return nil, fmt.Errorf("failed to initialize coder agent: %w", err)
129 }
130
131 // Set up callback for LSP state updates.
132 app.LSPManager.SetCallback(func(name string, client *lsp.Client) {
133 if client == nil {
134 updateLSPState(name, lsp.StateUnstarted, nil, nil, 0)
135 return
136 }
137 client.SetDiagnosticsCallback(updateLSPDiagnostics)
138 updateLSPState(name, client.GetServerState(), nil, client, 0)
139 })
140 go app.LSPManager.TrackConfigured()
141
142 return app, nil
143}
144
145// Config returns the pure-data configuration.
146func (app *App) Config() *config.Config {
147 return app.config.Config()
148}
149
150// Store returns the config store.
151func (app *App) Store() *config.ConfigStore {
152 return app.config
153}
154
155// Events returns the events channel for the application.
156func (app *App) Events() <-chan tea.Msg {
157 return app.events
158}
159
160// SendEvent pushes a message into the application's events channel.
161// It is non-blocking; the message is dropped if the channel is full.
162func (app *App) SendEvent(msg tea.Msg) {
163 select {
164 case app.events <- msg:
165 default:
166 }
167}
168
169// AgentNotifications returns the broker for agent notification events.
170func (app *App) AgentNotifications() *pubsub.Broker[notify.Notification] {
171 return app.agentNotifications
172}
173
174// resolveSession resolves which session to use for a non-interactive run
175// If continueSessionID is set, it looks up that session by ID
176// If useLast is set, it returns the most recently updated top-level session
177// Otherwise, it creates a new session
178func (app *App) resolveSession(ctx context.Context, continueSessionID string, useLast bool) (session.Session, error) {
179 switch {
180 case continueSessionID != "":
181 if app.Sessions.IsAgentToolSession(continueSessionID) {
182 return session.Session{}, fmt.Errorf("cannot continue an agent tool session: %s", continueSessionID)
183 }
184 sess, err := app.Sessions.Get(ctx, continueSessionID)
185 if err != nil {
186 return session.Session{}, fmt.Errorf("session not found: %s", continueSessionID)
187 }
188 if sess.ParentSessionID != "" {
189 return session.Session{}, fmt.Errorf("cannot continue a child session: %s", continueSessionID)
190 }
191 return sess, nil
192
193 case useLast:
194 sess, err := app.Sessions.GetLast(ctx)
195 if err != nil {
196 return session.Session{}, fmt.Errorf("no sessions found to continue")
197 }
198 return sess, nil
199
200 default:
201 return app.Sessions.Create(ctx, agent.DefaultSessionName)
202 }
203}
204
205// RunNonInteractive runs the application in non-interactive mode with the
206// given prompt, printing to stdout.
207func (app *App) RunNonInteractive(ctx context.Context, output io.Writer, prompt, largeModel, smallModel string, hideSpinner bool, continueSessionID string, useLast bool) error {
208 slog.Info("Running in non-interactive mode")
209
210 ctx, cancel := context.WithCancel(ctx)
211 defer cancel()
212
213 if largeModel != "" || smallModel != "" {
214 if err := app.overrideModelsForNonInteractive(ctx, largeModel, smallModel); err != nil {
215 return fmt.Errorf("failed to override models: %w", err)
216 }
217 }
218
219 var (
220 spinner *format.Spinner
221 stdoutTTY bool
222 stderrTTY bool
223 stdinTTY bool
224 progress bool
225 )
226
227 if f, ok := output.(*os.File); ok {
228 stdoutTTY = term.IsTerminal(f.Fd())
229 }
230 stderrTTY = term.IsTerminal(os.Stderr.Fd())
231 stdinTTY = term.IsTerminal(os.Stdin.Fd())
232 progress = app.config.Config().Options.Progress == nil || *app.config.Config().Options.Progress
233
234 if !hideSpinner && stderrTTY {
235 t := styles.DefaultStyles()
236
237 // Detect background color to set the appropriate color for the
238 // spinner's 'Generating...' text. Without this, that text would be
239 // unreadable in light terminals.
240 hasDarkBG := true
241 if f, ok := output.(*os.File); ok && stdinTTY && stdoutTTY {
242 hasDarkBG = lipgloss.HasDarkBackground(os.Stdin, f)
243 }
244 defaultFG := lipgloss.LightDark(hasDarkBG)(charmtone.Pepper, t.FgBase)
245
246 spinner = format.NewSpinner(ctx, cancel, anim.Settings{
247 Size: 10,
248 Label: "Generating",
249 LabelColor: defaultFG,
250 GradColorA: t.Primary,
251 GradColorB: t.Secondary,
252 CycleColors: true,
253 })
254 spinner.Start()
255 }
256
257 // Helper function to stop spinner once.
258 stopSpinner := func() {
259 if !hideSpinner && spinner != nil {
260 spinner.Stop()
261 spinner = nil
262 }
263 }
264
265 // Wait for MCP initialization to complete before reading MCP tools.
266 if err := mcp.WaitForInit(ctx); err != nil {
267 return fmt.Errorf("failed to wait for MCP initialization: %w", err)
268 }
269
270 // force update of agent models before running so mcp tools are loaded
271 app.AgentCoordinator.UpdateModels(ctx)
272
273 defer stopSpinner()
274
275 sess, err := app.resolveSession(ctx, continueSessionID, useLast)
276 if err != nil {
277 return fmt.Errorf("failed to create session for non-interactive mode: %w", err)
278 }
279
280 if continueSessionID != "" || useLast {
281 slog.Info("Continuing session for non-interactive run", "session_id", sess.ID)
282 } else {
283 slog.Info("Created session for non-interactive run", "session_id", sess.ID)
284 }
285
286 // Automatically approve all permission requests for this non-interactive
287 // session.
288 app.Permissions.AutoApproveSession(sess.ID)
289
290 type response struct {
291 result *fantasy.AgentResult
292 err error
293 }
294 done := make(chan response, 1)
295
296 go func(ctx context.Context, sessionID, prompt string) {
297 result, err := app.AgentCoordinator.Run(ctx, sess.ID, prompt)
298 if err != nil {
299 done <- response{
300 err: fmt.Errorf("failed to start agent processing stream: %w", err),
301 }
302 return
303 }
304 done <- response{
305 result: result,
306 }
307 }(ctx, sess.ID, prompt)
308
309 messageEvents := app.Messages.Subscribe(ctx)
310 messageReadBytes := make(map[string]int)
311 var printed bool
312
313 defer func() {
314 if progress && stderrTTY {
315 _, _ = fmt.Fprintf(os.Stderr, ansi.ResetProgressBar)
316 }
317
318 // Always print a newline at the end. If output is a TTY this will
319 // prevent the prompt from overwriting the last line of output.
320 _, _ = fmt.Fprintln(output)
321 }()
322
323 for {
324 if progress && stderrTTY {
325 // HACK: Reinitialize the terminal progress bar on every iteration
326 // so it doesn't get hidden by the terminal due to inactivity.
327 _, _ = fmt.Fprintf(os.Stderr, ansi.SetIndeterminateProgressBar)
328 }
329
330 select {
331 case result := <-done:
332 stopSpinner()
333 if result.err != nil {
334 if errors.Is(result.err, context.Canceled) || errors.Is(result.err, agent.ErrRequestCancelled) {
335 slog.Debug("Non-interactive: agent processing cancelled", "session_id", sess.ID)
336 return nil
337 }
338 return fmt.Errorf("agent processing failed: %w", result.err)
339 }
340 return nil
341
342 case event := <-messageEvents:
343 msg := event.Payload
344 if msg.SessionID == sess.ID && msg.Role == message.Assistant && len(msg.Parts) > 0 {
345 stopSpinner()
346
347 content := msg.Content().String()
348 readBytes := messageReadBytes[msg.ID]
349
350 if len(content) < readBytes {
351 slog.Error("Non-interactive: message content is shorter than read bytes", "message_length", len(content), "read_bytes", readBytes)
352 return fmt.Errorf("message content is shorter than read bytes: %d < %d", len(content), readBytes)
353 }
354
355 part := content[readBytes:]
356 // Trim leading whitespace. Sometimes the LLM includes leading
357 // formatting and intentation, which we don't want here.
358 if readBytes == 0 {
359 part = strings.TrimLeft(part, " \t")
360 }
361 // Ignore initial whitespace-only messages.
362 if printed || strings.TrimSpace(part) != "" {
363 printed = true
364 fmt.Fprint(output, part)
365 }
366 messageReadBytes[msg.ID] = len(content)
367 }
368
369 case <-ctx.Done():
370 stopSpinner()
371 return ctx.Err()
372 }
373 }
374}
375
376func (app *App) UpdateAgentModel(ctx context.Context) error {
377 if app.AgentCoordinator == nil {
378 return fmt.Errorf("agent configuration is missing")
379 }
380 return app.AgentCoordinator.UpdateModels(ctx)
381}
382
383// overrideModelsForNonInteractive parses the model strings and temporarily
384// overrides the model configurations, then rebuilds the agent.
385// Format: "model-name" (searches all providers) or "provider/model-name".
386// Model matching is case-insensitive.
387// If largeModel is provided but smallModel is not, the small model defaults to
388// the provider's default small model.
389func (app *App) overrideModelsForNonInteractive(ctx context.Context, largeModel, smallModel string) error {
390 providers := app.config.Config().Providers.Copy()
391
392 largeMatches, smallMatches, err := findModels(providers, largeModel, smallModel)
393 if err != nil {
394 return err
395 }
396
397 var largeProviderID string
398
399 // Override large model.
400 if largeModel != "" {
401 found, err := validateMatches(largeMatches, largeModel, "large")
402 if err != nil {
403 return err
404 }
405 largeProviderID = found.provider
406 slog.Info("Overriding large model for non-interactive run", "provider", found.provider, "model", found.modelID)
407 app.config.Config().Models[config.SelectedModelTypeLarge] = config.SelectedModel{
408 Provider: found.provider,
409 Model: found.modelID,
410 }
411 }
412
413 // Override small model.
414 switch {
415 case smallModel != "":
416 found, err := validateMatches(smallMatches, smallModel, "small")
417 if err != nil {
418 return err
419 }
420 slog.Info("Overriding small model for non-interactive run", "provider", found.provider, "model", found.modelID)
421 app.config.Config().Models[config.SelectedModelTypeSmall] = config.SelectedModel{
422 Provider: found.provider,
423 Model: found.modelID,
424 }
425
426 case largeModel != "":
427 // No small model specified, but large model was - use provider's default.
428 smallCfg := app.GetDefaultSmallModel(largeProviderID)
429 app.config.Config().Models[config.SelectedModelTypeSmall] = smallCfg
430 }
431
432 return app.AgentCoordinator.UpdateModels(ctx)
433}
434
435// GetDefaultSmallModel returns the default small model for the given
436// provider. Falls back to the large model if no default is found.
437func (app *App) GetDefaultSmallModel(providerID string) config.SelectedModel {
438 cfg := app.config.Config()
439 largeModelCfg := cfg.Models[config.SelectedModelTypeLarge]
440
441 // Find the provider in the known providers list to get its default small model.
442 knownProviders, _ := config.Providers(cfg)
443 var knownProvider *catwalk.Provider
444 for _, p := range knownProviders {
445 if string(p.ID) == providerID {
446 knownProvider = &p
447 break
448 }
449 }
450
451 // For unknown/local providers, use the large model as small.
452 if knownProvider == nil {
453 slog.Warn("Using large model as small model for unknown provider", "provider", providerID, "model", largeModelCfg.Model)
454 return largeModelCfg
455 }
456
457 defaultSmallModelID := knownProvider.DefaultSmallModelID
458 model := cfg.GetModel(providerID, defaultSmallModelID)
459 if model == nil {
460 slog.Warn("Default small model not found, using large model", "provider", providerID, "model", largeModelCfg.Model)
461 return largeModelCfg
462 }
463
464 slog.Info("Using provider default small model", "provider", providerID, "model", defaultSmallModelID)
465 return config.SelectedModel{
466 Provider: providerID,
467 Model: defaultSmallModelID,
468 MaxTokens: model.DefaultMaxTokens,
469 ReasoningEffort: model.DefaultReasoningEffort,
470 }
471}
472
473func (app *App) setupEvents() {
474 ctx, cancel := context.WithCancel(app.globalCtx)
475 app.eventsCtx = ctx
476 setupSubscriber(ctx, app.serviceEventsWG, "sessions", app.Sessions.Subscribe, app.events)
477 setupSubscriber(ctx, app.serviceEventsWG, "messages", app.Messages.Subscribe, app.events)
478 setupSubscriber(ctx, app.serviceEventsWG, "permissions", app.Permissions.Subscribe, app.events)
479 setupSubscriber(ctx, app.serviceEventsWG, "permissions-notifications", app.Permissions.SubscribeNotifications, app.events)
480 setupSubscriber(ctx, app.serviceEventsWG, "history", app.History.Subscribe, app.events)
481 setupSubscriber(ctx, app.serviceEventsWG, "agent-notifications", app.agentNotifications.Subscribe, app.events)
482 setupSubscriber(ctx, app.serviceEventsWG, "mcp", mcp.SubscribeEvents, app.events)
483 setupSubscriber(ctx, app.serviceEventsWG, "lsp", SubscribeLSPEvents, app.events)
484 cleanupFunc := func(context.Context) error {
485 cancel()
486 app.serviceEventsWG.Wait()
487 return nil
488 }
489 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
490}
491
492const subscriberSendTimeout = 2 * time.Second
493
494func setupSubscriber[T any](
495 ctx context.Context,
496 wg *sync.WaitGroup,
497 name string,
498 subscriber func(context.Context) <-chan pubsub.Event[T],
499 outputCh chan<- tea.Msg,
500) {
501 wg.Go(func() {
502 subCh := subscriber(ctx)
503 sendTimer := time.NewTimer(0)
504 <-sendTimer.C
505 defer sendTimer.Stop()
506
507 for {
508 select {
509 case event, ok := <-subCh:
510 if !ok {
511 slog.Debug("Subscription channel closed", "name", name)
512 return
513 }
514 var msg tea.Msg = event
515 if !sendTimer.Stop() {
516 select {
517 case <-sendTimer.C:
518 default:
519 }
520 }
521 sendTimer.Reset(subscriberSendTimeout)
522
523 select {
524 case outputCh <- msg:
525 case <-sendTimer.C:
526 slog.Debug("Message dropped due to slow consumer", "name", name)
527 case <-ctx.Done():
528 slog.Debug("Subscription cancelled", "name", name)
529 return
530 }
531 case <-ctx.Done():
532 slog.Debug("Subscription cancelled", "name", name)
533 return
534 }
535 }
536 })
537}
538
539func (app *App) InitCoderAgent(ctx context.Context) error {
540 coderAgentCfg := app.config.Config().Agents[config.AgentCoder]
541 if coderAgentCfg.ID == "" {
542 return fmt.Errorf("coder agent configuration is missing")
543 }
544 var err error
545 app.AgentCoordinator, err = agent.NewCoordinator(
546 ctx,
547 app.config,
548 app.Sessions,
549 app.Messages,
550 app.Permissions,
551 app.History,
552 app.FileTracker,
553 app.LSPManager,
554 app.agentNotifications,
555 )
556 if err != nil {
557 slog.Error("Failed to create coder agent", "err", err)
558 return err
559 }
560 return nil
561}
562
563// Subscribe sends events to the TUI as tea.Msgs.
564func (app *App) Subscribe(program *tea.Program) {
565 defer log.RecoverPanic("app.Subscribe", func() {
566 slog.Info("TUI subscription panic: attempting graceful shutdown")
567 program.Quit()
568 })
569
570 app.tuiWG.Add(1)
571 tuiCtx, tuiCancel := context.WithCancel(app.globalCtx)
572 app.cleanupFuncs = append(app.cleanupFuncs, func(context.Context) error {
573 slog.Debug("Cancelling TUI message handler")
574 tuiCancel()
575 app.tuiWG.Wait()
576 return nil
577 })
578 defer app.tuiWG.Done()
579
580 for {
581 select {
582 case <-tuiCtx.Done():
583 slog.Debug("TUI message handler shutting down")
584 return
585 case msg, ok := <-app.events:
586 if !ok {
587 slog.Debug("TUI message channel closed")
588 return
589 }
590 program.Send(msg)
591 }
592 }
593}
594
595// Shutdown performs a graceful shutdown of the application.
596func (app *App) Shutdown() {
597 start := time.Now()
598 defer func() { slog.Debug("Shutdown took " + time.Since(start).String()) }()
599
600 // First, cancel all agents and wait for them to finish. This must complete
601 // before closing the DB so agents can finish writing their state.
602 if app.AgentCoordinator != nil {
603 app.AgentCoordinator.CancelAll()
604 }
605
606 // Now run remaining cleanup tasks in parallel.
607 var wg sync.WaitGroup
608
609 // Shared shutdown context for all timeout-bounded cleanup.
610 shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
611 defer cancel()
612
613 // Send exit event
614 wg.Go(func() {
615 event.AppExited()
616 })
617
618 // Kill all background shells.
619 wg.Go(func() {
620 shell.GetBackgroundShellManager().KillAll(shutdownCtx)
621 })
622
623 // Shutdown all LSP clients.
624 wg.Go(func() {
625 app.LSPManager.KillAll(shutdownCtx)
626 })
627
628 // Call all cleanup functions.
629 for _, cleanup := range app.cleanupFuncs {
630 if cleanup != nil {
631 wg.Go(func() {
632 if err := cleanup(shutdownCtx); err != nil {
633 slog.Error("Failed to cleanup app properly on shutdown", "error", err)
634 }
635 })
636 }
637 }
638 wg.Wait()
639}
640
641// checkForUpdates checks for available updates.
642func (app *App) checkForUpdates(ctx context.Context) {
643 checkCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
644 defer cancel()
645
646 info, err := update.Check(checkCtx, version.Version, update.Default)
647 if err != nil || !info.Available() {
648 return
649 }
650 app.events <- UpdateAvailableMsg{
651 CurrentVersion: info.Current,
652 LatestVersion: info.Latest,
653 IsDevelopment: info.IsDevelopment(),
654 }
655}