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

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/hex"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "os"
  10         "path/filepath"
  11         "slices"
  12         "strings"
  13         "sync"
  14 )
  15
  16 // DaemonMessages owns the durable queue between the root and its children and
  17 // the routed inputs and root notifications. The mutex serializes queue writes
  18 // and delivery. Backups take it before the locks of the loops, memory, and
  19 // subagents. A stored log ID makes delivery idempotent, including a stop
  20 // between the log append and the wake.
  21 type DaemonMessages struct {
  22         mutex   sync.Mutex
  23         path    string
  24         loops   map[string]*TurnLoop
  25         ordered []*TurnLoop
  26         state   DaemonMessageFile
  27         failure error
  28         routes  *Routes
  29 }
  30
  31 // DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
  32 // for WhatsApp batches until the inbox removes them. A receipt stays when its
  33 // turn finishes before inbox removal. Notification receipts stay after their
  34 // turn finishes and do not expire in the local test interface.
  35 type DaemonMessageFile struct {
  36         Next            int             `json:"next"`
  37         Pending         []DaemonMessage `json:"pending"`
  38         WhatsAppBatches []int           `json:"whatsapp_batches,omitempty"`
  39         Notifications   []NotifyReceipt `json:"notifications,omitempty"`
  40 }
  41
  42 // DaemonMessage is one queued delivery. LogID is set before the log append. A
  43 // nil LogID means that delivery has not started. WhatsAppBatch marks a routed
  44 // external batch, whose event source is WhatsApp rather than its queue sender.
  45 // Notification marks root external data held outside memory until a turn
  46 // starts.
  47 type DaemonMessage struct {
  48         ID            int    `json:"id"`
  49         Source        string `json:"source"`
  50         Target        string `json:"target"`
  51         Kind          string `json:"kind"`
  52         Text          string `json:"text"`
  53         LogID         *int   `json:"log_id,omitempty"`
  54         WhatsAppBatch int    `json:"whatsapp_batch,omitempty"`
  55         Notification  bool   `json:"notification,omitempty"`
  56 }
  57
  58 // openDaemonMessages connects the root and its children. It restores queued
  59 // messages before the loops start. External events without a saved log ID stay
  60 // unlogged until their turn boundary. Prepared log IDs are restored before
  61 // startup writers can use them, without waking a turn. A queue entry stays
  62 // until its recipient finishes the turn that takes it, so a stop can interrupt
  63 // delivery or a turn. A routed batch with a removed child goes to the root.
  64 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
  65         routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
  66         if err != nil {
  67                 return nil, fmt.Errorf("open agent routes: %w", err)
  68         }
  69
  70         m := &DaemonMessages{
  71                 routes:  routes,
  72                 path:    filepath.Join(loops[0].agent.dir, "messages.json"),
  73                 ordered: append([]*TurnLoop(nil), loops...),
  74                 loops:   make(map[string]*TurnLoop),
  75         }
  76
  77         for _, loop := range loops {
  78                 m.loops[loop.agent.config.Name] = loop
  79                 loop.messages = m
  80                 loop.tools.messages = m
  81                 if loop.agent.config.Parent == "" {
  82                         loop.logger = loop.logger.With("name", loop.agent.config.Name)
  83                         loop.tools.routes = routes
  84                         if loop.tools.whatsapp != nil {
  85                                 loop.tools.whatsapp.mutex.Lock()
  86
  87                                 loop.tools.whatsapp.routes = routes
  88                                 loop.tools.whatsapp.mutex.Unlock()
  89                         }
  90                 }
  91         }
  92
  93         for _, child := range loops[1:] {
  94                 err := loops[0].runs.subagents.Reserve(child.agent.config.Name)
  95                 if err != nil {
  96                         return nil, fmt.Errorf("reserve child name: %w", err)
  97                 }
  98         }
  99
 100         data, err := os.ReadFile(m.path)
 101         if err != nil && !errors.Is(err, os.ErrNotExist) {
 102                 return nil, fmt.Errorf("read agent messages: %w", err)
 103         }
 104
 105         if err == nil {
 106                 err = json.Unmarshal(data, &m.state)
 107                 if err != nil {
 108                         return nil, fmt.Errorf("decode agent messages: %w", err)
 109                 }
 110         }
 111
 112         if m.state.Next < 0 {
 113                 return nil, errors.New("invalid agent message sequence")
 114         }
 115
 116         for i, number := range m.state.WhatsAppBatches {
 117                 if number <= 0 || slices.Contains(m.state.WhatsAppBatches[:i], number) {
 118                         return nil, errors.New("invalid WhatsApp batch receipt")
 119                 }
 120         }
 121
 122         if len(m.state.Notifications) > notifyReceipts {
 123                 return nil, errors.New("too many notification receipts")
 124         }
 125
 126         receipts := make(map[int]bool)
 127         keys := make(map[[2]string]bool)
 128         for _, receipt := range m.state.Notifications {
 129                 key := [2]string{receipt.Source, receipt.ID}
 130                 digest, err := hex.DecodeString(receipt.Digest)
 131                 if err != nil || len(digest) != 32 || receipt.Number <= 0 || receipt.Number > m.state.Next || receipts[receipt.Number] || keys[key] || !notifyIdentifier(receipt.Source) || !notifyIdentifier(receipt.ID) {
 132                         return nil, errors.New("invalid notification receipt")
 133                 }
 134
 135                 receipts[receipt.Number] = true
 136                 keys[key] = true
 137         }
 138
 139         previous := 0
 140         for i := range m.state.Pending {
 141                 message := &m.state.Pending[i]
 142                 source := m.loops[message.Source]
 143                 target := m.loops[message.Target]
 144                 if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
 145                         message.Target = message.Source
 146                         message.LogID = nil
 147                         target = source
 148                 }
 149
 150                 if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
 151                         return nil, errors.New("invalid queued agent message")
 152                 }
 153
 154                 switch {
 155                 case message.Notification:
 156                         if source != loops[0] || target != source || message.Kind != "work" || message.WhatsAppBatch != 0 || !receipts[message.ID] {
 157                                 return nil, errors.New("invalid queued notification")
 158                         }
 159                 case message.WhatsAppBatch != 0:
 160                         if message.WhatsAppBatch < 0 || message.Kind != "work" || source != loops[0] || (target != source && target.agent.config.Parent != message.Source) {
 161                                 return nil, errors.New("invalid queued WhatsApp batch")
 162                         }
 163                 case (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target):
 164                         return nil, errors.New("queued agents are not parent and child")
 165                 }
 166
 167                 previous = message.ID
 168         }
 169
 170         // Complete prepared log reservations before another restored delivery
 171         // or startup writer can use their IDs. They belong to an interrupted
 172         // turn, not newly accepted data. Only the next turn boundary wakes
 173         // them.
 174         for i := range m.state.Pending {
 175                 message := &m.state.Pending[i]
 176                 if !message.Notification || message.LogID == nil {
 177                         continue
 178                 }
 179
 180                 loop := m.loops[message.Target]
 181                 _, err = loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
 182                         message.LogID = &id
 183                         return m.save()
 184                 })
 185                 if err != nil {
 186                         return nil, fmt.Errorf("restore external event reservation: %w", err)
 187                 }
 188         }
 189
 190         for i := range m.state.Pending {
 191                 if m.state.Pending[i].Notification {
 192                         continue
 193                 }
 194
 195                 err = m.deliver(&m.state.Pending[i])
 196                 if err != nil {
 197                         return nil, fmt.Errorf("restore agent message: %w", err)
 198                 }
 199         }
 200
 201         return m, nil
 202 }
 203
 204 // deliver records one queued message and wakes its recipient. The caller holds
 205 // the mutex, or owns the queue before the loops start.
 206 func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 207         loop := m.loops[message.Target]
 208         logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
 209                 message.LogID = &id
 210                 return m.save()
 211         })
 212         if err != nil {
 213                 return fmt.Errorf("record agent message: %w", err)
 214         }
 215
 216         status := ""
 217         source := message.Source
 218         if message.WhatsAppBatch > 0 {
 219                 source = whatsappSource
 220         } else if message.Kind == "work" {
 221                 status = subagentFinished
 222         }
 223
 224         loop.notify(Event{
 225                 Kind:   message.Kind,
 226                 ID:     logged.ID,
 227                 Text:   message.Text,
 228                 Source: source,
 229                 Status: status,
 230         })
 231
 232         loop.Wake(logged)
 233
 234         if message.WhatsAppBatch > 0 {
 235                 loop.logger.Info("batch routed", "kind", whatsappSource, "id", message.WhatsAppBatch, "to", message.Target)
 236         } else {
 237                 loop.logger.Info("message received", "from", message.Source, "id", message.ID)
 238         }
 239
 240         return nil
 241 }
 242
 243 // save replaces the queue file under the mutex, or before the loops start.
 244 func (m *DaemonMessages) save() error {
 245         data, err := json.Marshal(m.state)
 246         if err != nil {
 247                 return fmt.Errorf("encode agent messages: %w", err)
 248         }
 249
 250         err = writeFileDurable(m.path, append(data, '\n'))
 251         if err != nil {
 252                 return fmt.Errorf("save agent messages: %w", err)
 253         }
 254
 255         return nil
 256 }
 257
 258 // AddChild holds the queue lock while open makes a child's files and loop. The
 259 // loop joins backups before its first task can be delivered.
 260 func (m *DaemonMessages) AddChild(parent, name string, open func() (*TurnLoop, error)) (*TurnLoop, error) {
 261         m.mutex.Lock()
 262         defer m.mutex.Unlock()
 263
 264         if m.loops[name] != nil {
 265                 return nil, fmt.Errorf("name %q is already used", name)
 266         }
 267
 268         root := m.loops[parent]
 269         if root == nil || root.agent.config.Parent != "" {
 270                 return nil, errors.New("only the root can create children")
 271         }
 272
 273         if m.failure != nil {
 274                 return nil, fmt.Errorf("agent message queue stopped: %w", m.failure)
 275         }
 276
 277         loop, err := open()
 278         if err != nil {
 279                 return nil, fmt.Errorf("open child: %w", err)
 280         }
 281
 282         loop.messages = m
 283         loop.tools.messages = m
 284         m.loops[name] = loop
 285         m.ordered = append(m.ordered, loop)
 286
 287         return loop, nil
 288 }
 289
 290 // Child returns only a direct child of parent. Loops live until the daemon
 291 // stops, so the returned loop stays open for the caller's turn.
 292 func (m *DaemonMessages) Child(parent, name string) *TurnLoop {
 293         m.mutex.Lock()
 294         defer m.mutex.Unlock()
 295
 296         loop := m.loops[name]
 297         if loop == nil || loop.agent.config.Parent != parent {
 298                 return nil
 299         }
 300
 301         return loop
 302 }
 303
 304 // ChildState lists the names and responsibilities of a parent's children in
 305 // name order. It returns no heading when the parent has no children.
 306 func (m *DaemonMessages) ChildState(parent string) string {
 307         m.mutex.Lock()
 308         defer m.mutex.Unlock()
 309
 310         var lines []string
 311
 312         for name, loop := range m.loops {
 313                 if loop.agent.config.Parent == parent {
 314                         lines = append(lines, name+": "+loop.agent.config.Responsibility)
 315                 }
 316         }
 317
 318         if len(lines) == 0 {
 319                 return ""
 320         }
 321
 322         slices.Sort(lines)
 323
 324         return "\n\nChildren:\n" + strings.Join(lines, "\n")
 325 }
 326
 327 // SnapshotAgents locks the queue and all registered loops, including children
 328 // made during this daemon run, before snapshot copies their files.
 329 func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() error) error {
 330         return m.Snapshot(func() error {
 331                 return daemonSnapshot(ctx, m.ordered, snapshot)
 332         })
 333 }
 334
 335 // Snapshot holds the queue lock while snapshot locks and copies agent files.
 336 // Its callback must take the memory locks after this lock, as delivery does.
 337 func (m *DaemonMessages) Snapshot(snapshot func() error) error {
 338         m.mutex.Lock()
 339         defer m.mutex.Unlock()
 340
 341         err := snapshot()
 342         if err != nil {
 343                 return fmt.Errorf("copy agent message snapshot: %w", err)
 344         }
 345
 346         return nil
 347 }
 348
 349 // Tell queues a user message from parent to its own child, then wakes the
 350 // child. It is the daemon entry point for parent-to-child messages.
 351 func (m *DaemonMessages) Tell(parent, child, text string) error {
 352         m.mutex.Lock()
 353         defer m.mutex.Unlock()
 354
 355         source := m.loops[parent]
 356         target := m.loops[child]
 357         if source == nil || target == nil || target.agent.config.Parent != parent {
 358                 return errors.New("the recipient is not a child of the sender")
 359         }
 360
 361         if m.failure != nil {
 362                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 363         }
 364
 365         m.add(parent, child, "user", text)
 366
 367         err := m.save()
 368         if err == nil {
 369                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 370         }
 371
 372         if err != nil {
 373                 return fmt.Errorf("send agent message: %w", m.failed(err))
 374         }
 375
 376         return nil
 377 }
 378
 379 // add appends a queue entry under the mutex. Its ID is never reused.
 380 func (m *DaemonMessages) add(source, target, kind, text string) {
 381         m.state.Next++
 382         m.state.Pending = append(m.state.Pending, DaemonMessage{
 383                 ID:     m.state.Next,
 384                 Source: source,
 385                 Target: target,
 386                 Kind:   kind,
 387                 Text:   text,
 388         })
 389 }
 390
 391 // failed stops all loops after a queue failure. Further sends must not change
 392 // the queue, because the outcome of the failed write is not known.
 393 func (m *DaemonMessages) failed(err error) error {
 394         m.failure = err
 395         for _, loop := range m.loops {
 396                 loop.fail(err)
 397         }
 398
 399         return err
 400 }
 401
 402 // RouteWhatsApp queues a routed batch as work and wakes its child. It returns
 403 // false for an unrouted source or a missing child. A saved receipt survives
 404 // turn acknowledgement until the inbox removes the batch. A repeat after a stop
 405 // between queueing and inbox removal cannot append a second entry.
 406 func (m *DaemonMessages) RouteWhatsApp(batch WhatsAppBatch, text string) (bool, error) {
 407         m.mutex.Lock()
 408         defer m.mutex.Unlock()
 409
 410         if m.failure != nil {
 411                 return false, fmt.Errorf("agent message queue stopped: %w", m.failure)
 412         }
 413
 414         if batch.Number <= 0 {
 415                 return false, errors.New("invalid WhatsApp batch number")
 416         }
 417
 418         if slices.Contains(m.state.WhatsAppBatches, batch.Number) {
 419                 return true, nil
 420         }
 421
 422         root := m.ordered[0].agent.config.Name
 423         name := m.routes.Target(whatsappSource + ":" + batch.Chat.ID)
 424         child := m.loops[name]
 425         if child == nil || child.agent.config.Parent != root {
 426                 return false, nil
 427         }
 428
 429         m.add(root, name, "work", text)
 430
 431         message := &m.state.Pending[len(m.state.Pending)-1]
 432         message.WhatsAppBatch = batch.Number
 433         m.state.WhatsAppBatches = append(m.state.WhatsAppBatches, batch.Number)
 434         err := m.save()
 435         if err == nil {
 436                 err = m.deliver(message)
 437         }
 438
 439         if err != nil {
 440                 return false, fmt.Errorf("route WhatsApp batch: %w", m.failed(err))
 441         }
 442
 443         return true, nil
 444 }
 445
 446 // PruneWhatsApp drops receipts only for batches absent from the durable inbox.
 447 // Queued turns keep their own log IDs until they finish. Receipts left by a
 448 // stop after inbox removal are removed by the next inbox delivery.
 449 func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
 450         m.mutex.Lock()
 451         defer m.mutex.Unlock()
 452
 453         if m.failure != nil {
 454                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 455         }
 456
 457         held := make(map[int]bool, len(batches))
 458         for _, batch := range batches {
 459                 held[batch.Number] = true
 460         }
 461
 462         before := len(m.state.WhatsAppBatches)
 463         m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
 464                 return !held[number]
 465         })
 466
 467         if len(m.state.WhatsAppBatches) == before {
 468                 return nil
 469         }
 470
 471         err := m.save()
 472         if err != nil {
 473                 return fmt.Errorf("prune WhatsApp receipts: %w", m.failed(err))
 474         }
 475
 476         return nil
 477 }
 478
 479 // Finish acknowledges the queued messages taken by a turn and queues its final
 480 // report in the same file replacement. A child with no final text sends
 481 // nothing. Call after a final reply, a failure, or a user interrupt, but not
 482 // when a daemon stop interrupts a model request.
 483 func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) error {
 484         m.mutex.Lock()
 485         defer m.mutex.Unlock()
 486
 487         if m.failure != nil {
 488                 return fmt.Errorf("agent message queue stopped: %w", m.failure)
 489         }
 490
 491         ids := make(map[int]bool, len(active))
 492         for _, message := range active {
 493                 ids[message.ID] = true
 494         }
 495
 496         name := loop.agent.config.Name
 497         before := len(m.state.Pending)
 498         m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {
 499                 return message.Target == name && message.LogID != nil && ids[*message.LogID]
 500         })
 501
 502         parent := loop.agent.config.Parent
 503         report := parent != "" && strings.TrimSpace(reply) != ""
 504         if report {
 505                 m.add(name, parent, "work", "["+name+"] "+reply)
 506         }
 507
 508         if !report && len(m.state.Pending) == before {
 509                 return nil
 510         }
 511
 512         err := m.save()
 513         if err == nil && report {
 514                 err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
 515         }
 516
 517         if err != nil {
 518                 return fmt.Errorf("finish agent messages: %w", m.failed(err))
 519         }
 520
 521         return nil
 522 }
 523
 524 // Notify accepts root external data without adding it to memory or an active
 525 // chain. A receipt is durable before success. Exact retries return the same
 526 // sequence even after completion. New keys fail when the receipt store is full.
 527 func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
 528         m.mutex.Lock()
 529         defer m.mutex.Unlock()
 530
 531         if m.failure != nil {
 532                 return Event{}, fmt.Errorf("notification queue stopped: %w", m.failure)
 533         }
 534
 535         if input == nil {
 536                 return Event{}, errors.New("missing notification data")
 537         }
 538
 539         err := notifyValidate(*input)
 540         if err != nil {
 541                 return Event{}, fmt.Errorf("validate external event: %w", err)
 542         }
 543
 544         digest := notifyDigest(input.Text)
 545         for _, receipt := range m.state.Notifications {
 546                 if receipt.Source != input.Source || receipt.ID != input.ID {
 547                         continue
 548                 }
 549
 550                 if receipt.Digest != digest {
 551                         return Event{}, errors.New("notification key conflicts with accepted text")
 552                 }
 553
 554                 return Event{
 555                         Kind:   "notify",
 556                         ID:     receipt.Number,
 557                         Status: "duplicate",
 558                 }, nil
 559         }
 560
 561         if len(m.state.Notifications) >= notifyReceipts {
 562                 return Event{}, errors.New("notification receipt store is full")
 563         }
 564
 565         root := m.ordered[0]
 566         text, err := notifyText(*input, root.clock())
 567         if err != nil {
 568                 return Event{}, fmt.Errorf("prepare external event: %w", err)
 569         }
 570
 571         name := root.agent.config.Name
 572         m.add(name, name, "work", text)
 573
 574         m.state.Pending[len(m.state.Pending)-1].Notification = true
 575         m.state.Notifications = append(m.state.Notifications, NotifyReceipt{
 576                 Source: input.Source,
 577                 ID:     input.ID,
 578                 Digest: digest,
 579                 Number: m.state.Next,
 580         })
 581
 582         err = m.save()
 583         if err != nil {
 584                 return Event{}, fmt.Errorf("accept external event: %w", m.failed(err))
 585         }
 586
 587         select {
 588         case root.wake <- struct{}{}:
 589         default:
 590         }
 591
 592         root.logger.Info("event accepted", "id", m.state.Next)
 593
 594         return Event{
 595                 Kind:   "notify",
 596                 ID:     m.state.Next,
 597                 Status: "accepted",
 598         }, nil
 599 }
 600
 601 // Notifications delivers at most twenty queued events to a root turn. It stops
 602 // before 64 KiB of envelope text, except that the first event always fits. Call
 603 // only between chains, never from take. Events stay queued until Finish, so
 604 // daemon shutdown can repeat model calls and tool effects, but AppendOnce does
 605 // not create duplicate delivery log entries.
 606 func (m *DaemonMessages) Notifications(loop *TurnLoop) error {
 607         m.mutex.Lock()
 608         defer m.mutex.Unlock()
 609
 610         if m.failure != nil {
 611                 return fmt.Errorf("external event queue stopped: %w", m.failure)
 612         }
 613
 614         count := 0
 615         size := 0
 616         for i := range m.state.Pending {
 617                 message := &m.state.Pending[i]
 618                 if !message.Notification || message.Target != loop.agent.config.Name {
 619                         continue
 620                 }
 621
 622                 if count == 20 || (count > 0 && size+len(message.Text) > 64<<10) {
 623                         break
 624                 }
 625
 626                 logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
 627                         message.LogID = &id
 628                         return m.save()
 629                 })
 630                 if err != nil {
 631                         return fmt.Errorf("deliver external event: %w", m.failed(err))
 632                 }
 633
 634                 loop.notify(Event{
 635                         Kind:   "work",
 636                         ID:     logged.ID,
 637                         Text:   logged.Text,
 638                         Source: notifySource,
 639                 })
 640
 641                 loop.Wake(logged)
 642
 643                 count++
 644                 size += len(message.Text)
 645         }
 646
 647         return nil
 648 }