Share general review and documentation rules

[onechat] / daemon_message.go

diff --git a/daemon_message.go b/daemon_message.go
index f37a88f5aa673a605452d002a658747c4a1c951b..8e36538aa5c3b8075b83e7b328748c7e1fb9381e 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -14,10 +14,10 @@ import (
 )
 
 // DaemonMessages owns the durable queue between the root and its children and
 )
 
 // DaemonMessages owns the durable queue between the root and its children and

-// the routed inputs and root notifications. 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.
+// 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
 type DaemonMessages struct {
        mutex   sync.Mutex
        path    string


@@ -42,8 +42,9 @@ type DaemonMessageFile struct {
 // 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.
 // 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 root external data held outside memory until a turn
-// starts.
+// 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"`
 type DaemonMessage struct {
        ID            int    `json:"id"`
        Source        string `json:"source"`


@@ -53,6 +54,7 @@ type DaemonMessage struct {
        LogID         *int   `json:"log_id,omitempty"`
        WhatsAppBatch int    `json:"whatsapp_batch,omitempty"`
        Notification  bool   `json:"notification,omitempty"`
        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


@@ -123,16 +125,23 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
                return nil, errors.New("too many notification receipts")
        }
 
                return nil, errors.New("too many notification receipts")
        }
 

-       receipts := make(map[int]bool)
+       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)
        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] || keys[key] || !notifyIdentifier(receipt.Source) || !notifyIdentifier(receipt.ID) {
+               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")
                }
 
                        return nil, errors.New("invalid notification receipt")
                }
 

-               receipts[receipt.Number] = true
+               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
        }
 
                keys[key] = true
        }
 


@@ -141,6 +150,19 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
                message := &m.state.Pending[i]
                source := m.loops[message.Source]
                target := m.loops[message.Target]
                message := &m.state.Pending[i]
                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
                if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
                        message.Target = message.Source
                        message.LogID = nil


@@ -153,7 +175,8 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 
                switch {
                case message.Notification:
 
                switch {
                case message.Notification:

-                       if source != loops[0] || target != source || message.Kind != "work" || message.WhatsAppBatch != 0 || !receipts[message.ID] {
+                       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:
                                return nil, errors.New("invalid queued notification")
                        }
                case message.WhatsAppBatch != 0:


@@ -164,6 +187,10 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
                        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
        }
 


@@ -201,6 +228,18 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
        return m, nil
 }
 
        return m, nil
 }
 

+// 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
 // the mutex, or owns the queue before the loops start.
 func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 // deliver records one queued message and wakes its recipient. The caller holds
 // the mutex, or owns the queue before the loops start.
 func (m *DaemonMessages) deliver(message *DaemonMessage) error {


@@ -496,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


@@ -521,9 +565,10 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string)
        return nil
 }
 
        return nil
 }
 

-// Notify accepts root external data without adding it to memory or an active
-// chain. A receipt is durable before success. Exact retries return the same
-// sequence even after completion. New keys fail when the receipt store is full.
+// 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()
 func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
        m.mutex.Lock()
        defer m.mutex.Unlock()


@@ -541,14 +586,28 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
                return Event{}, fmt.Errorf("validate external event: %w", err)
        }
 
                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
                }
 
        digest := notifyDigest(input.Text)
        for _, receipt := range m.state.Notifications {
                if receipt.Source != input.Source || receipt.ID != input.ID {
                        continue
                }
 

-               if receipt.Digest != digest {
-                       return Event{}, errors.New("notification key conflicts with accepted text")
+               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{
                }
 
                return Event{


@@ -562,21 +621,22 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
                return Event{}, errors.New("notification receipt store is full")
        }
 
                return Event{}, errors.New("notification receipt store is full")
        }
 

-       root := m.ordered[0]

        text, err := notifyText(*input, root.clock())
        if err != nil {
                return Event{}, fmt.Errorf("prepare external event: %w", err)
        }
 
        name := root.agent.config.Name
        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, name, "work", text)
+       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].Notification = true

+       m.state.Pending[len(m.state.Pending)-1].TargetID = target.agent.config.ID

        m.state.Notifications = append(m.state.Notifications, NotifyReceipt{
        m.state.Notifications = append(m.state.Notifications, NotifyReceipt{

-               Source: input.Source,
-               ID:     input.ID,
-               Digest: digest,
-               Number: m.state.Next,
+               TargetID: target.agent.config.ID,
+               Source:   input.Source,
+               ID:       input.ID,
+               Digest:   digest,
+               Number:   m.state.Next,

        })
 
        err = m.save()
        })
 
        err = m.save()


@@ -585,7 +645,7 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
        }
 
        select {
        }
 
        select {

-       case root.wake <- struct{}{}:
+       case target.wake <- struct{}{}:

        default:
        }
 
        default:
        }
 


@@ -598,11 +658,11 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
        }, nil
 }
 
        }, nil
 }
 

-// Notifications delivers at most twenty queued events to a root 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.
+// 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()
 func (m *DaemonMessages) Notifications(loop *TurnLoop) error {
        m.mutex.Lock()
        defer m.mutex.Unlock()


@@ -615,7 +675,16 @@ func (m *DaemonMessages) Notifications(loop *TurnLoop) error {
        size := 0
        for i := range m.state.Pending {
                message := &m.state.Pending[i]
        size := 0
        for i := range m.state.Pending {
                message := &m.state.Pending[i]

-               if !message.Notification || message.Target != loop.agent.config.Name {
+               if !message.Notification {
+                       continue
+               }
+
+               target := message.Target == loop.agent.config.Name
+               if message.TargetID != "" {
+                       target = message.TargetID == loop.agent.config.ID
+               }
+
+               if !target {

                        continue
                }
 
                        continue
                }