+
+// 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
+}
+
+// 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
+}
projects / onechat / blobdiff