X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/effd32b0f2e4d146b219b1727601b0b7f6593638..03ef9525ba0e62037b55606b851ee746967a003f:/daemon_message.go diff --git a/daemon_message.go b/daemon_message.go index 8ee26e8..8e36538 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,56 @@ import ( "sync" ) +// DaemonMessages owns the durable queue between the root and its children and +// 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 + 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 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"` + 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"` + TargetID string `json:"target_id,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 +75,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,11 +121,48 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { } } + if len(m.state.Notifications) > notifyReceipts { + return nil, errors.New("too many notification receipts") + } + + 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].Number != 0 || keys[key] || !notifyIdentifier(receipt.Source) || !notifyIdentifier(receipt.ID) { + return nil, errors.New("invalid notification receipt") + } + + 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 + } + previous := 0 for i := range m.state.Pending { 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 @@ -88,18 +173,52 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { return nil, errors.New("invalid queued agent message") } - if message.WhatsAppBatch != 0 { + switch { + case message.Notification: + 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: 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") } + if !message.Notification && message.TargetID != "" { + return nil, errors.New("agent message has a notification target identity") + } + 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,19 +228,16 @@ 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 +// 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 @@ -151,6 +267,7 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error { Source: source, Status: status, }) + loop.Wake(logged) if message.WhatsAppBatch > 0 { @@ -169,7 +286,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 +371,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 +415,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 +502,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 +515,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 @@ -417,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 @@ -442,24 +565,153 @@ 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 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 } -// 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 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 }