)
// 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
// 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"`
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
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)
- 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")
}
- 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
}
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
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("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
}
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 {
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
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()
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
}
- 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{}, 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
- 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].TargetID = target.agent.config.ID
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()
}
select {
- case root.wake <- struct{}{}:
+ case target.wake <- struct{}{}:
default:
}
}, 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()
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
}
projects / onechat / blobdiff