+// 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. Each
+// child runs its own loop and schedules with no client events. A failed child
+// stops the daemon, and backups lock all loops before they copy data.
+func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error), children ...*DaemonAgent) 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())
+ }
+
+ loops := []*TurnLoop{l}
+ for _, child := range children {
+ loop, err := child.Loop(nil, func(Event) {
+ })
+ if err != nil {
+ return errors.Join(fmt.Errorf("start child %q: %w", child.agent.config.Name, err), listener.Close())
+ }
+
+ loops = append(loops, loop)
+ }
+
+ messages, err := openDaemonMessages(loops)
+ if err != nil {
+ return errors.Join(fmt.Errorf("open agent messages: %w", err), listener.Close())
+ }
+
+ hub.snapshot = func(ctx context.Context, snapshot func() error) error {
+ return messages.SnapshotAgents(ctx, snapshot)
+ }
+
+ runtime := &DaemonChildren{
+ ctx: ctx,
+ cancel: cancel,
+ root: l,
+ messages: messages,
+ }
+ l.tools.children = runtime
+
+ var workers sync.WaitGroup
+
+ childErrors := make([]error, len(children))
+ for i, child := range children {
+ workers.Add(1)
+
+ go func() {
+ defer workers.Done()
+
+ childErrors[i] = daemonRunChild(ctx, loops[i+1], child.schedules)
+ if childErrors[i] != nil {
+ child.logger.Error("child stopped", "err", childErrors[i])
+ }
+
+ cancel()
+ }()
+ }
+
+ 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()
+ }()
projects / onechat / blobdiff