X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/d7bc0a6ac452bfc646610fdfd7c8eccb5c55218f..f4509310b5abe5012cb83bed3960404e8fa92217:/daemon_message.go diff --git a/daemon_message.go b/daemon_message.go index f37a88f..8e36538 100644 --- 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 -// 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 @@ -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. -// 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"` @@ -53,6 +54,7 @@ type DaemonMessage struct { 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 @@ -123,16 +125,23 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 } @@ -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] + 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 @@ -153,7 +175,8 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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: @@ -164,6 +187,10 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 } @@ -201,6 +228,18 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 { @@ -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 { - 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 @@ -521,9 +565,10 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) 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() @@ -541,14 +586,28 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { 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{ @@ -562,21 +621,22 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { 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() @@ -585,7 +645,7 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { } select { - case root.wake <- struct{}{}: + case target.wake <- struct{}{}: default: } @@ -598,11 +658,11 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { }, 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() @@ -615,7 +675,16 @@ func (m *DaemonMessages) Notifications(loop *TurnLoop) error { 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 }