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