X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/3cdbd017da9ae7a7045f4272f16d4b262fd01c79..4a42f7f208cbfead335eba9ebe615b35b1742130:/daemon.go diff --git a/daemon.go b/daemon.go index 5c94a4f..7945c2b 100644 --- a/daemon.go +++ b/daemon.go @@ -24,101 +24,6 @@ const ( 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. @@ -131,9 +36,17 @@ type DaemonHub struct { workers sync.WaitGroup } -// 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. +// 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. func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLoop) { for { conn, err := listener.Accept() @@ -254,14 +167,16 @@ func (h *DaemonHub) send(client *DaemonClient, event Event) { } } -// 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 *DaemonHub) remove(client *DaemonClient) { delete(h.clients, client) client.once.Do(func() { close(client.events) - _ = client.conn.Close() + // The client is removed already, so a close failure cannot + // restore it. + client.conn.Close() // nolint: errcheck }) } @@ -288,12 +203,170 @@ func (h *DaemonHub) CloseAll() { 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 +// 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) + if err != nil { + return fmt.Errorf("submit terminal message: %w", err) + } + + return nil + 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") + } + + 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) + } + + if err != nil { + return fmt.Errorf("stop subagent: %w", err) + } + + return nil + default: + return fmt.Errorf("unknown request %q", request.Type) + } +} + +// 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 fmt.Errorf("find data directory: %w", 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 fmt.Errorf("check daemon socket path: %w", err) + } + + config, err := loadConfig(dir) + if err != nil { + return fmt.Errorf("load daemon config: %w", err) + } + + botSettings, hasBot, err := loadTelegramConfig(dir) + if err != nil { + return fmt.Errorf("load Telegram config: %w", err) + } + + tokens, err := lockTokens(dir) + if err != nil { + return fmt.Errorf("lock login tokens: %w", 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 fmt.Errorf("open WhatsApp: %w", err) + } + + defer func() { + err = errors.Join(err, whatsapp.Close()) + }() + + address, auth, err := modelEndpoint(config, dir) + if err != nil { + return fmt.Errorf("configure model endpoint: %w", err) + } + + agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger) + if err != nil { + return fmt.Errorf("open daemon agents: %w", 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 fmt.Errorf("listen for daemon clients: %w", 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") + } + + err = daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) { + return root.Loop(whatsapp, notify) + }, agents[1:]...) + if err != nil { + return fmt.Errorf("serve daemon: %w", err) + } + + return nil } // daemonListenSocket listens on the Unix socket at path, with mode 0600. The @@ -302,7 +375,7 @@ type DaemonClient struct { func daemonListenSocket(path string) (net.Listener, error) { err := checkSocketPath(path) if err != nil { - return nil, err + return nil, fmt.Errorf("check daemon socket path: %w", err) } err = os.Remove(path) @@ -383,6 +456,7 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w root: l, messages: messages, } + l.tools.children = runtime var workers sync.WaitGroup @@ -484,58 +558,3 @@ func daemonRunChild(ctx context.Context, loop *TurnLoop, schedules *Schedules) e 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 "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") - } - - 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) - } -}