Order model and chat declarations

[onechat] / daemon.go

diff --git a/daemon.go b/daemon.go
index 2d8301788daf789dd1ae62aa76f3b3f115728aaf..dd75d316316a4195ef2b79b4fc42b3c4b27e6bcf 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -7,7 +7,6 @@ import (
        "errors"
        "fmt"
        "io"
        "errors"
        "fmt"
        "io"

-       "log"

        "net"
        "os"
        "path/filepath"
        "net"
        "os"
        "path/filepath"


@@ -25,118 +24,26 @@ const (
        daemonClientEvents = 1024
 )
 
        daemonClientEvents = 1024
 )
 

-// 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. 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.
-func daemon(ctx context.Context, out io.Writer) (err error) {
-       dir, err := dataDir()
-       if err != nil {
-               return err
-       }
-
-       // Check the socket path before the daemon writes a file, so that a data
-       // directory with a long path stays empty.
-       socket := filepath.Join(dir, socketFileName)
-       err = checkSocketPath(socket)
-       if err != nil {
-               return err
-       }
-
-       config, err := loadConfig(dir)
-       if err != nil {
-               return err
-       }
-
-       botSettings, hasBot, err := loadTelegramConfig(dir)
-       if err != nil {
-               return err
-       }
-
-       tokens, err := lockTokens(dir)
-       if err != nil {
-               return err
-       }
-
-       defer func() {
-               err = errors.Join(err, tokens.Close())
-       }()
-
-       whatsappSetDeviceProps()
-       whatsapp, err := openWhatsApp(ctx, dir, time.Now, log.New(out, "whatsapp: ", log.LstdFlags), whatsappNewClient)
-       if err != nil {
-               return err
-       }
-
-       defer func() {
-               err = errors.Join(err, whatsapp.Close())
-       }()
-
-       address, auth, err := modelEndpoint(config, dir)
-       if err != nil {
-               return err
-       }
-
-       agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
-       if err != nil {
-               return err
-       }
-
-       client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
-       memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
-       if err != nil {
-               return err
-       }
-
-       defer func() {
-               err = errors.Join(err, memory.Close())
-       }()
-
-       schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), time.Now)
-       if err != nil {
-               return err
-       }
-
-       subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), time.Now, subagentNameRandom)
-       defer func() {
-               err = errors.Join(err, subagents.Close())
-       }()
-
-       listener, err := daemonListenSocket(socket)
-       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)
-       if err != nil {
-               return errors.Join(err, listener.Close())
-       }
-
-       var telegram *TelegramBot
-       if hasBot {
-               telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), log.New(out, "telegram: ", log.LstdFlags))
-               _, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner")
-               if err != nil {
-                       return errors.Join(err, listener.Close())
-               }
-       }
-
-       return daemonServe(ctx, listener, telegram, whatsapp, func(notify func(Event)) (*TurnLoop, error) {
-               return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, notify)
-       })
-}
-

 // DaemonHub holds the connected clients and sends each event to all of them. It
 // DaemonHub holds the connected clients and sends each event to all of them. It

-// runs the WhatsApp commands of the clients on whatsapp.
+// runs the WhatsApp commands of the clients on whatsapp. snapshot locks all
+// agents and subagents during a backup copy.

 type DaemonHub struct {
        whatsapp *WhatsApp
 type DaemonHub struct {
        whatsapp *WhatsApp

+       snapshot func(context.Context, func() error) error

 
        mutex   sync.Mutex
        clients map[*DaemonClient]bool
        workers sync.WaitGroup
 }
 
 
        mutex   sync.Mutex
        clients map[*DaemonClient]bool
        workers sync.WaitGroup
 }
 

+// 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
+}
+

 // 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.


@@ -153,10 +60,12 @@ func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLo
                }
 
                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)
 


@@ -193,16 +102,30 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop)
 
        scanner := bufio.NewScanner(client.conn)
        scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
 
        scanner := bufio.NewScanner(client.conn)
        scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)

+

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

+

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

-                       err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
-                               h.Send(client, Event{
-                                       Kind: "notice",
-                                       Text: text,
+                       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 {


@@ -250,6 +173,7 @@ 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()
        })
 }


