Bind external events to permanent agent IDs

[onechat] / daemon_message.go

diff --git a/daemon_message.go b/daemon_message.go
index 8415e49d4b7110d596eaed1013e79e6f29ce09b7..8e36538aa5c3b8075b83e7b328748c7e1fb9381e 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -2,6 +2,7 @@ package main
 
 import (
        "context"
 
 import (
        "context"

+       "encoding/hex"

        "encoding/json"
        "errors"
        "fmt"
        "encoding/json"
        "errors"
        "fmt"


@@ -12,21 +13,82 @@ import (
        "sync"
 )
 
        "sync"
 )
 

+// DaemonMessages owns the durable queue between the root and its children and
+// the routed inputs and notifications for permanent agents. The mutex
+// serializes queue writes and delivery. Backups take it before the locks of the
+// loops, memory, and subagents. A stored log ID makes delivery idempotent,
+// including a stop between the log append and the wake.
+type DaemonMessages struct {
+       mutex   sync.Mutex
+       path    string
+       loops   map[string]*TurnLoop
+       ordered []*TurnLoop
+       state   DaemonMessageFile
+       failure error
+       routes  *Routes
+}
+
+// DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
+// for WhatsApp batches until the inbox removes them. A receipt stays when its
+// turn finishes before inbox removal. Notification receipts stay after their
+// turn finishes and do not expire in the local test interface.
+type DaemonMessageFile struct {
+       Next            int             `json:"next"`
+       Pending         []DaemonMessage `json:"pending"`
+       WhatsAppBatches []int           `json:"whatsapp_batches,omitempty"`
+       Notifications   []NotifyReceipt `json:"notifications,omitempty"`
+}
+
+// DaemonMessage is one queued delivery. LogID is set before the log append. A
+// nil LogID means that delivery has not started. WhatsAppBatch marks a routed
+// external batch, whose event source is WhatsApp rather than its queue sender.
+// Notification marks external data held outside memory until a turn starts.
+// TargetID binds a notification to its permanent agent across name changes. An
+// empty TargetID identifies a root notification.
+type DaemonMessage struct {
+       ID            int    `json:"id"`
+       Source        string `json:"source"`
+       Target        string `json:"target"`
+       Kind          string `json:"kind"`
+       Text          string `json:"text"`
+       LogID         *int   `json:"log_id,omitempty"`
+       WhatsAppBatch int    `json:"whatsapp_batch,omitempty"`
+       Notification  bool   `json:"notification,omitempty"`
+       TargetID      string `json:"target_id,omitempty"`
+}
+

 // openDaemonMessages connects the root and its children. It restores queued
 // openDaemonMessages connects the root and its children. It restores queued

-// messages before the loops start. A queue entry stays until its recipient
-// finishes the turn that takes it, so a stop can interrupt delivery or a turn.
+// messages before the loops start. External events without a saved log ID stay
+// unlogged until their turn boundary. Prepared log IDs are restored before
+// startup writers can use them, without waking a turn. A queue entry stays
+// until its recipient finishes the turn that takes it, so a stop can interrupt
+// delivery or a turn. A routed batch with a removed child goes to the root.

 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {

+       routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
+       if err != nil {
+               return nil, fmt.Errorf("open agent routes: %w", err)
+       }
+

        m := &DaemonMessages{
        m := &DaemonMessages{

+               routes:  routes,

                path:    filepath.Join(loops[0].agent.dir, "messages.json"),
                ordered: append([]*TurnLoop(nil), loops...),
                loops:   make(map[string]*TurnLoop),
        }
                path:    filepath.Join(loops[0].agent.dir, "messages.json"),
                ordered: append([]*TurnLoop(nil), loops...),
                loops:   make(map[string]*TurnLoop),
        }

+

        for _, loop := range loops {
                m.loops[loop.agent.config.Name] = loop
                loop.messages = m
                loop.tools.messages = m
                if loop.agent.config.Parent == "" {
                        loop.logger = loop.logger.With("name", loop.agent.config.Name)
        for _, loop := range loops {
                m.loops[loop.agent.config.Name] = loop
                loop.messages = m
                loop.tools.messages = m
                if loop.agent.config.Parent == "" {
                        loop.logger = loop.logger.With("name", loop.agent.config.Name)

+                       loop.tools.routes = routes
+                       if loop.tools.whatsapp != nil {
+                               loop.tools.whatsapp.mutex.Lock()
+
+                               loop.tools.whatsapp.routes = routes
+                               loop.tools.whatsapp.mutex.Unlock()
+                       }

                }
        }
 
                }
        }
 


@@ -53,22 +115,110 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
                return nil, errors.New("invalid agent message sequence")
        }
 
                return nil, errors.New("invalid agent message sequence")
        }
 

