X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/effd32b0f2e4d146b219b1727601b0b7f6593638..b961d7cdc59213673c3e4f86d3dae99daf0a1935:/daemon_message.go?ds=inline

diff --git a/daemon_message.go b/daemon_message.go
index 8ee26e8..7665678 100644
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -12,6 +12,43 @@ import (
 	"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.
@@ -28,6 +65,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
@@ -109,21 +147,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 +174,7 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
 		Source: source,
 		Status: status,
 	})
+
 	loop.Wake(logged)
 
 	if message.WhatsAppBatch > 0 {
@@ -254,6 +278,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 +322,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 +409,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 +422,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
@@ -441,25 +466,3 @@ 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"`
-}
-
-// 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"`
-}