"sync"
)
+// 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
+}
+
+// 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"`
+}
+
+// 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"`
+}
+
// 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.
ordered: append([]*TurnLoop(nil), loops...),
loops: make(map[string]*TurnLoop),
}
+
for _, loop := range loops {
m.loops[loop.agent.config.Name] = loop
loop.messages = m
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 {
Source: source,
Status: status,
})
+
loop.Wake(logged)
if message.WhatsAppBatch > 0 {
})
}
+// 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 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
m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
return !held[number]
})
+
if len(m.state.WhatsAppBatches) == before {
return nil
}
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
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"`
-}
-
-// 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"`
-}
projects / onechat / blobdiff