// 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.
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, 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")
+ }
+ }
+
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.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) {
+ if 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) {
return nil, errors.New("queued agents are not parent and child")
}
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.
+// 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
}
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 nil
}
+// 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()
+
+ 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 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
+}
+
// 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++
return nil
}
-// DaemonMessageFile holds the sequence and the unacknowledged deliveries.
+// 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"`
+ Next int `json:"next"`
+ Pending []DaemonMessage `json:"pending"`
+ WhatsAppBatches []int `json:"whatsapp_batches,omitempty"`
}
// DaemonMessage is one queued delivery. LogID is set before the log append. A
-// nil LogID means that delivery has not started.
+// 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"`
+ 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"`
}
projects / onechat / blobdiff