X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/a5c83e92d7feea2e104b5515043a76c8c4c56c60..f5c601f1801aaa276f8c335fcb51c373edc0b50c:/daemon.go?ds=inline diff --git a/daemon.go b/daemon.go index 31a6714..dd75d31 100644 --- a/daemon.go +++ b/daemon.go @@ -24,105 +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()) - }() - - 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:]...) -} - // 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. @@ -135,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. @@ -292,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