Sync agent queue replacements before success

[onechat] / daemon_message.go

diff --git a/daemon_message.go b/daemon_message.go
index b70947d7a0e934e3b44fa15b01d7574b7afabb05..59733be7022e89ce73fe433afe2866ff16651c4a 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -65,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
@@ -173,6 +174,7 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
                Source: source,
                Status: status,
        })
+
        loop.Wake(logged)
 
        if message.WhatsAppBatch > 0 {
@@ -191,7 +193,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)
        }
@@ -276,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 {
@@ -306,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
@@ -370,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
        }
@@ -382,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