+       for i, number := range m.state.WhatsAppBatches {
+               if number <= 0 || slices.Contains(m.state.WhatsAppBatches[:i], number) {
+                       return nil, errors.New("invalid WhatsApp batch receipt")
+               }
+       }
+
+       if len(m.state.Notifications) > notifyReceipts {
+               return nil, errors.New("too many notification receipts")
+       }
+
+       receipts := make(map[int]NotifyReceipt)
+       keys := make(map[[2]string]bool)
+       for _, receipt := range m.state.Notifications {
+               key := [2]string{receipt.Source, receipt.ID}
+               digest, err := hex.DecodeString(receipt.Digest)
+               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) {
+                       return nil, errors.New("invalid notification receipt")
+               }
+
+               if receipt.TargetID != "" {
+                       id, err := hex.DecodeString(receipt.TargetID)
+                       if err != nil || len(id) != agentIDBytes {
+                               return nil, errors.New("invalid notification target identity")
+                       }
+               }
+
+               receipts[receipt.Number] = receipt
+               keys[key] = true
+       }
+

        previous := 0
        previous := 0

-       for _, message := range m.state.Pending {
+       for i := range m.state.Pending {
+               message := &m.state.Pending[i]

                source := m.loops[message.Source]
                target := m.loops[message.Target]
                source := m.loops[message.Source]
                target := m.loops[message.Target]

+               if message.Notification {
+                       target = loops[0]
+                       if message.TargetID != "" {
+                               target = m.notificationTarget(message.TargetID)
+                       }
+
+                       if target == nil {
+                               return nil, errors.New("queued notification target is missing")
+                       }
+
+                       message.Target = target.agent.config.Name
+               }
+
+               if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
+                       message.Target = message.Source
+                       message.LogID = nil
+                       target = source
+               }
+

                if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
                        return nil, errors.New("invalid queued agent message")
                }
 
                if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
                        return nil, errors.New("invalid queued agent message")
                }
 

-               if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
+               switch {
+               case message.Notification:
+                       receipt := receipts[message.ID]
+                       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 {
+                               return nil, errors.New("invalid queued notification")
+                       }
+               case message.WhatsAppBatch != 0:
+                       if message.WhatsAppBatch < 0 || message.Kind != "work" || source != loops[0] || (target != source && target.agent.config.Parent != message.Source) {
+                               return nil, errors.New("invalid queued WhatsApp batch")
+                       }
+               case (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target):

                        return nil, errors.New("queued agents are not parent and child")
                }
 
                        return nil, errors.New("queued agents are not parent and child")
                }
 

+               if !message.Notification && message.TargetID != "" {
+                       return nil, errors.New("agent message has a notification target identity")
+               }
+

                previous = message.ID
        }
 
                previous = message.ID
        }
 

