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