| 622 | } |
| 623 | |
| 624 | func (s *mcpServer) startWatcher(ctx context.Context, inv *serpent.Invocation) { |
| 625 | go func() { |
| 626 | for retrier := retry.New(time.Second, 30*time.Second); retrier.Wait(ctx); { |
| 627 | eventsCh, errCh, err := s.aiAgentAPIClient.SubscribeEvents(ctx) |
| 628 | if err == nil { |
| 629 | retrier.Reset() |
| 630 | loop: |
| 631 | for { |
| 632 | select { |
| 633 | case <-ctx.Done(): |
| 634 | return |
| 635 | case event := <-eventsCh: |
| 636 | switch ev := event.(type) { |
| 637 | case agentapi.EventStatusChange: |
| 638 | state := codersdk.WorkspaceAppStatusStateWorking |
| 639 | if ev.Status == agentapi.StatusStable { |
| 640 | state = codersdk.WorkspaceAppStatusStateIdle |
| 641 | } |
| 642 | err := s.queue.Push(taskReport{ |
| 643 | state: state, |
| 644 | }) |
| 645 | if err != nil { |
| 646 | cliui.Warnf(inv.Stderr, "Failed to queue update: %s", err) |
| 647 | return |
| 648 | } |
| 649 | case agentapi.EventMessageUpdate: |
| 650 | if ev.Role == agentapi.RoleUser { |
| 651 | err := s.queue.Push(taskReport{ |
| 652 | messageID: &ev.Id, |
| 653 | state: codersdk.WorkspaceAppStatusStateWorking, |
| 654 | }) |
| 655 | if err != nil { |
| 656 | cliui.Warnf(inv.Stderr, "Failed to queue update: %s", err) |
| 657 | return |
| 658 | } |
| 659 | } |
| 660 | } |
| 661 | case err := <-errCh: |
| 662 | if !errors.Is(err, context.Canceled) { |
| 663 | cliui.Warnf(inv.Stderr, "Received error from screen event watcher: %s", err) |
| 664 | } |
| 665 | break loop |
| 666 | } |
| 667 | } |
| 668 | } else { |
| 669 | cliui.Warnf(inv.Stderr, "Failed to watch screen events: %s", err) |
| 670 | } |
| 671 | } |
| 672 | }() |
| 673 | } |
| 674 | |
| 675 | func (s *mcpServer) startServer(ctx context.Context, inv *serpent.Invocation, instructions string, allowedTools []string) error { |
| 676 | cliui.Infof(inv.Stderr, "Starting MCP server") |