X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/5a93b6124068c2a8748fc0bfbb2a2d151815bcf0..80e348724041a694002fc38f57e8868b9b9a2be9:/daemon.go diff --git a/daemon.go b/daemon.go index 2a93f1d..dd75d31 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,6 +36,14 @@ type DaemonHub struct { 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. @@ -288,12 +201,103 @@ 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 +// 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 @@ -353,6 +357,10 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w 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) { @@ -493,6 +501,27 @@ func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, an _, 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")