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 // A routed batch with a removed child goes to the root.
  19 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
  20         routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
  21         if err != nil {
  22                 return nil, fmt.Errorf("open agent routes: %w", err)
  23         }
  24
  25         m := &DaemonMessages{
  26                 routes:  routes,
  27                 path:    filepath.Join(loops[0].agent.dir, "messages.json"),
  28                 ordered: append([]*TurnLoop(nil), loops...),
  29                 loops:   make(map[string]*TurnLoop),
  30         }
  31         for _, loop := range loops {
  32                 m.loops[loop.agent.config.Name] = loop
  33                 loop.messages = m
  34                 loop.tools.messages = m
  35                 if loop.agent.config.Parent == "" {
  36                         loop.logger = loop.logger.With("name", loop.agent.config.Name)
  37                         loop.tools.routes = routes
  38                         if loop.tools.whatsapp != nil {
  39                                 loop.tools.whatsapp.mutex.Lock()
  40
  41                                 loop.tools.whatsapp.routes = routes
  42                                 loop.tools.whatsapp.mutex.Unlock()
  43                         }
  44                 }
  45         }
  46
  47         for _, child := range loops[1:] {
  48                 err := loops[0].runs.subagents.Reserve(child.agent.config.Name)
  49                 if err != nil {
  50                         return nil, fmt.Errorf("reserve child name: %w", err)
  51                 }
  52         }
  53
  54         data, err := os.ReadFile(m.path)
  55         if err != nil && !errors.Is(err, os.ErrNotExist) {
  56                 return nil, fmt.Errorf("read agent messages: %w", err)
  57         }
  58
  59         if err == nil {
  60                 err = json.Unmarshal(data, &m.state)
  61                 if err != nil {
  62                         return nil, fmt.Errorf("decode agent messages: %w", err)
  63                 }
  64         }
  65
  66         if m.state.Next < 0 {
  67                 return nil, errors.New("invalid agent message sequence")
  68         }
  69
  70         for i, number := range m.state.WhatsAppBatches {
  71                 if number <= 0 || slices.Contains(m.state.WhatsAppBatches[:i], number) {
  72                         return nil, errors.New("invalid WhatsApp batch receipt")
  73                 }
  74         }
  75
  76         previous := 0
  77         for i := range m.state.Pending {
  78                 message := &m.state.Pending[i]
  79                 source := m.loops[message.Source]
  80                 target := m.loops[message.Target]
  81                 if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
  82                         message.Target = message.Source
  83                         message.LogID = nil
  84                         target = source
  85                 }
  86
  87                 if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
  88                         return nil, errors.New("invalid queued agent message")
  89                 }
  90
  91                 if message.WhatsAppBatch != 0 {
  92                         if message.WhatsAppBatch < 0 || message.Kind != "work" || source != loops[0] || (target != source && target.agent.config.Parent != message.Source) {
  93                                 return nil, errors.New("invalid queued WhatsApp batch")
  94                         }
  95                 } else if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
  96                         return nil, errors.New("queued agents are not parent and child")
  97                 }
  98
  99                 previous = message.ID
 100         }
 101
 102         for i := range m.state.Pending {
 103                 err = m.deliver(&m.state.Pending[i])
 104                 if err != nil {
 105                         return nil, fmt.Errorf("restore agent message: %w", err)
 106                 }
 107         }
 108
 109         return m, nil
 110 }
 111
 112 // DaemonMessages owns the durable queue between the root and its children and
 113 // the routed inputs. The mutex serializes queue writes and delivery. Backups
 114 // take it before the locks of the loops, memory, and subagents. A stored log ID
 115 // makes delivery idempotent, including a stop between the log append and the
 116 // wake.
 117 type DaemonMessages struct {
 118         mutex   sync.Mutex
 119         path    string
 120         loops   map[string]*TurnLoop
 121         ordered []*TurnLoop
 122         state   DaemonMessageFile
 123         failure error
 124         routes  *Routes
 125 }
 126
 127 // deliver records one queued message and wakes its recipient. The caller holds
 128 // the mutex, or owns the queue before the loops start.
 129 func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 130         loop := m.loops[message.Target]
 131         logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
 132                 message.LogID = &id
 133                 return m.save()
 134         })
 135         if err != nil {
 136                 return fmt.Errorf("record agent message: %w", err)
 137         }
 138
 139         status := ""
 140         source := message.Source
 141         if message.WhatsAppBatch > 0 {
 142                 source = whatsappSource
 143         } else if message.Kind == "work" {
 144                 status = subagentFinished
 145         }
 146
 147         loop.notify(Event{
 148                 Kind:   message.Kind,
 149                 ID:     logged.ID,
 150                 Text:   message.Text,
 151                 Source: source,
 152                 Status: status,
 153         })
 154         loop.Wake(logged)
 155
 156         if message.WhatsAppBatch > 0 {
 157                 loop.logger.Info("batch routed", "kind", whatsappSource, "id", message.WhatsAppBatch, "to", message.Target)
 158         } else {
 159                 loop.logger.Info("message received", "from", message.Source, "id", message.ID)
 160         }
 161
 162         return nil
 163 }
 164
 165 // save replaces the queue file under the mutex, or before the loops start.
 166 func (m *DaemonMessages) save() error {
 167         data, err := json.Marshal(m.state)
 168         if err != nil {
 169                 return fmt.Errorf("encode agent messages: %w", err)
 170         }
 171
 172         err = writeFileAtomic(m.path, append(data, '\n'))
 173         if err != nil {
 174                 return fmt.Errorf("save agent messages: %w", err)
 175         }
 176
 177         return nil
 178 }
 179
 180 // AddChild holds the queue lock while open makes a child's files and loop. The
 181 // loop joins backups before its first task can be delivered.
 182 func (m *DaemonMessages) AddChild(parent, name string, open func() (*TurnLoop, error)) (*TurnLoop, error) {
 183         m.mutex.Lock()
 184         defer m.mutex.Unlock()
 185
 186         if m.loops[name] != nil {
 187                 return nil, fmt.Errorf("name %q is already used", name)
 188         }
 189
 190         root := m.loops[parent]
 191         if root == nil || root.agent.config.Parent != "" {
 192                 return nil, errors.New("only the root can create children")
 193         }
 194
 195         if m.failure != nil {
 196                 return nil, fmt.Errorf("agent message queue stopped: %w", m.failure)
 197         }
 198
 199         loop, err := open()
 200         if err != nil {
 201                 return nil, fmt.Errorf("open child: %w", err)
 202         }
 203
 204         loop.messages = m
 205         loop.tools.messages = m
 206         m.loops[name] = loop
 207         m.ordered = append(m.ordered, loop)
 208
 209         return loop, nil
 210 }
 211
 212 // Child returns only a direct child of parent. Loops live until the daemon
 213 // stops, so the returned loop stays open for the caller's turn.
 214 func (m *DaemonMessages) Child(parent, name string) *TurnLoop {
 215         m.mutex.Lock()
 216         defer m.mutex.Unlock()
 217
 218         loop := m.loops[name]
 219         if loop == nil || loop.agent.config.Parent != parent {
 220                 return nil
 221         }
 222
 223         return loop
 224 }
 225
 226 // ChildState lists the names and responsibilities of a parent's children in
 227 // name order. It returns no heading when the parent has no children.
 228 func (m *DaemonMessages) ChildState(parent string) string {
 229         m.mutex.Lock()
 230         defer m.mutex.Unlock()
 231
 232         var lines []string
 233
 234         for name, loop := range m.loops {
 235                 if loop.agent.config.Parent == parent {
 236                         lines = append(lines, name+": "+loop.agent.config.Responsibility)
 237                 }
 238         }
 239
 240         if len(lines) == 0 {
 241                 return ""
 242         }
 243
 244         slices.Sort(lines)
 245
 246         return "\n\nChildren:\n" + strings.Join(lines, "\n")
 247 }
 248
 249 // SnapshotAgents locks the queue and all registered loops, including children
 250 // made during this daemon run, before snapshot copies their files.
 251 func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() error) error {
 252         return m.Snapshot(func() error {
 253                 return daemonSnapshot(ctx, m.ordered, snapshot)
 254         })
 255 }
 256
 257 // Tell queues a user message from parent to its own child, then wakes the
 258 // child. It is the daemon entry point for parent-to-child messages.
 259 func (m *DaemonMessages) Tell(parent, child, text string) error {
 260         m.mutex.Lock()
 261         defer m.mutex.Unlock()
 262
 263         source := m.loops[parent]
 264         target := m.loops[child]
 265         if source == nil || target == nil || target.agent.config.Parent != parent {
 266                 return errors.New("the recipient is not a child of the sender")
 267         }
 268
 269         if m.failure != nil {
 270                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 271         }
 272
 273         m.add(parent, child, "user", text)
 274
 275         err := m.save()
 276         if err == nil {
 277                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 278         }
 279
 280         if err != nil {
 281                 return fmt.Errorf("send agent message: %w", m.failed(err))
 282         }
 283
 284         return nil
 285 }
 286
 287 // RouteWhatsApp queues a routed batch as work and wakes its child. It returns
 288 // false for an unrouted source or a missing child. A saved receipt survives
 289 // turn acknowledgement until the inbox removes the batch. A repeat after a stop
 290 // between queueing and inbox removal cannot append a second entry.
 291 func (m *DaemonMessages) RouteWhatsApp(batch WhatsAppBatch, text string) (bool, error) {
 292         m.mutex.Lock()
 293         defer m.mutex.Unlock()
 294
 295         if m.failure != nil {
 296                 return false, fmt.Errorf("agent message queue stopped: %w", m.failure)
 297         }
 298
 299         if batch.Number <= 0 {
 300                 return false, errors.New("invalid WhatsApp batch number")
 301         }
 302
 303         if slices.Contains(m.state.WhatsAppBatches, batch.Number) {
 304                 return true, nil
 305         }
 306
 307         root := m.ordered[0].agent.config.Name
 308         name := m.routes.Target(whatsappSource + ":" + batch.Chat.ID)
 309         child := m.loops[name]
 310         if child == nil || child.agent.config.Parent != root {
 311                 return false, nil
 312         }
 313
 314         m.add(root, name, "work", text)
 315
 316         message := &m.state.Pending[len(m.state.Pending)-1]
 317         message.WhatsAppBatch = batch.Number
 318         m.state.WhatsAppBatches = append(m.state.WhatsAppBatches, batch.Number)
 319         err := m.save()
 320         if err == nil {
 321                 err = m.deliver(message)
 322         }
 323
 324         if err != nil {
 325                 return false, fmt.Errorf("route WhatsApp batch: %w", m.failed(err))
 326         }
 327
 328         return true, nil
 329 }
 330
 331 // PruneWhatsApp drops receipts only for batches absent from the durable inbox.
 332 // Queued turns keep their own log IDs until they finish. Receipts left by a
 333 // stop after inbox removal are removed by the next inbox delivery.
 334 func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
 335         m.mutex.Lock()
 336         defer m.mutex.Unlock()
 337
 338         if m.failure != nil {
 339                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 340         }
 341
 342         held := make(map[int]bool, len(batches))
 343         for _, batch := range batches {
 344                 held[batch.Number] = true
 345         }
 346
 347         before := len(m.state.WhatsAppBatches)
 348         m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
 349                 return !held[number]
 350         })
 351         if len(m.state.WhatsAppBatches) == before {
 352                 return nil
 353         }
 354
 355         err := m.save()
 356         if err != nil {
 357                 return fmt.Errorf("prune WhatsApp receipts: %w", m.failed(err))
 358         }
 359
 360         return nil
 361 }
 362
 363 // add appends a queue entry under the mutex. Its ID is never reused.
 364 func (m *DaemonMessages) add(source, target, kind, text string) {
 365         m.state.Next++
 366         m.state.Pending = append(m.state.Pending, DaemonMessage{
 367                 ID:     m.state.Next,
 368                 Source: source,
 369                 Target: target,
 370                 Kind:   kind,
 371                 Text:   text,
 372         })
 373 }
 374
 375 // failed stops all loops after a queue failure. Further sends must not change
 376 // the queue, because the outcome of the failed write is not known.
 377 func (m *DaemonMessages) failed(err error) error {
 378         m.failure = err
 379         for _, loop := range m.loops {
 380                 loop.fail(err)
 381         }
 382
 383         return err
 384 }
 385
 386 // Snapshot holds the queue lock while snapshot locks and copies agent files.
 387 // Its callback must take the memory locks after this lock, as delivery does.
 388 func (m *DaemonMessages) Snapshot(snapshot func() error) error {
 389         m.mutex.Lock()
 390         defer m.mutex.Unlock()
 391
 392         err := snapshot()
 393         if err != nil {
 394                 return fmt.Errorf("copy agent message snapshot: %w", err)
 395         }
 396
 397         return nil
 398 }
 399
 400 // Finish acknowledges the queued messages taken by a turn and queues its final
 401 // report in the same file replacement. A child with no final text sends
 402 // nothing. Call after a final reply, a failure, or a user interrupt, but not
 403 // when a daemon stop interrupts a model request.
 404 func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) error {
 405         m.mutex.Lock()
 406         defer m.mutex.Unlock()
 407
 408         if m.failure != nil {
 409                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 410         }
 411
 412         ids := make(map[int]bool, len(active))
 413         for _, message := range active {
 414                 ids[message.ID] = true
 415         }
 416
 417         name := loop.agent.config.Name
 418         before := len(m.state.Pending)
 419         m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {
 420                 return message.Target == name && message.LogID != nil && ids[*message.LogID]
 421         })
 422
 423         parent := loop.agent.config.Parent
 424         report := parent != "" && strings.TrimSpace(reply) != ""
 425         if report {
 426                 m.add(name, parent, "work", "["+name+"] "+reply)
 427         }
 428
 429         if !report && len(m.state.Pending) == before {
 430                 return nil
 431         }
 432
 433         err := m.save()
 434         if err == nil && report {
 435                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 436         }
 437
 438         if err != nil {
 439                 return fmt.Errorf("finish agent messages: %w", m.failed(err))
 440         }
 441
 442         return nil
 443 }
 444
 445 // DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
 446 // for WhatsApp batches until the inbox removes them. A receipt stays when its
 447 // turn finishes before inbox removal.
 448 type DaemonMessageFile struct {
 449         Next            int             `json:"next"`
 450         Pending         []DaemonMessage `json:"pending"`
 451         WhatsAppBatches []int           `json:"whatsapp_batches,omitempty"`
 452 }
 453
 454 // DaemonMessage is one queued delivery. LogID is set before the log append. A
 455 // nil LogID means that delivery has not started. WhatsAppBatch marks a routed
 456 // external batch, whose event source is WhatsApp rather than its queue sender.
 457 type DaemonMessage struct {
 458         ID            int    `json:"id"`
 459         Source        string `json:"source"`
 460         Target        string `json:"target"`
 461         Kind          string `json:"kind"`
 462         Text          string `json:"text"`
 463         LogID         *int   `json:"log_id,omitempty"`
 464         WhatsAppBatch int    `json:"whatsapp_batch,omitempty"`
 465 }