package main
import (
+ "context"
+ "encoding/hex"
"encoding/json"
"errors"
"fmt"
"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.
+// 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 {
+ return nil, fmt.Errorf("open agent routes: %w", err)
+ }
+
m := &DaemonMessages{
- path: filepath.Join(loops[0].agent.dir, "messages.json"),
- loops: make(map[string]*TurnLoop),
+ routes: routes,
+ path: filepath.Join(loops[0].agent.dir, "messages.json"),
+ ordered: append([]*TurnLoop(nil), loops...),
+ loops: make(map[string]*TurnLoop),
}
+
for _, loop := range loops {
m.loops[loop.agent.config.Name] = loop
loop.messages = m
+ loop.tools.messages = m
if loop.agent.config.Parent == "" {
loop.logger = loop.logger.With("name", loop.agent.config.Name)
+ loop.tools.routes = routes
+ if loop.tools.whatsapp != nil {
+ loop.tools.whatsapp.mutex.Lock()
+
+ loop.tools.whatsapp.routes = routes
+ loop.tools.whatsapp.mutex.Unlock()
+ }
+ }
+ }
+
+ for _, child := range loops[1:] {
+ err := loops[0].runs.subagents.Reserve(child.agent.config.Name)
+ if err != nil {
+ return nil, fmt.Errorf("reserve child name: %w", err)
}
}
return nil, errors.New("invalid agent message sequence")
}
+ for i, number := range m.state.WhatsAppBatches {
+ if number <= 0 || slices.Contains(m.state.WhatsAppBatches[:i], number) {
+ return nil, errors.New("invalid WhatsApp batch receipt")
+ }
+ }
+
+ 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 _, message := range m.state.Pending {
+ 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
+ target = source
+ }
+
if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
return nil, errors.New("invalid queued agent message")
}
- if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
+ 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")
+ }
+ 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)
return m, nil
}
-// DaemonMessages owns the durable queue between the root and its children. 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
- state DaemonMessageFile
- failure error
+// 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
}
status := ""
- if message.Kind == "work" {
+ source := message.Source
+ if message.WhatsAppBatch > 0 {
+ source = whatsappSource
+ } else if message.Kind == "work" {
status = subagentFinished
}
Kind: message.Kind,
ID: logged.ID,
Text: message.Text,
- Source: message.Source,
+ Source: source,
Status: status,
})
+
loop.Wake(logged)
- loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+
+ if message.WhatsAppBatch > 0 {
+ loop.logger.Info("batch routed", "kind", whatsappSource, "id", message.WhatsAppBatch, "to", message.Target)
+ } else {
+ loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+ }
return nil
}
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)
}
return nil
}
+// AddChild holds the queue lock while open makes a child's files and loop. The
+// loop joins backups before its first task can be delivered.
+func (m *DaemonMessages) AddChild(parent, name string, open func() (*TurnLoop, error)) (*TurnLoop, error) {
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ if m.loops[name] != nil {
+ return nil, fmt.Errorf("name %q is already used", name)
+ }
+
+ root := m.loops[parent]
+ if root == nil || root.agent.config.Parent != "" {
+ return nil, errors.New("only the root can create children")
+ }
+
+ if m.failure != nil {
+ return nil, fmt.Errorf("agent message queue stopped: %w", m.failure)
+ }
+
+ loop, err := open()
+ if err != nil {
+ return nil, fmt.Errorf("open child: %w", err)
+ }
+
+ loop.messages = m
+ loop.tools.messages = m
+ m.loops[name] = loop
+ m.ordered = append(m.ordered, loop)
+
+ return loop, nil
+}
+
+// Child returns only a direct child of parent. Loops live until the daemon
+// stops, so the returned loop stays open for the caller's turn.
+func (m *DaemonMessages) Child(parent, name string) *TurnLoop {
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ loop := m.loops[name]
+ if loop == nil || loop.agent.config.Parent != parent {
+ return nil
+ }
+
+ return loop
+}
+
+// ChildState lists the names and responsibilities of a parent's children in
+// name order. It returns no heading when the parent has no children.
+func (m *DaemonMessages) ChildState(parent string) string {
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ var lines []string
+
+ for name, loop := range m.loops {
+ if loop.agent.config.Parent == parent {
+ lines = append(lines, name+": "+loop.agent.config.Responsibility)
+ }
+ }
+
+ if len(lines) == 0 {
+ return ""
+ }
+
+ slices.Sort(lines)
+
+ return "\n\nChildren:\n" + strings.Join(lines, "\n")
+}
+
+// SnapshotAgents locks the queue and all registered loops, including children
+// made during this daemon run, before snapshot copies their files.
+func (m *DaemonMessages) SnapshotAgents(ctx context.Context, snapshot func() error) error {
+ return m.Snapshot(func() error {
+ return daemonSnapshot(ctx, m.ordered, snapshot)
+ })
+}
+
+// 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 {
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 {
+// 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
+// between queueing and inbox removal cannot append a second entry.
+func (m *DaemonMessages) RouteWhatsApp(batch WhatsAppBatch, text string) (bool, error) {
m.mutex.Lock()
defer m.mutex.Unlock()
- err := snapshot()
+ if m.failure != nil {
+ return false, fmt.Errorf("agent message queue stopped: %w", m.failure)
+ }
+
+ if batch.Number <= 0 {
+ return false, errors.New("invalid WhatsApp batch number")
+ }
+
+ if slices.Contains(m.state.WhatsAppBatches, batch.Number) {
+ return true, nil
+ }
+
+ root := m.ordered[0].agent.config.Name
+ name := m.routes.Target(whatsappSource + ":" + batch.Chat.ID)
+ child := m.loops[name]
+ if child == nil || child.agent.config.Parent != root {
+ return false, nil
+ }
+
+ m.add(root, name, "work", text)
+
+ message := &m.state.Pending[len(m.state.Pending)-1]
+ message.WhatsAppBatch = batch.Number
+ m.state.WhatsAppBatches = append(m.state.WhatsAppBatches, batch.Number)
+ err := m.save()
+ if err == nil {
+ err = m.deliver(message)
+ }
+
if err != nil {
- return fmt.Errorf("copy agent message snapshot: %w", err)
+ return false, fmt.Errorf("route WhatsApp batch: %w", m.failed(err))
+ }
+
+ return true, nil
+}
+
+// PruneWhatsApp drops receipts only for batches absent from the durable inbox.
+// Queued turns keep their own log IDs until they finish. Receipts left by a
+// stop after inbox removal are removed by the next inbox delivery.
+func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ if m.failure != nil {
+ return fmt.Errorf("agent message queue stopped: %w", m.failure)
+ }
+
+ held := make(map[int]bool, len(batches))
+ for _, batch := range batches {
+ held[batch.Number] = true
+ }
+
+ before := len(m.state.WhatsAppBatches)
+ m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
+ return !held[number]
+ })
+
+ if len(m.state.WhatsAppBatches) == before {
+ return nil
+ }
+
+ err := m.save()
+ if err != nil {
+ return fmt.Errorf("prune WhatsApp receipts: %w", m.failed(err))
}
return nil
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
}
-// DaemonMessageFile holds the sequence and the unacknowledged deliveries.
-type DaemonMessageFile struct {
- Next int `json:"next"`
- Pending []DaemonMessage `json:"pending"`
+// 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.
-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"`
+// 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