@@ -267,20 +191,113 @@ func (h *DaemonHub) Broadcast(event Event) {
 // CloseAll drops every client and waits for their readers and writers.
 func (h *DaemonHub) CloseAll() {
        h.mutex.Lock()
 // CloseAll drops every client and waits for their readers and writers.
 func (h *DaemonHub) CloseAll() {
        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()
 }
 

-// 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
+// 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 {
+               return err
+       }
+
+       // Check the socket path before the daemon writes a file, so that a data
+       // directory with a long path stays empty.
+       socket := filepath.Join(dir, socketFileName)
+       err = checkSocketPath(socket)
+       if err != nil {
+               return err
+       }
+
+       config, err := loadConfig(dir)
+       if err != nil {
+               return err
+       }
+
+       botSettings, hasBot, err := loadTelegramConfig(dir)
+       if err != nil {
+               return err
+       }
+
+       tokens, err := lockTokens(dir)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, tokens.Close())
+       }()
+
+       whatsappSetDeviceProps()
+
+       logger := daemonLogger(out, "onechat: ")
+       whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, whatsapp.Close())
+       }()
+
+       address, auth, err := modelEndpoint(config, dir)
+       if err != nil {
+               return err
+       }
+
+       agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, agents.Close())
+       }()
+
+       whatsapp.inbox.voice = &VoiceTranscriber{config: config.Voice}
+
+       root := agents[0]
+       agent := root.agent
+       schedules := root.schedules
+
+       listener, err := daemonListenSocket(socket)
+       if err != nil {
+               return err
+       }
+
+       linked := "not linked"
+       if whatsapp.isLinked() {
+               linked = "linked"
+       }
+
+       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())
+
+       var telegram *TelegramBot
+
+       if hasBot {
+               telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))
+
+               telegram.voice = &VoiceTranscriber{config: config.Voice}
+
+               logger.Info("Telegram bot started")
+       }
+
+       return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+               return root.Loop(whatsapp, notify)
+       }, agents[1:]...)

 }
 
 // daemonListenSocket listens on the Unix socket at path, with mode 0600. The
 }
 
 // daemonListenSocket listens on the Unix socket at path, with mode 0600. The


@@ -313,9 +330,12 @@ func daemonListenSocket(path string) (net.Listener, error) {
 // 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
 // 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. 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, start func(notify func(Event)) (*TurnLoop, error)) error {
+// 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()
 
        ctx, cancel := context.WithCancel(ctx)
        defer cancel()
 


@@ -337,8 +357,58 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
                return errors.Join(err, listener.Close())
        }
 
                return errors.Join(err, listener.Close())
        }
 

+       if bot != nil {
+               l.tools.imageShow.bot = bot
+       }
+
+       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
        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)
        workers.Add(2)

+

        go func() {
                defer workers.Done()
 
        go func() {
                defer workers.Done()
 


@@ -346,20 +416,37 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
        }()
 
        var whatsappErr error
        }()
 
        var whatsappErr error

+

        go func() {
                defer workers.Done()
 
                whatsappErr = whatsapp.Run(ctx, l, notify)
        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
                cancel()
        }()
 
        var botErr error

+

        if bot != nil {
                workers.Add(1)
        if bot != nil {
                workers.Add(1)

+

                go func() {
                        defer workers.Done()
 
                        botErr = bot.Run(ctx, l, whatsapp)
                go func() {
                        defer workers.Done()
 
                        botErr = bot.Run(ctx, l, whatsapp)

+

                        cancel()
                }()
        }
                        cancel()
                }()
        }


@@ -370,11 +457,36 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
        }
 
        cancel()
        }
 
        cancel()

+

        err = errors.Join(err, listener.Close())
        err = errors.Join(err, listener.Close())

+

        hub.CloseAll()
        workers.Wait()
 
        hub.CloseAll()
        workers.Wait()
 

-       return errors.Join(err, whatsappErr, botErr)
+       return errors.Join(err, whatsappErr, schedulesErr, botErr, errors.Join(childErrors...), runtime.Close())
+}
+
+// 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
+       }
+
+       cancel()
+
+       return errors.Join(err, <-done)

 }
 
 // daemonRunRequest runs one request of a client. The answers of a WhatsApp
 }
 
 // daemonRunRequest runs one request of a client. The answers of a WhatsApp


@@ -389,6 +501,27 @@ func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, an
                _, err := l.Submit(terminalSource, request.Text)
 
                return err
                _, err := l.Submit(terminalSource, request.Text)
 
                return err

+       case "image":
+               if !filepath.IsAbs(request.Path) {
+                       return errors.New("use /image PATH [caption]")
+               }
+
+               data, err := imageReadFile(request.Path)
+               if err != nil {
+                       return fmt.Errorf("read terminal image: %w", err)
+               }
+
+               name, err := l.agent.Images().Save(data)
+               if err != nil {
+                       return fmt.Errorf("store terminal image: %w", err)
+               }
+
+               _, err = l.Submit(terminalSource, imageText([]string{name}, request.Text))
+               if err != nil {
+                       return fmt.Errorf("submit terminal image: %w", err)
+               }
+
+               return nil

        case "interrupt":
                if !l.Interrupt() {
                        return errors.New("no turn runs")
        case "interrupt":
                if !l.Interrupt() {
                        return errors.New("no turn runs")