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 {
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 {
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())
+ 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
}
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
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 {
// 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()
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() {
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
_, 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")
projects / onechat / blobdiff