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(), agent.Images()), 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), agent.config.Name, 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                 openChild: func(ctx context.Context, child *Agent) (*DaemonAgent, error) {
  77                         return openDaemonAgent(ctx, child, address, auth, compactor, clock, logger.With("name", child.config.Name))
  78                 },
  79         }, nil
  80 }
  81
  82 // DaemonAgent owns the open files and model client of one persistent agent. Its
  83 // loop must stop before Close releases the files.
  84 type DaemonAgent struct {
  85         agent     *Agent
  86         memory    *Memory
  87         client    *ModelClient
  88         subagents *Subagents
  89         schedules *Schedules
  90         clock     func() time.Time
  91         logger    *slog.Logger
  92         openChild func(context.Context, *Agent) (*DaemonAgent, error)
  93 }
  94
  95 // Loop makes the turn loop with its event destination. Only the root has a
  96 // WhatsApp link. Child events stay with their own logs.
  97 func (a *DaemonAgent) Loop(whatsapp *WhatsApp, notify func(Event)) (*TurnLoop, error) {
  98         loop, err := newTurnLoop(a.agent, a.memory, a.subagents, whatsapp, a.schedules, a.client, a.clock, a.logger, notify)
  99         if err != nil {
 100                 return nil, err
 101         }
 102
 103         loop.openChild = a.openChild
 104
 105         return loop, nil
 106 }
 107
 108 // Close flushes and closes the subagents and the memory.
 109 func (a *DaemonAgent) Close() error {
 110         return errors.Join(a.subagents.Close(), a.memory.Close())
 111 }
 112
 113 // DaemonAgents holds the root first and then its children in directory order.
 114 type DaemonAgents []*DaemonAgent
 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 }