+// daemonServe runs the loop that start makes and serves the clients of listener
+// until ctx ends or the loop fails. It closes listener. It runs whatsapp, which
+// logs its batches and notes in the loop and tells all clients its notices, and
+// a failure of whatsapp stops the agent. It runs schedules, which log their
+// runs in the loop, and a failure of schedules stops the agent too. When bot is
+// not nil, it runs the bot too, and a failure of the bot stops the agent.
+func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ hub := &DaemonHub{
+ whatsapp: whatsapp,
+ clients: make(map[*DaemonClient]bool),
+ }
+
+ notify := hub.Broadcast
+ if bot != nil {
+ notify = func(event Event) {
+ hub.Broadcast(event)
+ bot.Notify(event)
+ }
+ }
+
+ l, err := start(notify)
+ if err != nil {
+ return errors.Join(err, listener.Close())
+ }
+
+ var workers sync.WaitGroup
+ workers.Add(2)
+ go func() {
+ defer workers.Done()
+
+ hub.Accept(ctx, listener, l)
+ }()
+
+ var whatsappErr error
+ go func() {
+ defer workers.Done()
+
+ whatsappErr = whatsapp.Run(ctx, l, notify)
+ cancel()
+ }()
+
+ var schedulesErr error
+ workers.Add(1)
+ go func() {
+ defer workers.Done()
+
+ schedulesErr = schedules.Run(ctx, l)
+ cancel()
+ }()
+
+ var botErr error
+ if bot != nil {
+ workers.Add(1)
+ go func() {
+ defer workers.Done()
+
+ botErr = bot.Run(ctx, l, whatsapp)
+ cancel()
+ }()
+ }
+
+ err = l.Run(ctx)
+ if errors.Is(err, context.Canceled) {
+ err = nil
+ }
+
+ cancel()
+ err = errors.Join(err, listener.Close())
+ hub.CloseAll()
+ workers.Wait()
+
+ return errors.Join(err, whatsappErr, schedulesErr, botErr)
+}
+
+// daemonRunRequest runs one request of a client. The answers of a WhatsApp
+// command go to answer.
+func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, answer func(text string)) error {
+ switch request.Type {
+ case "send":
+ if request.Text == "" {
+ return errors.New("empty message")
+ }
+
+ _, err := l.Submit(terminalSource, request.Text)
+
+ return err
+ case "interrupt":
+ if !l.Interrupt() {
+ return errors.New("no turn runs")
+ }
+
+ return nil
+ case "whatsapp":
+ whatsapp.Command(request.Text, answer)
+
+ return nil
+ case "stop":
+ found, err := l.StopSubagent(request.Text)
+ if err == nil && !found {
+ err = fmt.Errorf("no subagent %s runs", request.Text)
+ }
+
+ return err
+ default:
+ return fmt.Errorf("unknown request %q", request.Type)
+ }
+}
+
+// daemonLogger returns the logger of the part of the daemon with prefix, such
+// as "telegram: ", on out. The journal adds the time to each line and sets
+// JOURNAL_STREAM for a service, so then the lines have no time stamp. Otherwise
+// each line has the standard time stamp.
+func daemonLogger(out io.Writer, prefix string) *log.Logger {
+ if os.Getenv("JOURNAL_STREAM") != "" {
+ return log.New(out, prefix, 0)
projects / onechat / blobdiff