X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/effd32b0f2e4d146b219b1727601b0b7f6593638..98d137c3e945a09f76e85bd18238558c668fdca4:/daemon_message.go?ds=sidebyside

diff --git a/daemon_message.go b/daemon_message.go
index 8ee26e8..f37a88f 100644
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -12,10 +13,54 @@ import (
 	"sync"
 )
 
+// 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.
+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 root external data held outside memory until a turn
+// starts.
+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"`
+}
+
 // 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.
-// A routed batch with a removed child goes to the root.
+// 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) {
 	routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
 	if err != nil {
@@ -28,6 +73,7 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 		ordered: append([]*TurnLoop(nil), loops...),
 		loops:   make(map[string]*TurnLoop),
 	}
+
 	for _, loop := range loops {
 		m.loops[loop.agent.config.Name] = loop
 		loop.messages = m
@@ -73,6 +119,23 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 		}
 	}
 
+	if len(m.state.Notifications) > notifyReceipts {
+		return nil, errors.New("too many notification receipts")
+	}
+
+	receipts := make(map[int]bool)
+	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) {
+			return nil, errors.New("invalid notification receipt")
+		}
+
+		receipts[receipt.Number] = true
+		keys[key] = true
+	}
+
 	previous := 0
 	for i := range m.state.Pending {
 		message := &m.state.Pending[i]
@@ -88,18 +151,47 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 			return nil, errors.New("invalid queued agent message")
 		}
 
-		if message.WhatsAppBatch != 0 {
+		switch {
+		case message.Notification:
+			if source != loops[0] || target != source || message.Kind != "work" || message.WhatsAppBatch != 0 || !receipts[message.ID] {
+				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")
 			}
-		} else if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
+		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")
 		}
 
 		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 {
+		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)
@@ -109,21 +201,6 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
 	return m, nil
 }
 
-// DaemonMessages owns the durable queue between the root and its children and
-// the routed inputs. 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
-}
-
 // 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 {
@@ -151,6 +228,7 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 		Source: source,
 		Status: status,
 	})
+
 	loop.Wake(logged)
 
 	if message.WhatsAppBatch > 0 {
@@ -169,7 +247,7 @@ func (m *DaemonMessages) save() error {
 		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)
 	}
@@ -254,6 +332,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 {
@@ -284,6 +376,29 @@ func (m *DaemonMessages) Tell(parent, child, text string) error {
 	return nil
 }
 
+// add appends a queue entry under the mutex. Its ID is never reused.
+func (m *DaemonMessages) add(source, target, kind, text string) {
+	m.state.Next++
+	m.state.Pending = append(m.state.Pending, DaemonMessage{
+		ID:     m.state.Next,
+		Source: source,
+		Target: target,
+		Kind:   kind,
+		Text:   text,
+	})
+}
+
+// failed stops all loops after a queue failure. Further sends must not change
+// the queue, because the outcome of the failed write is not known.
+func (m *DaemonMessages) failed(err error) error {
+	m.failure = err
+	for _, loop := range m.loops {
+		loop.fail(err)
+	}
+
+	return err
+}
+
 // 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
@@ -348,6 +463,7 @@ func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
 	m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
 		return !held[number]
 	})
+
 	if len(m.state.WhatsAppBatches) == before {
 		return nil
 	}
@@ -360,43 +476,6 @@ func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
 	return nil
 }
 
-// add appends a queue entry under the mutex. Its ID is never reused.
-func (m *DaemonMessages) add(source, target, kind, text string) {
-	m.state.Next++
-	m.state.Pending = append(m.state.Pending, DaemonMessage{
-		ID:     m.state.Next,
-		Source: source,
-		Target: target,
-		Kind:   kind,
-		Text:   text,
-	})
-}
-
-// failed stops all loops after a queue failure. Further sends must not change
-// the queue, because the outcome of the failed write is not known.
-func (m *DaemonMessages) failed(err error) error {
-	m.failure = err
-	for _, loop := range m.loops {
-		loop.fail(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 {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-
-	err := snapshot()
-	if err != nil {
-		return fmt.Errorf("copy agent message snapshot: %w", err)
-	}
-
-	return nil
-}
-
 // Finish acknowledges the queued messages taken by a turn and queues its final
 // report in the same file replacement. A child with no final text sends
 // nothing. Call after a final reply, a failure, or a user interrupt, but not
@@ -442,24 +521,128 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string)
 	return nil
 }
 
-// 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.
-type DaemonMessageFile struct {
-	Next            int             `json:"next"`
-	Pending         []DaemonMessage `json:"pending"`
-	WhatsAppBatches []int           `json:"whatsapp_batches,omitempty"`
+// 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.
+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)
+	}
+
+	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")
+		}
+
+		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")
+	}
+
+	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.state.Pending[len(m.state.Pending)-1].Notification = true
+	m.state.Notifications = append(m.state.Notifications, NotifyReceipt{
+		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 root.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. WhatsAppBatch marks a routed
-// external batch, whose event source is WhatsApp rather than its queue sender.
-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"`
+// 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.
+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 || message.Target != loop.agent.config.Name {
+			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
 }