git.giacomolli.uk - onechat/blob - daemon_agent.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "log/slog"
   8         "path/filepath"
   9         "time"
  10 )
  11
  12 // openDaemonAgents opens the root and its direct children. An invalid child
  13 // stops startup, and each opened agent is closed when startup fails.
  14 func openDaemonAgents(ctx context.Context, dir, address string, auth ModelAuthorizer, compactor ModelConfig, clock func() time.Time, logger *slog.Logger) (agents DaemonAgents, err error) {
  15         root, err := openAgent(filepath.Join(dir, daemonAgentDirName))
  16         if err != nil {
  17                 return nil, fmt.Errorf("open root agent: %w", err)
  18         }
  19
  20         if root.config.Parent != "" {
  21                 return nil, errors.New("root agent cannot have a parent")
  22         }
  23
  24         children, err := root.Children()
  25         if err != nil {
  26                 return nil, fmt.Errorf("open child agents: %w", err)
  27         }
  28
  29         defer func() {
  30                 if err != nil {
  31                         err = errors.Join(err, agents.Close())
  32                 }
  33         }()
  34
  35         for _, agent := range append([]*Agent{root}, children...) {
  36                 part := logger
  37                 if agent.config.Parent != "" {
  38                         part = logger.With("name", agent.config.Name)
  39                 }
  40
  41                 var opened *DaemonAgent
  42
  43                 opened, err = openDaemonAgent(ctx, agent, address, auth, compactor, clock, part)
  44                 if err != nil {
  45                         return agents, fmt.Errorf("open agent %q: %w", agent.config.Name, err)
  46                 }
  47
  48                 agents = append(agents, opened)
  49         }
  50
  51         return agents, nil
  52 }
  53
  54 // openDaemonAgent opens the memory, model client, schedules, and subagents of
  55 // one agent. All model calls use its cache key and usage file.
  56 func openDaemonAgent(ctx context.Context, agent *Agent, address string, auth ModelAuthorizer, compactor ModelConfig, clock func() time.Time, logger *slog.Logger) (*DaemonAgent, error) {
  57         client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
  58         memory, err := openMemory(ctx, agent.dir, clock, newCompactor(client, agent.Prompt, compactor, agent.Tools()), logger)
  59         if err != nil {
  60                 return nil, fmt.Errorf("open memory: %w", err)
  61         }
  62
  63         schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), clock, logger)
  64         if err != nil {
  65                 return nil, errors.Join(fmt.Errorf("open schedules: %w", err), memory.Close())
  66         }
  67
  68         return &DaemonAgent{
  69                 agent:     agent,
  70                 memory:    memory,
  71                 client:    client,
  72                 subagents: newSubagents(filepath.Join(agent.dir, subagentDirName), clock, subagentNameRandom),
  73                 schedules: schedules,
  74                 clock:     clock,
  75                 logger:    logger,
  76         }, nil
  77 }
  78
  79 // DaemonAgent owns the open files and model client of one persistent agent. Its
  80 // loop must stop before Close releases the files.
  81 type DaemonAgent struct {
  82         agent     *Agent
  83         memory    *Memory
  84         client    *ModelClient
  85         subagents *Subagents
  86         schedules *Schedules
  87         clock     func() time.Time
  88         logger    *slog.Logger
  89 }
  90
  91 // Loop makes the turn loop with its event destination. Only the root has a
  92 // WhatsApp link. Child events stay with their own logs.
  93 func (a *DaemonAgent) Loop(whatsapp *WhatsApp, notify func(Event)) (*TurnLoop, error) {
  94         return newTurnLoop(a.agent, a.memory, a.subagents, whatsapp, a.schedules, a.client, a.clock, a.logger, notify)
  95 }
  96
  97 // Close flushes and closes the subagents and the memory.
  98 func (a *DaemonAgent) Close() error {
  99         return errors.Join(a.subagents.Close(), a.memory.Close())
 100 }
 101
 102 // DaemonAgents holds the root first and then its children in directory order.
 103 type DaemonAgents []*DaemonAgent
 104
 105 // Close closes all agents, including after a partial startup.
 106 func (a DaemonAgents) Close() error {
 107         var err error
 108
 109         for _, agent := range a {
 110                 err = errors.Join(err, agent.Close())
 111         }
 112
 113         return err
 114 }
 115
 116 // ChildNames returns the names for the daemon's ready line.
 117 func (a DaemonAgents) ChildNames() []string {
 118         names := make([]string, 0, len(a)-1)
 119         for _, child := range a[1:] {
 120                 names = append(names, child.agent.config.Name)
 121         }
 122
 123         return names
 124 }
 125
 126 // daemonSnapshot flushes and locks all agent and subagent logs before copy.
 127 // Loops must stay in the same order for every snapshot.
 128 func daemonSnapshot(ctx context.Context, loops []*TurnLoop, snapshot func() error) error {
 129         if len(loops) == 0 {
 130                 err := snapshot()
 131                 if err != nil {
 132                         return fmt.Errorf("copy agent files: %w", err)
 133                 }
 134
 135                 return nil
 136         }
 137
 138         err := loops[0].runs.Snapshot(ctx, func() error {
 139                 return daemonSnapshot(ctx, loops[1:], snapshot)
 140         })
 141         if err != nil {
 142                 return fmt.Errorf("snapshot agent %q: %w", loops[0].agent.config.Name, err)
 143         }
 144
 145         return nil
 146 }