+       // Complete prepared log reservations before another restored delivery
+       // or startup writer can use their IDs. They belong to an interrupted
+       // turn, not newly accepted data. Only the next turn boundary wakes
+       // them.

        for i := range m.state.Pending {
        for i := range m.state.Pending {

+               message := &m.state.Pending[i]
+               if !message.Notification || message.LogID == nil {
+                       continue
+               }
+
+               loop := m.loops[message.Target]
+               _, err = loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
+                       message.LogID = &id
+                       return m.save()
+               })
+               if err != nil {
+                       return nil, fmt.Errorf("restore external event reservation: %w", err)
+               }
+       }
+
+       for i := range m.state.Pending {
+               if m.state.Pending[i].Notification {
+                       continue
+               }
+

                err = m.deliver(&m.state.Pending[i])
                if err != nil {
                        return nil, fmt.Errorf("restore agent message: %w", err)
                err = m.deliver(&m.state.Pending[i])
                if err != nil {
                        return nil, fmt.Errorf("restore agent message: %w", err)


@@ -78,17 +228,16 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
        return m, nil
 }
 
        return m, nil
 }
 

-// DaemonMessages owns the durable queue between the root and its children. The
-// mutex serializes queue writes and delivery. Backups take it before the locks
-// of the loops, memory, and subagents. A stored log ID makes delivery
-// idempotent, including a stop between the log append and the wake.
-type DaemonMessages struct {
-       mutex   sync.Mutex
-       path    string
-       loops   map[string]*TurnLoop
-       ordered []*TurnLoop
-       state   DaemonMessageFile
-       failure error
+// notificationTarget resolves a stored identity without accepting names or
+// task-worker identities. The caller holds the queue lock or owns startup.
+func (m *DaemonMessages) notificationTarget(id string) *TurnLoop {
+       for _, loop := range m.ordered {
+               if strings.EqualFold(loop.agent.config.ID, id) {
+                       return loop
+               }
+       }
+
+       return nil

 }
 
 // deliver records one queued message and wakes its recipient. The caller holds
 }
 
 // deliver records one queued message and wakes its recipient. The caller holds


@@ -104,7 +253,10 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
        }
 
        status := ""
        }
 
        status := ""

-       if message.Kind == "work" {
+       source := message.Source
+       if message.WhatsAppBatch > 0 {
+               source = whatsappSource
+       } else if message.Kind == "work" {

                status = subagentFinished
        }
 
                status = subagentFinished
        }
 


@@ -112,11 +264,17 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
                Kind:   message.Kind,
                ID:     logged.ID,
                Text:   message.Text,
                Kind:   message.Kind,
                ID:     logged.ID,
                Text:   message.Text,

-               Source: message.Source,
+               Source: source,

                Status: status,
        })
                Status: status,
        })

+

        loop.Wake(logged)
        loop.Wake(logged)

-       loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+
+       if message.WhatsAppBatch > 0 {
+               loop.logger.Info("batch routed", "kind", whatsappSource, "id", message.WhatsAppBatch, "to", message.Target)
+       } else {
+               loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+       }

 
        return nil
 }
 
        return nil
 }


@@ -128,7 +286,7 @@ func (m *DaemonMessages) save() error {
                return fmt.Errorf("encode agent messages: %w", err)
        }
 
                return fmt.Errorf("encode agent messages: %w", err)
        }
 

-       err = writeFileAtomic(m.path, append(data, '\n'))
+       err = writeFileDurable(m.path, append(data, '\n'))

        if err != nil {
                return fmt.Errorf("save agent messages: %w", err)
        }
        if err != nil {
                return fmt.Errorf("save agent messages: %w", err)
        }


@@ -213,6 +371,20 @@ func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() err
        })
 }
 
        })
 }
 

