Deliver messages between parent and child agents

[onechat] / daemon.go

diff --git a/daemon.go b/daemon.go
index 68073d0be5342b81eb0dc93203226cb0ee063954..958d6f7bd2283fe8b2796c14ae345e7b5f14e604 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -15,45 +15,22 @@ import (
 )
 
 const (
 )
 
 const (

-       // agentDirName is the directory of the root agent in the data directory.
-       agentDirName = "agent"
-
-       // socketFileName is the Unix socket of the daemon in the data directory.
-       socketFileName = "onechat.sock"
-
-       // tokenLockFileName is the lock file of the token file in the data
+       // daemonAgentDirName is the directory of the root agent in the data

        // directory.
        // directory.

-       tokenLockFileName = "auth.lock"
-
-       // requestBytes is the most bytes of one request line of a client.
-       requestBytes = 16 << 20
-
-       // clientEvents is the number of events that can wait for a slow client
-       // before the daemon drops it.
-       clientEvents = 1024
+       daemonAgentDirName = "agent"

 
 

-       // socketPathBytes is the longest path of a Unix socket on Linux.
-       socketPathBytes = 107
-
-       // terminalSource is the source of the user messages from the clients of
-       // the socket.
-       terminalSource = "terminal"
+       // daemonClientEvents is the number of events that can wait for a slow
+       // client before the daemon drops it.
+       daemonClientEvents = 1024

 )
 
 )
 

-// errTokensInUse tells that another onechat process uses the token file.
-var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon")
-
-// clientRequest is one line from a client: "send" with the text of a user
-// message, or "interrupt".
-type clientRequest struct {
-       Type string `json:"type"`
-       Text string `json:"text,omitempty"`
-}
-
-// daemon runs "onechat daemon": it owns the root agent and its files and serves
-// the clients on the Unix socket in the data directory until ctx ends. It holds
-// the lock of the token file, so it is the only process that refreshes the
-// tokens.
+// daemon runs "onechat daemon": it owns the root, its children, and their
+// files, and serves the clients on the Unix socket in the data directory until
+// ctx ends. When telegram.json exists, it also runs the Telegram bot. It also
+// keeps a linked WhatsApp device connected. It holds the lock of the token
+// file, so it is the only process that refreshes the tokens and opens the
+// WhatsApp session. Its log lines go to out, through the loggers of
+// daemonLogger.

 func daemon(ctx context.Context, out io.Writer) (err error) {
        dir, err := dataDir()
        if err != nil {
 func daemon(ctx context.Context, out io.Writer) (err error) {
        dir, err := dataDir()
        if err != nil {


@@ -73,6 +50,11 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                return err
        }
 
                return err
        }
 

+       botSettings, hasBot, err := loadTelegramConfig(dir)
+       if err != nil {
+               return err
+       }
+

        tokens, err := lockTokens(dir)
        if err != nil {
                return err
        tokens, err := lockTokens(dir)
        if err != nil {
                return err


@@ -82,112 +64,95 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                err = errors.Join(err, tokens.Close())
        }()
 
                err = errors.Join(err, tokens.Close())
        }()
 

-       address, auth, err := modelEndpoint(config, dir)
-       if err != nil {
-               return err
-       }
-
-       agent, err := openAgent(filepath.Join(dir, agentDirName))
-       if err != nil {
-               return err
-       }
+       whatsappSetDeviceProps()

 
 

-       client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
-       memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
+       logger := daemonLogger(out, "onechat: ")
+       whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)

        if err != nil {
                return err
        }
 
        defer func() {
        if err != nil {
                return err
        }
 
        defer func() {

-               err = errors.Join(err, memory.Close())
+               err = errors.Join(err, whatsapp.Close())

        }()
 
        }()
 

-       listener, err := listenSocket(socket)
+       address, auth, err := modelEndpoint(config, dir)

        if err != nil {
                return err
        }
 
        if err != nil {
                return err
        }
 

-       _, err = fmt.Fprintf(out, "onechat: %s on %s, %s at %s effort, in %s\n", agent.config.Name, listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir)
+       agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger)

        if err != nil {
        if err != nil {

-               return errors.Join(err, listener.Close())
+               return err

        }
 
        }
 

