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

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "fmt"
   8         "os"
   9         "path/filepath"
  10         "slices"
  11         "strings"
  12         "sync"
  13 )
  14
  15 // openDaemonMessages connects the root and its children. It restores queued
  16 // messages before the loops start. A queue entry stays until its recipient
  17 // finishes the turn that takes it, so a stop can interrupt delivery or a turn.
  18 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
  19         routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
  20         if err != nil {
  21                 return nil, fmt.Errorf("open agent routes: %w", err)
  22         }
  23
  24         m := &DaemonMessages{
  25                 routes:  routes,
  26                 path:    filepath.Join(loops[0].agent.dir, "messages.json"),
  27                 ordered: append([]*TurnLoop(nil), loops...),
  28                 loops:   make(map[string]*TurnLoop),
  29         }
  30         for _, loop := range loops {
  31                 m.loops[loop.agent.config.Name] = loop
  32                 loop.messages = m
  33                 loop.tools.messages = m
  34                 if loop.agent.config.Parent == "" {
  35                         loop.logger = loop.logger.With("name", loop.agent.config.Name)
  36                         loop.tools.routes = routes
  37                         if loop.tools.whatsapp != nil {
  38                                 loop.tools.whatsapp.mutex.Lock()
  39
  40                                 loop.tools.whatsapp.routes = routes
  41                                 loop.tools.whatsapp.mutex.Unlock()
  42                         }
  43                 }
  44         }
  45
  46         for _, child := range loops[1:] {
  47                 err := loops[0].runs.subagents.Reserve(child.agent.config.Name)
  48                 if err != nil {
  49                         return nil, fmt.Errorf("reserve child name: %w", err)
  50                 }
  51         }
  52
  53         data, err := os.ReadFile(m.path)
  54         if err != nil && !errors.Is(err, os.ErrNotExist) {
  55                 return nil, fmt.Errorf("read agent messages: %w", err)
  56         }
  57
  58         if err == nil {
  59                 err = json.Unmarshal(data, &m.state)
  60                 if err != nil {
  61                         return nil, fmt.Errorf("decode agent messages: %w", err)
  62                 }
  63         }
  64
  65         if m.state.Next < 0 {
  66                 return nil, errors.New("invalid agent message sequence")
  67         }
  68
  69         previous := 0
  70         for _, message := range m.state.Pending {
  71                 source := m.loops[message.Source]
  72                 target := m.loops[message.Target]
  73                 if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
  74                         return nil, errors.New("invalid queued agent message")
  75                 }
  76
  77                 if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
  78                         return nil, errors.New("queued agents are not parent and child")
  79                 }
  80
  81                 previous = message.ID
  82         }
  83
  84         for i := range m.state.Pending {
  85                 err = m.deliver(&m.state.Pending[i])
  86                 if err != nil {
  87                         return nil, fmt.Errorf("restore agent message: %w", err)
  88                 }
  89         }
  90
  91         return m, nil
  92 }
  93
  94 // DaemonMessages owns the durable queue between the root and its children. The
  95 // mutex serializes queue writes and delivery. Backups take it before the locks
  96 // of the loops, memory, and subagents. A stored log ID makes delivery
  97 // idempotent, including a stop between the log append and the wake.
  98 type DaemonMessages struct {
  99         mutex   sync.Mutex
 100         path    string
 101         loops   map[string]*TurnLoop
 102         ordered []*TurnLoop
 103         state   DaemonMessageFile
 104         failure error
 105         routes  *Routes
 106 }
 107
 108 // deliver records one queued message and wakes its recipient. The caller holds
 109 // the mutex, or owns the queue before the loops start.
 110 func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 111         loop := m.loops[message.Target]
 112         logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
 113                 message.LogID = &id
 114                 return m.save()
 115         })
 116         if err != nil {
 117                 return fmt.Errorf("record agent message: %w", err)
 118         }
 119
 120         status := ""
 121         if message.Kind == "work" {
 122                 status = subagentFinished
 123         }
 124
 125         loop.notify(Event{
 126                 Kind:   message.Kind,
 127                 ID:     logged.ID,
 128                 Text:   message.Text,
 129                 Source: message.Source,
 130                 Status: status,
 131         })
 132         loop.Wake(logged)
 133         loop.logger.Info("message received", "from", message.Source, "id", message.ID)
 134
 135         return nil
 136 }
 137
 138 // save replaces the queue file under the mutex, or before the loops start.
 139 func (m *DaemonMessages) save() error {
 140         data, err := json.Marshal(m.state)
 141         if err != nil {
 142                 return fmt.Errorf("encode agent messages: %w", err)
 143         }
 144
 145         err = writeFileAtomic(m.path, append(data, '\n'))
 146         if err != nil {
 147                 return fmt.Errorf("save agent messages: %w", err)
 148         }
 149
 150         return nil
 151 }
 152
 153 // AddChild holds the queue lock while open makes a child's files and loop. The
 154 // loop joins backups before its first task can be delivered.
 155 func (m *DaemonMessages) AddChild(parent, name string, open func() (*TurnLoop, error)) (*TurnLoop, error) {
 156         m.mutex.Lock()
 157         defer m.mutex.Unlock()
 158
 159         if m.loops[name] != nil {
 160                 return nil, fmt.Errorf("name %q is already used", name)
 161         }
 162
 163         root := m.loops[parent]
 164         if root == nil || root.agent.config.Parent != "" {
 165                 return nil, errors.New("only the root can create children")
 166         }
 167
 168         if m.failure != nil {
 169                 return nil, fmt.Errorf("agent message queue stopped: %w", m.failure)
 170         }
 171
 172         loop, err := open()
 173         if err != nil {
 174                 return nil, fmt.Errorf("open child: %w", err)
 175         }
 176
 177         loop.messages = m
 178         loop.tools.messages = m
 179         m.loops[name] = loop
 180         m.ordered = append(m.ordered, loop)
 181
 182         return loop, nil
 183 }
 184
 185 // Child returns only a direct child of parent. Loops live until the daemon
 186 // stops, so the returned loop stays open for the caller's turn.
 187 func (m *DaemonMessages) Child(parent, name string) *TurnLoop {
 188         m.mutex.Lock()
 189         defer m.mutex.Unlock()
 190
 191         loop := m.loops[name]
 192         if loop == nil || loop.agent.config.Parent != parent {
 193                 return nil
 194         }
 195
 196         return loop
 197 }
 198
 199 // ChildState lists the names and responsibilities of a parent's children in
 200 // name order. It returns no heading when the parent has no children.
 201 func (m *DaemonMessages) ChildState(parent string) string {
 202         m.mutex.Lock()
 203         defer m.mutex.Unlock()
 204
 205         var lines []string
 206
 207         for name, loop := range m.loops {
 208                 if loop.agent.config.Parent == parent {
 209                         lines = append(lines, name+": "+loop.agent.config.Responsibility)
 210                 }
 211         }
 212
 213         if len(lines) == 0 {
 214                 return ""
 215         }
 216
 217         slices.Sort(lines)
 218
 219         return "\n\nChildren:\n" + strings.Join(lines, "\n")
 220 }
 221
 222 // SnapshotAgents locks the queue and all registered loops, including children
 223 // made during this daemon run, before snapshot copies their files.
 224 func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() error) error {
 225         return m.Snapshot(func() error {
 226                 return daemonSnapshot(ctx, m.ordered, snapshot)
 227         })
 228 }
 229
 230 // Tell queues a user message from parent to its own child, then wakes the
 231 // child. It is the daemon entry point for parent-to-child messages.
 232 func (m *DaemonMessages) Tell(parent, child, text string) error {
 233         m.mutex.Lock()
 234         defer m.mutex.Unlock()
 235
 236         source := m.loops[parent]
 237         target := m.loops[child]
 238         if source == nil || target == nil || target.agent.config.Parent != parent {
 239                 return errors.New("the recipient is not a child of the sender")
 240         }
 241
 242         if m.failure != nil {
 243                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 244         }
 245
 246         m.add(parent, child, "user", text)
 247
 248         err := m.save()
 249         if err == nil {
 250                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 251         }
 252
 253         if err != nil {
 254                 return fmt.Errorf("send agent message: %w", m.failed(err))
 255         }
 256
 257         return nil
 258 }
 259
 260 // add appends a queue entry under the mutex. Its ID is never reused.
 261 func (m *DaemonMessages) add(source, target, kind, text string) {
 262         m.state.Next++
 263         m.state.Pending = append(m.state.Pending, DaemonMessage{
 264                 ID:     m.state.Next,
 265                 Source: source,
 266                 Target: target,
 267                 Kind:   kind,
 268                 Text:   text,
 269         })
 270 }
 271
 272 // failed stops all loops after a queue failure. Further sends must not change
 273 // the queue, because the outcome of the failed write is not known.
 274 func (m *DaemonMessages) failed(err error) error {
 275         m.failure = err
 276         for _, loop := range m.loops {
 277                 loop.fail(err)
 278         }
 279
 280         return err
 281 }
 282
 283 // Snapshot holds the queue lock while snapshot locks and copies agent files.
 284 // Its callback must take the memory locks after this lock, as delivery does.
 285 func (m *DaemonMessages) Snapshot(snapshot func() error) error {
 286         m.mutex.Lock()
 287         defer m.mutex.Unlock()
 288
 289         err := snapshot()
 290         if err != nil {
 291                 return fmt.Errorf("copy agent message snapshot: %w", err)
 292         }
 293
 294         return nil
 295 }
 296
 297 // Finish acknowledges the queued messages taken by a turn and queues its final
 298 // report in the same file replacement. A child with no final text sends
 299 // nothing. Call after a final reply, a failure, or a user interrupt, but not
 300 // when a daemon stop interrupts a model request.
 301 func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) error {
 302         m.mutex.Lock()
 303         defer m.mutex.Unlock()
 304
 305         if m.failure != nil {
 306                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 307         }
 308
 309         ids := make(map[int]bool, len(active))
 310         for _, message := range active {
 311                 ids[message.ID] = true
 312         }
 313
 314         name := loop.agent.config.Name
 315         before := len(m.state.Pending)
 316         m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {
 317                 return message.Target == name && message.LogID != nil && ids[*message.LogID]
 318         })
 319
 320         parent := loop.agent.config.Parent
 321         report := parent != "" && strings.TrimSpace(reply) != ""
 322         if report {
 323                 m.add(name, parent, "work", "["+name+"] "+reply)
 324         }
 325
 326         if !report && len(m.state.Pending) == before {
 327                 return nil
 328         }
 329
 330         err := m.save()
 331         if err == nil && report {
 332                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 333         }
 334
 335         if err != nil {
 336                 return fmt.Errorf("finish agent messages: %w", m.failed(err))
 337         }
 338
 339         return nil
 340 }
 341
 342 // DaemonMessageFile holds the sequence and the unacknowledged deliveries.
 343 type DaemonMessageFile struct {
 344         Next    int             `json:"next"`
 345         Pending []DaemonMessage `json:"pending"`
 346 }
 347
 348 // DaemonMessage is one queued delivery. LogID is set before the log append. A
 349 // nil LogID means that delivery has not started.
 350 type DaemonMessage struct {
 351         ID     int    `json:"id"`
 352         Source string `json:"source"`
 353         Target string `json:"target"`
 354         Kind   string `json:"kind"`
 355         Text   string `json:"text"`
 356         LogID  *int   `json:"log_id,omitempty"`
 357 }