+// Snapshot holds the queue lock while snapshot locks and copies agent files.
+// Its callback must take the memory locks after this lock, as delivery does.
+func (m *DaemonMessages) Snapshot(snapshot func() error) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       err := snapshot()
+       if err != nil {
+               return fmt.Errorf("copy agent message snapshot: %w", err)
+       }
+
+       return nil
+}
+

 // Tell queues a user message from parent to its own child, then wakes the
 // child. It is the daemon entry point for parent-to-child messages.
 func (m *DaemonMessages) Tell(parent, child, text string) error {
 // Tell queues a user message from parent to its own child, then wakes the
 // child. It is the daemon entry point for parent-to-child messages.
 func (m *DaemonMessages) Tell(parent, child, text string) error {


@@ -266,15 +438,78 @@ func (m *DaemonMessages) failed(err error) error {
        return err
 }
 
        return err
 }
 

-// Snapshot holds the queue lock while snapshot locks and copies agent files.
-// Its callback must take the memory locks after this lock, as delivery does.
-func (m *DaemonMessages) Snapshot(snapshot func() error) error {
+// RouteWhatsApp queues a routed batch as work and wakes its child. It returns
+// false for an unrouted source or a missing child. A saved receipt survives
+// turn acknowledgement until the inbox removes the batch. A repeat after a stop
+// between queueing and inbox removal cannot append a second entry.
+func (m *DaemonMessages) RouteWhatsApp(batch WhatsAppBatch, text string) (bool, error) {

        m.mutex.Lock()
        defer m.mutex.Unlock()
 
        m.mutex.Lock()
        defer m.mutex.Unlock()
 

-       err := snapshot()
+       if m.failure != nil {
+               return false, fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       if batch.Number <= 0 {
+               return false, errors.New("invalid WhatsApp batch number")
+       }
+
+       if slices.Contains(m.state.WhatsAppBatches, batch.Number) {
+               return true, nil
+       }
+
+       root := m.ordered[0].agent.config.Name
+       name := m.routes.Target(whatsappSource + ":" + batch.Chat.ID)
+       child := m.loops[name]
+       if child == nil || child.agent.config.Parent != root {
+               return false, nil
+       }
+
+       m.add(root, name, "work", text)
+
+       message := &m.state.Pending[len(m.state.Pending)-1]
+       message.WhatsAppBatch = batch.Number
+       m.state.WhatsAppBatches = append(m.state.WhatsAppBatches, batch.Number)
+       err := m.save()
+       if err == nil {
+               err = m.deliver(message)
+       }
+

        if err != nil {
        if err != nil {

-               return fmt.Errorf("copy agent message snapshot: %w", err)
+               return false, fmt.Errorf("route WhatsApp batch: %w", m.failed(err))
+       }
+
+       return true, nil
+}
+
+// PruneWhatsApp drops receipts only for batches absent from the durable inbox.
+// Queued turns keep their own log IDs until they finish. Receipts left by a
+// stop after inbox removal are removed by the next inbox delivery.
+func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       held := make(map[int]bool, len(batches))
+       for _, batch := range batches {
+               held[batch.Number] = true
+       }
+
+       before := len(m.state.WhatsAppBatches)
+       m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
+               return !held[number]
+       })
+
+       if len(m.state.WhatsAppBatches) == before {
+               return nil
+       }
+
+       err := m.save()
+       if err != nil {
+               return fmt.Errorf("prune WhatsApp receipts: %w", m.failed(err))

        }
 
        return nil
        }
 
        return nil


@@ -300,7 +535,12 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string)
        name := loop.agent.config.Name
        before := len(m.state.Pending)
        m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {
        name := loop.agent.config.Name
        before := len(m.state.Pending)
        m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {

-               return message.Target == name && message.LogID != nil && ids[*message.LogID]
+               target := message.Target == name
+               if message.Notification && message.TargetID != "" {
+                       target = message.TargetID == loop.agent.config.ID
+               }
+
+               return target && message.LogID != nil && ids[*message.LogID]

        })
 
        parent := loop.agent.config.Parent
        })
 
        parent := loop.agent.config.Parent


@@ -325,19 +565,153 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string)
        return nil
 }
 
        return nil
 }
 

