Prove direct cancellation during a pending sync

[onechat] / daemon_message.go

diff --git a/daemon_message.go b/daemon_message.go
index 8ee26e83b4a5d1cbcf5692cefe8d8331494a85cd..7665678468186a29b88e53ef38b1ddbad79d57e0 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -12,6 +12,43 @@ import (
        "sync"
 )
 
        "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.
 // 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),
        }
                ordered: append([]*TurnLoop(nil), loops...),
                loops:   make(map[string]*TurnLoop),
        }

+

        for _, loop := range loops {
                m.loops[loop.agent.config.Name] = loop
                loop.messages = m
        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
 }
 
        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 {
 // 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,
        })
                Source: source,
                Status: status,
        })

+

        loop.Wake(logged)
 
        if message.WhatsAppBatch > 0 {
        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 {
 // 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
 }
 
        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
 // 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]
        })
        m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
                return !held[number]
        })

+

        if len(m.state.WhatsAppBatches) == before {
                return nil
        }
        if len(m.state.WhatsAppBatches) == before {
                return nil
        }


@@ -360,43 +422,6 @@ func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
        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
 // 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
 }
 
        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"`
-}