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/skills"
38 "github.com/charmbracelet/crush/internal/ui/anim"
39 "github.com/charmbracelet/crush/internal/ui/styles"
40 "github.com/charmbracelet/crush/internal/update"
41 "github.com/charmbracelet/crush/internal/version"
42 "github.com/charmbracelet/x/ansi"
43 "github.com/charmbracelet/x/exp/charmtone"
44 "github.com/charmbracelet/x/term"
45)
46
47// UpdateAvailableMsg is sent when a new version is available.
48type UpdateAvailableMsg struct {
49 CurrentVersion string
50 LatestVersion string
51 IsDevelopment bool
52}
53
54type App struct {
55 Sessions session.Service
56 Messages message.Service
57 History history.Service
58 Permissions permission.Service
59 FileTracker filetracker.Service
60
61 AgentCoordinator agent.Coordinator
62
63 LSPManager *lsp.Manager
64
65 config *config.ConfigStore
66
67 serviceEventsWG *sync.WaitGroup
68 eventsCtx context.Context
69 events *pubsub.Broker[tea.Msg]
70 tuiWG *sync.WaitGroup
71
72 // global context and cleanup functions
73 globalCtx context.Context
74 cleanupFuncs []func(context.Context) error
75 agentNotifications *pubsub.Broker[notify.Notification]
76}
77
78// New initializes a new application instance.
79func New(ctx context.Context, conn *sql.DB, store *config.ConfigStore) (*App, error) {
80 q := db.New(conn)
81 sessions := session.NewService(q, conn)
82 messages := message.NewService(q)
83 files := history.NewService(q, conn)
84 cfg := store.Config()
85 skipPermissionsRequests := store.Overrides().SkipPermissionRequests
86 var allowedTools []string
87 if cfg.Permissions != nil && cfg.Permissions.AllowedTools != nil {
88 allowedTools = cfg.Permissions.AllowedTools
89 }
90
91 app := &App{
92 Sessions: sessions,
93 Messages: messages,
94 History: files,
95 Permissions: permission.NewPermissionService(store.WorkingDir(), skipPermissionsRequests, allowedTools),
96 FileTracker: filetracker.NewService(q),
97 LSPManager: lsp.NewManager(store),
98
99 globalCtx: ctx,
100
101 config: store,
102
103 events: pubsub.NewBroker[tea.Msg](),
104 serviceEventsWG: &sync.WaitGroup{},
105 tuiWG: &sync.WaitGroup{},
106 agentNotifications: pubsub.NewBroker[notify.Notification](),
107 }
108
109 app.setupEvents()
110
111 // Check for updates in the background.
112 go app.checkForUpdates(ctx)
113
114 go mcp.Initialize(ctx, app.Permissions, store)
115
116 // Release the shared database connection on shutdown. The pool
117 // closes the underlying *sql.DB when the last reference is released.
118 dataDir := cfg.Options.DataDirectory
119 app.cleanupFuncs = append(
120 app.cleanupFuncs,
121 func(context.Context) error { return db.Release(dataDir) },
122 func(ctx context.Context) error { return mcp.Close(ctx) },
123 )
124
125 // TODO: remove the concept of agent config, most likely.
126 if !cfg.IsConfigured() {
127 slog.Warn("No agent configuration found")
128 return app, nil
129 }
130 if err := app.InitCoderAgent(ctx); err != nil {
131 return nil, fmt.Errorf("failed to initialize coder agent: %w", err)
132 }
133
134 // Set up callback for LSP state updates.
135 app.LSPManager.SetCallback(func(name string, client *lsp.Client) {
136 if client == nil {
137 updateLSPState(name, lsp.StateUnstarted, nil, nil, 0)
138 return
139 }
140 client.SetDiagnosticsCallback(updateLSPDiagnostics)
141 updateLSPState(name, client.GetServerState(), nil, client, 0)
142 })
143 go app.LSPManager.TrackConfigured()
144
145 return app, nil
146}
147
148// Config returns the pure-data configuration.
149func (app *App) Config() *config.Config {
150 return app.config.Config()
151}
152
153// Store returns the config store.
154func (app *App) Store() *config.ConfigStore {
155 return app.config
156}
157
158// Events returns a per-caller subscription channel for application events.
159// Each caller receives its own channel; all callers receive every event.
160func (app *App) Events(ctx context.Context) <-chan pubsub.Event[tea.Msg] {
161 return app.events.Subscribe(ctx)
162}
163
164// SendEvent publishes a message to all event subscribers.
165func (app *App) SendEvent(msg tea.Msg) {
166 app.events.Publish(pubsub.UpdatedEvent, msg)
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.ThemeForProvider(app.config.Config().Models[config.SelectedModelTypeLarge].Provider)
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.WorkingLabelColor)
245
246 spinner = format.NewSpinner(ctx, cancel, anim.Settings{
247 Size: 10,
248 Label: "Generating",
249 LabelColor: defaultFG,
250 GradColorA: t.WorkingGradFromColor,
251 GradColorB: t.WorkingGradToColor,
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 setupSubscriber(ctx, app.serviceEventsWG, "skills", skills.SubscribeEvents, app.events)
485 cleanupFunc := func(context.Context) error {
486 cancel()
487 app.serviceEventsWG.Wait()
488 app.events.Shutdown()
489 return nil
490 }
491 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
492}
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 broker *pubsub.Broker[tea.Msg],
500) {
501 wg.Go(func() {
502 subCh := subscriber(ctx)
503 for {
504 select {
505 case event, ok := <-subCh:
506 if !ok {
507 slog.Debug("Subscription channel closed", "name", name)
508 return
509 }
510 broker.Publish(pubsub.UpdatedEvent, tea.Msg(event))
511 case <-ctx.Done():
512 slog.Debug("Subscription cancelled", "name", name)
513 return
514 }
515 }
516 })
517}
518
519func (app *App) InitCoderAgent(ctx context.Context) error {
520 coderAgentCfg := app.config.Config().Agents[config.AgentCoder]
521 if coderAgentCfg.ID == "" {
522 return fmt.Errorf("coder agent configuration is missing")
523 }
524 var err error
525 app.AgentCoordinator, err = agent.NewCoordinator(
526 ctx,
527 app.config,
528 app.Sessions,
529 app.Messages,
530 app.Permissions,
531 app.History,
532 app.FileTracker,
533 app.LSPManager,
534 app.agentNotifications,
535 )
536 if err != nil {
537 slog.Error("Failed to create coder agent", "err", err)
538 return err
539 }
540 return nil
541}
542
543// Subscribe sends events to the TUI as tea.Msgs.
544func (app *App) Subscribe(program *tea.Program) {
545 defer log.RecoverPanic("app.Subscribe", func() {
546 slog.Info("TUI subscription panic: attempting graceful shutdown")
547 program.Quit()
548 })
549
550 app.tuiWG.Add(1)
551 tuiCtx, tuiCancel := context.WithCancel(app.globalCtx)
552 app.cleanupFuncs = append(app.cleanupFuncs, func(context.Context) error {
553 slog.Debug("Cancelling TUI message handler")
554 tuiCancel()
555 app.tuiWG.Wait()
556 return nil
557 })
558 defer app.tuiWG.Done()
559
560 events := app.events.Subscribe(tuiCtx)
561 for {
562 select {
563 case <-tuiCtx.Done():
564 slog.Debug("TUI message handler shutting down")
565 return
566 case ev, ok := <-events:
567 if !ok {
568 slog.Debug("TUI message channel closed")
569 return
570 }
571 program.Send(ev.Payload)
572 }
573 }
574}
575
576// Shutdown performs a graceful shutdown of the application.
577func (app *App) Shutdown() {
578 start := time.Now()
579 defer func() { slog.Debug("Shutdown took " + time.Since(start).String()) }()
580
581 // First, cancel all agents and wait for them to finish. This must complete
582 // before closing the DB so agents can finish writing their state.
583 if app.AgentCoordinator != nil {
584 app.AgentCoordinator.CancelAll()
585 }
586
587 // Now run remaining cleanup tasks in parallel.
588 var wg sync.WaitGroup
589
590 // Shared shutdown context for all timeout-bounded cleanup.
591 shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
592 defer cancel()
593
594 // Send exit event
595 wg.Go(func() {
596 event.AppExited()
597 })
598
599 // Kill all background shells.
600 wg.Go(func() {
601 shell.GetBackgroundShellManager().KillAll(shutdownCtx)
602 })
603
604 // Shutdown all LSP clients.
605 wg.Go(func() {
606 app.LSPManager.KillAll(shutdownCtx)
607 })
608
609 // Call all cleanup functions.
610 for _, cleanup := range app.cleanupFuncs {
611 if cleanup != nil {
612 wg.Go(func() {
613 if err := cleanup(shutdownCtx); err != nil {
614 slog.Error("Failed to cleanup app properly on shutdown", "error", err)
615 }
616 })
617 }
618 }
619 wg.Wait()
620}
621
622// checkForUpdates checks for available updates.
623func (app *App) checkForUpdates(ctx context.Context) {
624 checkCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
625 defer cancel()
626
627 info, err := update.Check(checkCtx, version.Version, update.Default)
628 if err != nil || !info.Available() {
629 return
630 }
631 app.events.Publish(pubsub.UpdatedEvent, UpdateAvailableMsg{
632 CurrentVersion: info.Current,
633 LatestVersion: info.Latest,
634 IsDevelopment: info.IsDevelopment(),
635 })
636}