X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/790279faf0fda789d93a75f6f5996e65d5de4a18..12bd9e96f3b3e67290484e3829461e054110db27:/daemon.go?ds=inline diff --git a/daemon.go b/daemon.go index ef73896..5c94a4f 100644 --- a/daemon.go +++ b/daemon.go @@ -24,12 +24,13 @@ const ( 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. Its -// log lines go to out, through the loggers of daemonLogger. +// 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 { @@ -80,30 +81,18 @@ func daemon(ctx context.Context, out io.Writer) (err error) { 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), logger) + agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger) if err != nil { return err } defer func() { - err = errors.Join(err, memory.Close()) + err = errors.Join(err, agents.Close()) }() - schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), time.Now, logger) - if err != nil { - return err - } - - subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), time.Now, subagentNameRandom) - defer func() { - err = errors.Join(err, subagents.Close()) - }() + root := agents[0] + agent := root.agent + schedules := root.schedules listener, err := daemonListenSocket(socket) if err != nil { @@ -115,7 +104,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { linked = "linked" } - logger.Info("ready", "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len()) + 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 @@ -126,14 +115,16 @@ func daemon(ctx context.Context, out io.Writer) (err error) { } return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) { - return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, logger, notify) - }) + 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. +// runs the WhatsApp commands of the clients on whatsapp. snapshot locks all +// agents and subagents during a backup copy. type DaemonHub struct { whatsapp *WhatsApp + snapshot func(context.Context, func() error) error mutex sync.Mutex clients map[*DaemonClient]bool @@ -204,12 +195,24 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) 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 { @@ -325,8 +328,10 @@ func daemonListenSocket(path string) (net.Listener, error) { // logs its batches and notes in the loop and tells all clients its notices, and // 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. -func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error { +// 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() @@ -348,8 +353,56 @@ 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) { + }) + 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 + 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) go func() { @@ -406,7 +459,30 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w hub.CloseAll() workers.Wait() - return errors.Join(err, whatsappErr, schedulesErr, 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 @@ -421,6 +497,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")