-       return serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) {
-               return newLoop(agent, memory, client, time.Now, notify)
-       })
-}
+       defer func() {
+               err = errors.Join(err, agents.Close())
+       }()

 
 

-// serveAgent runs the loop that start makes and serves the clients of
-// listener until ctx ends or the loop fails. It closes listener.
-func serveAgent(ctx context.Context, listener net.Listener, start func(notify func(Event)) (*loop, error)) error {
-       ctx, cancel := context.WithCancel(ctx)
-       defer cancel()
+       root := agents[0]
+       agent := root.agent
+       schedules := root.schedules

 
 

-       hub := &hub{
-               clients: make(map[*hubClient]bool),
+       listener, err := daemonListenSocket(socket)
+       if err != nil {
+               return err

        }
 
        }
 

-       l, err := start(hub.Broadcast)
-       if err != nil {
-               return errors.Join(err, listener.Close())
+       linked := "not linked"
+       if whatsapp.isLinked() {
+               linked = "linked"

        }
 
        }
 

-       var workers sync.WaitGroup
-       workers.Add(1)
-       go func() {
-               defer workers.Done()
+       logger.Info("ready", "version", version, "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len(), "children", agents.ChildNames())

 
 

-               hub.Accept(ctx, listener, l)
-       }()
+       var telegram *TelegramBot

 
 

-       err = l.Run(ctx)
-       cancel()
-       err = errors.Join(err, listener.Close())
-       hub.CloseAll()
-       workers.Wait()
+       if hasBot {
+               telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))

 
 

-       if errors.Is(err, context.Canceled) {
-               return nil
+               logger.Info("Telegram bot started")

        }
 
        }
 

-       return err
+       return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+               return root.Loop(whatsapp, notify)
+       }, agents[1:]...)

 }
 
 }
 

-// hub holds the connected clients and sends each event to all of them.
-type hub struct {
+// DaemonHub holds the connected clients and sends each event to all of them. It
+// runs the WhatsApp commands of the clients on whatsapp. snapshot locks all
+// agents and subagents during a backup copy.
+type DaemonHub struct {
+       whatsapp *WhatsApp
+       snapshot func(context.Context, func() error) error
+

        mutex   sync.Mutex
        mutex   sync.Mutex

-       clients map[*hubClient]bool
+       clients map[*DaemonClient]bool

        workers sync.WaitGroup
 }
 
        workers sync.WaitGroup
 }
 

-// hubClient is one connected client. Its events wait in a channel for the
-// writer, so a slow client does not stop the agent.
-type hubClient struct {
-       conn   net.Conn
-       events chan Event
-       once   sync.Once
-}
-

 // Accept serves each client of listener until the listener closes. A new
 // client gets a ready event with the name of the agent, and a busy event when a
 // turn runs.
 // Accept serves each client of listener until the listener closes. A new
 // client gets a ready event with the name of the agent, and a busy event when a
 // turn runs.