-// DaemonMessageFile holds the sequence and the unacknowledged deliveries.
-type DaemonMessageFile struct {
-       Next    int             `json:"next"`
-       Pending []DaemonMessage `json:"pending"`
+// Notify accepts external data for the root or a permanent child without adding
+// it to memory or an active chain. A receipt is durable before success. Exact
+// retries return the same sequence even after completion. Retry keys bind text
+// and permanent agent ID. New keys fail when the receipt store is full.
+func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return Event{}, fmt.Errorf("notification queue stopped: %w", m.failure)
+       }
+
+       if input == nil {
+               return Event{}, errors.New("missing notification data")
+       }
+
+       err := notifyValidate(*input)
+       if err != nil {
+               return Event{}, fmt.Errorf("validate external event: %w", err)
+       }
+
+       root := m.ordered[0]
+       target := root
+       if input.AgentID != "" {
+               target = m.notificationTarget(input.AgentID)
+               if target == nil {
+                       return Event{}, errors.New("notification target identity is unknown")
+               }
+       }
+
+       digest := notifyDigest(input.Text)
+       for _, receipt := range m.state.Notifications {
+               if receipt.Source != input.Source || receipt.ID != input.ID {
+                       continue
+               }
+
+               targetID := receipt.TargetID
+               if targetID == "" {
+                       targetID = root.agent.config.ID
+               }
+
+               if receipt.Digest != digest || targetID != target.agent.config.ID {
+                       return Event{}, errors.New("notification key conflicts with accepted target or text")
+               }
+
+               return Event{
+                       Kind:   "notify",
+                       ID:     receipt.Number,
+                       Status: "duplicate",
+               }, nil
+       }
+
+       if len(m.state.Notifications) >= notifyReceipts {
+               return Event{}, errors.New("notification receipt store is full")
+       }
+
+       text, err := notifyText(*input, root.clock())
+       if err != nil {
+               return Event{}, fmt.Errorf("prepare external event: %w", err)
+       }
+
+       name := root.agent.config.Name
+       m.add(name, target.agent.config.Name, "work", text)
+
+       m.state.Pending[len(m.state.Pending)-1].Notification = true
+       m.state.Pending[len(m.state.Pending)-1].TargetID = target.agent.config.ID
+       m.state.Notifications = append(m.state.Notifications, NotifyReceipt{
+               TargetID: target.agent.config.ID,
+               Source:   input.Source,
+               ID:       input.ID,
+               Digest:   digest,
+               Number:   m.state.Next,
+       })
+
+       err = m.save()
+       if err != nil {
+               return Event{}, fmt.Errorf("accept external event: %w", m.failed(err))
+       }
+
+       select {
+       case target.wake <- struct{}{}:
+       default:
+       }
+
+       root.logger.Info("event accepted", "id", m.state.Next)
+
+       return Event{
+               Kind:   "notify",
+               ID:     m.state.Next,
+               Status: "accepted",
+       }, nil

 }
 
 }
 

-// DaemonMessage is one queued delivery. LogID is set before the log append. A
-// nil LogID means that delivery has not started.
-type DaemonMessage struct {
-       ID     int    `json:"id"`
-       Source string `json:"source"`
-       Target string `json:"target"`
-       Kind   string `json:"kind"`
-       Text   string `json:"text"`
-       LogID  *int   `json:"log_id,omitempty"`
+// Notifications delivers at most twenty queued events to a recipient turn. It
+// stops before 64 KiB of envelope text, except that the first event always
+// fits. Call only between chains, never from take. Events stay queued until
+// Finish, so daemon shutdown can repeat model calls and tool effects, but
+// AppendOnce does not create duplicate delivery log entries.
+func (m *DaemonMessages) Notifications(loop *TurnLoop) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return fmt.Errorf("external event queue stopped: %w", m.failure)
+       }
+
+       count := 0
+       size := 0
+       for i := range m.state.Pending {
+               message := &m.state.Pending[i]
+               if !message.Notification {
+                       continue
+               }
+
+               target := message.Target == loop.agent.config.Name
+               if message.TargetID != "" {
+                       target = message.TargetID == loop.agent.config.ID
+               }
+
+               if !target {
+                       continue
+               }
+
+               if count == 20 || (count > 0 && size+len(message.Text) > 64<<10) {
+                       break
+               }
+
+               logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
+                       message.LogID = &id
+                       return m.save()
+               })
+               if err != nil {
+                       return fmt.Errorf("deliver external event: %w", m.failed(err))
+               }
+
+               loop.notify(Event{
+                       Kind:   "work",
+                       ID:     logged.ID,
+                       Text:   logged.Text,
+                       Source: notifySource,
+               })
+
+               loop.Wake(logged)
+
+               count++
+               size += len(message.Text)
+       }
+
+       return nil

 }
 }