Separate multiline Go statements

[onechat] / daemon.go

diff --git a/daemon.go b/daemon.go
index 3a54b1ae94374c7d53c3938f1eeff0ae8fcb9768..9f71d801be1c985ee70590e7118e69eeba91abbb 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -24,101 +24,6 @@ const (
        daemonClientEvents = 1024
 )
 
        daemonClientEvents = 1024
 )
 

-// 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())
-       }()
-
-       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: "))
-
-               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:]...)
-}
-

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


@@ -131,6 +36,14 @@ type DaemonHub struct {
        workers sync.WaitGroup
 }
 
        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.


@@ -288,12 +201,103 @@ func (h *DaemonHub) CloseAll() {
        h.workers.Wait()
 }
 
        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


@@ -353,6 +357,10 @@ 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) {
        loops := []*TurnLoop{l}
        for _, child := range children {
                loop, err := child.Loop(nil, func(Event) {


@@ -364,10 +372,24 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
                loops = append(loops, loop)
        }
 
                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 {
        hub.snapshot = func(ctx context.Context, snapshot func() error) error {

-               return daemonSnapshot(ctx, loops, snapshot)
+               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))
        var workers sync.WaitGroup
 
        childErrors := make([]error, len(children))


@@ -442,7 +464,7 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
        hub.CloseAll()
        workers.Wait()
 
        hub.CloseAll()
        workers.Wait()
 

-       return errors.Join(err, whatsappErr, schedulesErr, botErr, errors.Join(childErrors...))
+       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.
 }
 
 // daemonRunChild runs a child's turn loop and its schedules until either stops.


@@ -480,6 +502,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")