-func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
+func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLoop) {

        for {
                conn, err := listener.Accept()
                if err != nil {
                        return
                }
 
        for {
                conn, err := listener.Accept()
                if err != nil {
                        return
                }
 

-               client := &hubClient{
+               client := &DaemonClient{

                        conn:   conn,
                        conn:   conn,

-                       events: make(chan Event, clientEvents),
+                       events: make(chan Event, daemonClientEvents),

                }
 
                h.mutex.Lock()
                }
 
                h.mutex.Lock()

+

                h.clients[client] = true
                h.mutex.Unlock()
 
                h.workers.Add(2)
                h.clients[client] = true
                h.mutex.Unlock()
 
                h.workers.Add(2)

+

                go h.write(client)
                go h.read(ctx, client, l)
 
                go h.write(client)
                go h.read(ctx, client, l)
 


@@ -205,7 +170,7 @@ func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
 }
 
 // write sends the events of client as JSON lines until its channel closes.
 }
 
 // write sends the events of client as JSON lines until its channel closes.

-func (h *hub) write(client *hubClient) {
+func (h *DaemonHub) write(client *DaemonClient) {

        defer h.workers.Done()
 
        encoder := json.NewEncoder(client.conn)
        defer h.workers.Done()
 
        encoder := json.NewEncoder(client.conn)


@@ -218,17 +183,36 @@ func (h *hub) write(client *hubClient) {
 }
 
 // read runs the requests of client until it closes its connection.
 }
 
 // read runs the requests of client until it closes its connection.

-func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
+func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) {

        defer h.workers.Done()
        defer h.drop(client)
 
        scanner := bufio.NewScanner(client.conn)
        defer h.workers.Done()
        defer h.drop(client)
 
        scanner := bufio.NewScanner(client.conn)

-       scanner.Buffer(make([]byte, 64*1024), requestBytes)
+       scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
+

        for ctx.Err() == nil && scanner.Scan() {
        for ctx.Err() == nil && scanner.Scan() {

-               var request clientRequest
+               var request SocketRequest
+

                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {
                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {

-                       err = runRequest(l, request)
+                       if request.Type == "backup" {
+                               var size int64
+
+                               size, err = backupData(ctx, filepath.Dir(l.agent.dir), request.Text, h.whatsapp, h.snapshot)
+                               if err == nil {
+                                       h.Send(client, Event{
+                                               Kind: "backup",
+                                               Text: fmt.Sprintf("%s (%d bytes)", request.Text, size),
+                                       })
+                               }
+                       } else {
+                               err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
+                                       h.Send(client, Event{
+                                               Kind: "notice",
+                                               Text: text,
+                                       })
+                               })
+                       }

                }
 
                if err != nil {
                }
 
                if err != nil {


@@ -242,15 +226,23 @@ func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
 
 // Send gives event to one client, or drops the client when its events do not
 // fit.
 
 // Send gives event to one client, or drops the client when its events do not
 // fit.

-func (h *hub) Send(client *hubClient, event Event) {
+func (h *DaemonHub) Send(client *DaemonClient, event Event) {

        h.mutex.Lock()
        defer h.mutex.Unlock()
 
        h.send(client, event)
 }
 
        h.mutex.Lock()
        defer h.mutex.Unlock()
 
        h.send(client, event)
 }
 

+// drop removes client and closes its connection.
+func (h *DaemonHub) drop(client *DaemonClient) {
+       h.mutex.Lock()
+       defer h.mutex.Unlock()
+
+       h.remove(client)
+}
+

 // send gives event to client. The caller holds the mutex.
 // send gives event to client. The caller holds the mutex.

-func (h *hub) send(client *hubClient, event Event) {
+func (h *DaemonHub) send(client *DaemonClient, event Event) {

        if !h.clients[client] {
                return
        }
        if !h.clients[client] {
                return
        }


@@ -262,71 +254,52 @@ func (h *hub) send(client *hubClient, event Event) {
        }
 }
 
        }
 }
 

-// Broadcast gives event to every client.
-func (h *hub) Broadcast(event Event) {
-       h.mutex.Lock()
-       defer h.mutex.Unlock()
-
-       for client := range h.clients {
-               h.send(client, event)
-       }
-}
-
-// drop removes client and closes its connection.
-func (h *hub) drop(client *hubClient) {
-       h.mutex.Lock()
-       defer h.mutex.Unlock()
-
-       h.remove(client)
-}
-

 // remove removes client, closes its connection, and ends its writer. The
 // caller holds the mutex.
 // remove removes client, closes its connection, and ends its writer. The
 // caller holds the mutex.

-func (h *hub) remove(client *hubClient) {
+func (h *DaemonHub) remove(client *DaemonClient) {

        delete(h.clients, client)
        client.once.Do(func() {
                close(client.events)
        delete(h.clients, client)
        client.once.Do(func() {
                close(client.events)

+

                _ = client.conn.Close()
        })
 }
 
                _ = client.conn.Close()
        })
 }
 

+// Broadcast gives event to every client.
+func (h *DaemonHub) Broadcast(event Event) {
+       h.mutex.Lock()
+       defer h.mutex.Unlock()
+
+       for client := range h.clients {
+               h.send(client, event)
+       }
+}
+

 // CloseAll drops every client and waits for their readers and writers.
 // CloseAll drops every client and waits for their readers and writers.

-func (h *hub) CloseAll() {
+func (h *DaemonHub) CloseAll() {

        h.mutex.Lock()
        h.mutex.Lock()

+

        for client := range h.clients {
                h.remove(client)
        }
        for client := range h.clients {
                h.remove(client)
        }

+

        h.mutex.Unlock()
 
        h.workers.Wait()
 }
 
        h.mutex.Unlock()
 
        h.workers.Wait()
 }
 

-// runRequest runs one request of a client.
-func runRequest(l *loop, request clientRequest) 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
-       default:
-               return fmt.Errorf("unknown request %q", request.Type)
-       }
+// DaemonClient is one connected client. Its events wait in a channel for the
+// writer, so a slow client does not stop the agent.
+type DaemonClient struct {
+       conn   net.Conn
+       events chan Event
+       once   sync.Once

 }
 
 }
 

-// listenSocket listens on the Unix socket at path, with mode 0600. The caller
-// owns the chat, so a socket file at path is left from a daemon that ended,
-// and listenSocket removes it.
-func listenSocket(path string) (net.Listener, error) {
+// daemonListenSocket listens on the Unix socket at path, with mode 0600. The
+// caller owns the chat, so a socket file at path is left from a daemon that
+// ended, and daemonListenSocket removes it.
+func daemonListenSocket(path string) (net.Listener, error) {

        err := checkSocketPath(path)
        if err != nil {
                return nil, err
        err := checkSocketPath(path)
        if err != nil {
                return nil, err


@@ -350,23 +323,188 @@ func listenSocket(path string) (net.Listener, error) {
        return listener, nil
 }
 
        return listener, nil
 }
 

-// checkSocketPath makes sure that path fits in the address of a Unix socket.
-func checkSocketPath(path string) error {
-       if len(path) > socketPathBytes {
-               return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
+// 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)

        }
 
        }
 

-       return nil
+       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.Snapshot(func() error {
+                       return daemonSnapshot(ctx, loops, snapshot)
+               })
+       }
+
+       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()
+               }()
+       }
+
+       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, errors.Join(childErrors...))

 }
 
 }
 

-// lockTokens takes the lock of the token file in dir. Each process that reads
-// or refreshes the tokens holds it, because a refresh in one process rotates
-// the refresh token of the other and logs it out.
-func lockTokens(dir string) (*os.File, error) {
-       lock, err := lockFile(filepath.Join(dir, tokenLockFileName))
-       if errors.Is(err, errLocked) {
-               return nil, errTokensInUse
+// daemonRunChild runs a child's turn loop and its schedules until either stops.
+// Its events stay in its own log.
+func daemonRunChild(ctx context.Context, loop *TurnLoop, schedules *Schedules) error {
+       ctx, cancel := context.WithCancel(ctx)
+       defer cancel()
+
+       done := make(chan error, 1)
+       go func() {
+               done <- schedules.Run(ctx, loop)
+
+               cancel()
+       }()
+
+       err := loop.Run(ctx)
+       if errors.Is(err, context.Canceled) {
+               err = nil

        }
 
        }
 

-       return lock, err
+       cancel()
+
+       return errors.Join(err, <-done)
+}
+
+// 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)
+       }

 }
 }