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