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