Deliver messages between parent and child agents

A parent can queue a user message for its own child through the daemon.
The child's final reply queues a work message for the parent with the
child's name in brackets. Each delivery wakes its recipient, including
one that is busy, and an empty final reply sends nothing.

The durable queue keeps inputs until their turn completes and saves a
child report with its input acknowledgments. Saved log positions prevent
repeat entries after restart. Interrupted turns can repeat model calls
and tools. Queue failures stop the loops, and backups lock the queue
before the agent files. Turn logs name their agent. Tools and prompts
stay unchanged.

diff --git a/README.rst b/README.rst
index 473d1e53a3ddbf8811cc60664700146af9ed69e4..b60f984379791785347d2cdfea98ad15da77dc0f 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -150,10 +150,20 @@ Each child has its own memory, model client, usage file, subagents,
 schedules, and turn loop. Its prompt limits work to its area, asks for
 only reports that need the parent or the user, and permits an empty
 reply. Children have no WhatsApp tools and draft replies instead of
 schedules, and turn loop. Its prompt limits work to its area, asks for
 only reports that need the parent or the user, and permits an empty
 reply. Children have no WhatsApp tools and draft replies instead of

-sending them. Clients and WhatsApp still address only the root. There
-are no messages or tools between parent and child agents. A child's
-saved schedules and subagents can resume at startup. The ready log line
-lists the child names.
+sending them. Clients and WhatsApp still address only the root. A
+child's saved schedules and subagents can resume at startup. The ready
+log line lists the child names.
+
+A message from a parent to its own child enters the child's log as
+``user`` and wakes it. A child's final reply reaches the parent as
+``work`` with the prefix ``[NAME] `` and wakes it. An empty reply sends
+nothing. A busy recipient takes the message between tool calls or in its
+next turn. ``agent/messages.json``, with mode 0600, keeps messages until
+their recipient finishes its turn. A daemon stop preserves queued
+messages and reports for the next start without a second log entry. An
+interrupted turn can repeat its model calls and tools after restart.
+Backups lock this queue with the agent files. No tools send messages
+between parent and child agents yet.

 
 **************
  User service
 
 **************
  User service

diff --git a/daemon.go b/daemon.go
index 3a54b1ae94374c7d53c3938f1eeff0ae8fcb9768..958d6f7bd2283fe8b2796c14ae345e7b5f14e604 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -364,8 +364,15 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w
                loops = append(loops, loop)
        }
 
                loops = append(loops, loop)
        }
 

+       messages, err := openDaemonMessages(loops)
+       if err != nil {
+               return errors.Join(fmt.Errorf("open agent messages: %w", err), listener.Close())
+       }
+

        hub.snapshot = func(ctx context.Context, snapshot func() error) error {
        hub.snapshot = func(ctx context.Context, snapshot func() error) error {

-               return daemonSnapshot(ctx, loops, snapshot)
+               return messages.Snapshot(func() error {
+                       return daemonSnapshot(ctx, loops, snapshot)
+               })

        }
 
        var workers sync.WaitGroup
        }
 
        var workers sync.WaitGroup

diff --git a/daemon_agent_test.go b/daemon_agent_test.go
index 7b1407b220e1586daf31d0c6b908a6e8a9fcd19a..3fceb34ecea885f349b7bbf2e8690faafea95a22 100644 (file)
--- a/daemon_agent_test.go
+++ b/daemon_agent_test.go
@@ -3,6 +3,7 @@ package main
 import (
        "context"
        "io"
 import (
        "context"
        "io"

+       "net/http"

        "os"
        "path/filepath"
        "strings"
        "os"
        "path/filepath"
        "strings"


@@ -28,6 +29,15 @@ func TestDaemonOpensAndRunsTheRootAndTwoChildren(t *testing.T) {
        require.Len(t, agents, 3)
        require.Equal(t, []string{"Garden", "Library"}, agents.ChildNames())
 
        require.Len(t, agents, 3)
        require.Equal(t, []string{"Garden", "Library"}, agents.ChildNames())
 

+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               text := ""
+               if call.body["prompt_cache_key"] == agents[0].agent.CacheKey() {
+                       text = "The decision is recorded."
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
+       }
+

        for _, agent := range agents {
                at := clock.Now().Add(time.Minute).Format(scheduleAtLayout)
                _, err = agent.schedules.Set("Record the decision for "+agent.agent.config.Name, &at, nil)
        for _, agent := range agents {
                at := clock.Now().Add(time.Minute).Format(scheduleAtLayout)
                _, err = agent.schedules.Set("Record the decision for "+agent.agent.config.Name, &at, nil)


@@ -56,7 +66,13 @@ func TestDaemonOpensAndRunsTheRootAndTwoChildren(t *testing.T) {
 
        require.Eventually(t, func() bool {
                for _, agent := range agents {
 
        require.Eventually(t, func() bool {
                for _, agent := range agents {

-                       if agent.memory.Len() != 2 {
+                       count := 1
+                       if agent == agents[0] {
+                               count = 2
+                       }
+
+                       usage, err := os.ReadFile(filepath.Join(agent.agent.dir, "usage.jsonl"))
+                       if err != nil || !strings.HasSuffix(string(usage), "\n") || agent.memory.Len() != count {

                                return false
                        }
                }
                                return false
                        }
                }

diff --git a/daemon_message.go b/daemon_message.go
new file mode 100644 (file)
index 0000000..db15db8
--- /dev/null
+++ b/daemon_message.go
@@ -0,0 +1,255 @@
+package main
+
+import (
+       "encoding/json"
+       "errors"
+       "fmt"
+       "os"
+       "path/filepath"
+       "slices"
+       "strings"
+       "sync"
+)
+
+// 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.
+func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
+       m := &DaemonMessages{
+               path:  filepath.Join(loops[0].agent.dir, "messages.json"),
+               loops: make(map[string]*TurnLoop),
+       }
+       for _, loop := range loops {
+               m.loops[loop.agent.config.Name] = loop
+               loop.messages = m
+               if loop.agent.config.Parent == "" {
+                       loop.logger = loop.logger.With("name", loop.agent.config.Name)
+               }
+       }
+
+       data, err := os.ReadFile(m.path)
+       if err != nil && !errors.Is(err, os.ErrNotExist) {
+               return nil, fmt.Errorf("read agent messages: %w", err)
+       }
+
+       if err == nil {
+               err = json.Unmarshal(data, &m.state)
+               if err != nil {
+                       return nil, fmt.Errorf("decode agent messages: %w", err)
+               }
+       }
+
+       if m.state.Next < 0 {
+               return nil, errors.New("invalid agent message sequence")
+       }
+
+       previous := 0
+       for _, message := range m.state.Pending {
+               source := m.loops[message.Source]
+               target := m.loops[message.Target]
+               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) {
+                       return nil, errors.New("queued agents are not parent and child")
+               }
+
+               previous = message.ID
+       }
+
+       for i := range m.state.Pending {
+               err = m.deliver(&m.state.Pending[i])
+               if err != nil {
+                       return nil, fmt.Errorf("restore agent message: %w", err)
+               }
+       }
+
+       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.
+type DaemonMessages struct {
+       mutex   sync.Mutex
+       path    string
+       loops   map[string]*TurnLoop
+       state   DaemonMessageFile
+       failure 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 {
+       loop := m.loops[message.Target]
+       logged, err := loop.memory.AppendOnce(message.Kind, message.Text, message.LogID, func(id int) error {
+               message.LogID = &id
+               return m.save()
+       })
+       if err != nil {
+               return fmt.Errorf("record agent message: %w", err)
+       }
+
+       status := ""
+       if message.Kind == "work" {
+               status = subagentFinished
+       }
+
+       loop.notify(Event{
+               Kind:   message.Kind,
+               ID:     logged.ID,
+               Text:   message.Text,
+               Source: message.Source,
+               Status: status,
+       })
+       loop.Wake(logged)
+       loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+
+       return nil
+}
+
+// save replaces the queue file under the mutex, or before the loops start.
+func (m *DaemonMessages) save() error {
+       data, err := json.Marshal(m.state)
+       if err != nil {
+               return fmt.Errorf("encode agent messages: %w", err)
+       }
+
+       err = writeFileAtomic(m.path, append(data, '\n'))
+       if err != nil {
+               return fmt.Errorf("save agent messages: %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 {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       source := m.loops[parent]
+       target := m.loops[child]
+       if source == nil || target == nil || target.agent.config.Parent != parent {
+               return errors.New("the recipient is not a child of the sender")
+       }
+
+       if m.failure != nil {
+               return fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       m.add(parent, child, "user", text)
+
+       err := m.save()
+       if err == nil {
+               err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
+       }
+
+       if err != nil {
+               return fmt.Errorf("send agent message: %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++
+       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
+// when a daemon stop interrupts a model request.
+func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       ids := make(map[int]bool, len(active))
+       for _, message := range active {
+               ids[message.ID] = true
+       }
+
+       name := loop.agent.config.Name
+       before := len(m.state.Pending)
+       m.state.Pending = slices.DeleteFunc(m.state.Pending, func(message DaemonMessage) bool {
+               return message.Target == name && message.LogID != nil && ids[*message.LogID]
+       })
+
+       parent := loop.agent.config.Parent
+       report := parent != "" && strings.TrimSpace(reply) != ""
+       if report {
+               m.add(name, parent, "work", "["+name+"] "+reply)
+       }
+
+       if !report && len(m.state.Pending) == before {
+               return nil
+       }
+
+       err := m.save()
+       if err == nil && report {
+               err = m.deliver(&m.state.Pending[len(m.state.Pending)-1])
+       }
+
+       if err != nil {
+               return fmt.Errorf("finish agent messages: %w", m.failed(err))
+       }
+
+       return nil
+}
+
+// DaemonMessageFile holds the sequence and the unacknowledged deliveries.
+type DaemonMessageFile struct {
+       Next    int             `json:"next"`
+       Pending []DaemonMessage `json:"pending"`
+}
+
+// DaemonMessage is one queued delivery. LogID is set before the log append. A
+// nil LogID means that delivery has not started.
+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"`
+}

diff --git a/daemon_message_test.go b/daemon_message_test.go
new file mode 100644 (file)
index 0000000..f9c25f1
--- /dev/null
+++ b/daemon_message_test.go
@@ -0,0 +1,625 @@
+package main
+
+import (
+       "context"
+       "encoding/json"
+       "errors"
+       "net/http"
+       "os"
+       "path/filepath"
+       "sync"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
+       server := newModelFake(t)
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               text := "I will ask the user."
+               if call.body["model"] == "gpt-test-garden" {
+                       text = "Approve the gate size."
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
+       }
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       test.Start(t, 0, 1, 2)
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
+       test.events[0].WaitFor(t, "idle", 1)
+       test.events[1].WaitFor(t, "idle", 1)
+
+       require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
+       require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
+       require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
+       require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
+       require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
+       require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
+       require.Equal(t, 2, server.Calls())
+       require.Zero(t, test.agents[2].memory.Len())
+       require.Empty(t, test.events[2].Kinds())
+
+       first, err := test.agents[1].memory.Message(0)
+       require.NoError(t, err)
+       require.Equal(t, "user", first.Kind)
+
+       report, err := test.agents[0].memory.Message(0)
+       require.NoError(t, err)
+       require.Equal(t, "work", report.Kind)
+       require.Equal(t, "[Garden] Approve the gate size.", report.Text)
+       require.Contains(t, test.logs.String(), "turn started name=Garden")
+       require.Contains(t, test.logs.String(), "turn done name=Garden")
+       require.NotContains(t, test.logs.String(), "gate size")
+       require.Empty(t, test.Queue(t).Pending)
+
+       info, err := os.Stat(test.messages.path)
+       require.NoError(t, err)
+       require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+}
+
+func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
+       server := newModelFake(t)
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               text := "I will ask the user."
+               if call.body["model"] == "gpt-test-garden" {
+                       text = "Approve the gate size."
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
+       }
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       telegram := newTelegramFake(t)
+       bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
+       test.loops[0].notify = func(event Event) {
+               test.events[0].Add(event)
+               bot.Notify(event)
+       }
+
+       ctx, cancel := context.WithCancel(t.Context())
+       defer cancel()
+
+       done := make(chan struct{})
+       go func() {
+               defer close(done)
+
+               bot.send(ctx)
+       }()
+
+       t.Cleanup(func() {
+               cancel()
+               <-done
+       })
+
+       test.Start(t, 0, 1)
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
+       test.events[0].WaitFor(t, "idle", 1)
+       require.Eventually(t, func() bool {
+               return len(telegram.Sent()) >= 2
+       }, 10*time.Second, time.Millisecond)
+       require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
+       require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
+       require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
+}
+
+func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
+       server := newModelFake(t)
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
+
+       failure := errors.New("fake queued message failure")
+       test.loops[1].fail(failure)
+
+       err := test.loops[1].Run(t.Context())
+       require.ErrorIs(t, err, failure)
+       require.ErrorContains(t, err, "check turn loop failure")
+       require.Zero(t, server.Calls())
+}
+
+func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
+       testCases := []struct {
+               name string
+               text string
+       }{
+               {
+                       name: "empty final reply",
+                       text: "",
+               },
+               {
+                       name: "whitespace final reply",
+                       text: " \n\t ",
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       server := newModelFake(t)
+                       server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+                               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
+                       }
+                       test := newDaemonMessageTest(t, t.TempDir(), server)
+                       test.Start(t, 0, 1, 2)
+                       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
+                       test.events[1].WaitFor(t, "idle", 1)
+                       require.Equal(t, 1, server.Calls())
+                       require.Zero(t, test.agents[0].memory.Len())
+                       require.Empty(t, test.events[0].Kinds())
+                       require.False(t, test.loops[0].waiting())
+                       require.Empty(t, test.events[2].Kinds())
+                       require.Empty(t, test.Queue(t).Pending)
+               })
+       }
+}
+
+func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
+       server := newModelFake(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               if call.number == 0 {
+                       close(arrived)
+                       <-release
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+       }
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       test.Start(t, 0, 1)
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
+       <-arrived
+       require.True(t, test.loops[1].Busy())
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
+       require.Equal(t, 1, server.Calls())
+       require.Len(t, test.Queue(t).Pending, 2)
+       close(release)
+       test.events[1].WaitFor(t, "idle", 2)
+       require.Equal(t, 2, server.Calls())
+
+       texts := turnInputTexts(t, server.Call(1))
+       require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
+       require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestBusyParentQueuesItsChildReport(t *testing.T) {
+       server := newModelFake(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               text := "The plan is ready."
+               if call.number == 0 {
+                       close(arrived)
+                       <-release
+
+                       text = "I will continue."
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
+       }
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       test.Start(t, 0, 1)
+
+       _, err := test.loops[0].Submit("terminal", "Start the work.")
+       require.NoError(t, err)
+       <-arrived
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
+       test.events[1].WaitFor(t, "idle", 1)
+       require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
+       require.True(t, test.loops[0].Busy())
+       require.Equal(t, 2, server.Calls())
+       close(release)
+       test.events[0].WaitFor(t, "idle", 2)
+       require.Equal(t, 3, server.Calls())
+
+       texts := turnInputTexts(t, server.Call(2))
+       require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
+       require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
+       server := newModelFake(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               if call.number == 0 {
+                       close(arrived)
+                       <-release
+                       modelWriteEvents(w, modelCompletedEvent([]any{
+                               modelMessageItem("I will check the plan."),
+                               modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
+                       }, 100, 0, 5))
+
+                       return
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
+       }
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       test.Start(t, 1)
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
+       <-arrived
+       require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
+       close(release)
+       test.events[1].WaitFor(t, "idle", 1)
+       require.Equal(t, 2, server.Calls())
+       require.Equal(t, 1, test.events[1].Count("busy"))
+       require.Equal(t, 1, test.events[0].Count("work"))
+       require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
+       require.NotContains(t, test.events[0].Last("work"), "I will check")
+
+       texts := turnInputTexts(t, server.Call(1))
+       require.Equal(t, "Also check the date.", texts[len(texts)-1])
+
+       queue := test.Queue(t)
+       require.Len(t, queue.Pending, 1)
+       require.Equal(t, "work", queue.Pending[0].Kind)
+}
+
+func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+       }
+       first := newDaemonMessageTest(t, dir, server)
+       require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
+       require.Equal(t, 1, first.agents[1].memory.Len())
+       first.Close(t)
+
+       second := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, second.agents[1].memory.Len())
+       require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
+       second.Start(t, 0, 1)
+       second.events[1].WaitFor(t, "idle", 1)
+       require.Equal(t, 1, server.Calls())
+       require.Empty(t, second.Queue(t).Pending)
+       second.Close(t)
+
+       third := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, third.agents[1].memory.Len())
+       require.False(t, third.loops[1].waiting())
+       require.Empty(t, third.events[1].Kinds())
+}
+
+func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       first := newDaemonMessageTest(t, dir, server)
+       first.Start(t, 1)
+       require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
+       first.events[1].WaitFor(t, "idle", 1)
+       require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
+       require.Len(t, first.Queue(t).Pending, 1)
+       first.Close(t)
+
+       second := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, second.agents[0].memory.Len())
+       require.False(t, second.loops[1].waiting())
+       second.Start(t, 0, 1)
+       second.events[0].WaitFor(t, "idle", 1)
+       require.Equal(t, 2, server.Calls())
+       require.Equal(t, 2, second.agents[0].memory.Len())
+       require.Empty(t, second.Queue(t).Pending)
+       second.Close(t)
+
+       third := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 2, third.agents[0].memory.Len())
+       require.False(t, third.loops[0].waiting())
+       require.Empty(t, third.events[0].Kinds())
+}
+
+func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               if call.number == 0 {
+                       close(arrived)
+                       <-release
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+       }
+       first := newDaemonMessageTest(t, dir, server)
+       first.Start(t, 1)
+       require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
+       <-arrived
+       first.Close(t)
+       close(release)
+
+       second := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, second.agents[1].memory.Len())
+       second.Start(t, 1)
+       second.events[1].WaitFor(t, "idle", 1)
+       require.Equal(t, 2, server.Calls())
+       require.Equal(t, 1, second.agents[1].memory.Len())
+       require.Empty(t, second.Queue(t).Pending)
+}
+
+func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+               if call.number == 1 {
+                       close(arrived)
+                       <-release
+               }
+
+               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
+       }
+       first := newDaemonMessageTest(t, dir, server)
+       first.Start(t, 0, 1)
+       require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
+       <-arrived
+       first.Close(t)
+       close(release)
+
+       second := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, second.agents[0].memory.Len())
+       require.False(t, second.loops[1].waiting())
+       second.Start(t, 0, 1)
+       second.events[0].WaitFor(t, "idle", 1)
+       require.Equal(t, 3, server.Calls())
+       require.Equal(t, 2, second.agents[0].memory.Len())
+       require.Empty(t, second.Queue(t).Pending)
+
+       texts := turnInputTexts(t, server.Call(2))
+       require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
+}
+
+func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
+       server := newModelFake(t)
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       require.NoError(t, os.Mkdir(test.messages.path, 0o700))
+       require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
+       require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
+
+       for _, loop := range test.loops {
+               require.ErrorContains(t, loop.failed(), "save agent messages")
+               require.Zero(t, loop.memory.Len())
+       }
+}
+
+func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
+       server := newModelFake(t)
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       entered := make(chan struct{})
+       release := make(chan struct{})
+
+       var once sync.Once
+
+       defer once.Do(func() {
+               close(release)
+       })
+
+       snapshot := make(chan error, 1)
+       go func() {
+               snapshot <- test.messages.Snapshot(func() error {
+                       return daemonSnapshot(t.Context(), test.loops, func() error {
+                               close(entered)
+                               <-release
+
+                               return nil
+                       })
+               })
+       }()
+
+       <-entered
+
+       sent := make(chan error, 1)
+       started := make(chan struct{})
+       go func() {
+               close(started)
+
+               sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
+       }()
+
+       <-started
+       require.Never(t, func() bool {
+               return len(sent) > 0
+       }, 50*time.Millisecond, time.Millisecond)
+       once.Do(func() {
+               close(release)
+       })
+       require.NoError(t, <-snapshot)
+       require.NoError(t, <-sent)
+       require.Equal(t, 1, test.agents[1].memory.Len())
+       require.Len(t, test.Queue(t).Pending, 1)
+}
+
+func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
+       server := newModelFake(t)
+
+       test := newDaemonMessageTest(t, t.TempDir(), server)
+       testCases := []struct {
+               name   string
+               source string
+               target string
+       }{
+               {
+                       name:   "no sibling messages",
+                       source: "Garden",
+                       target: "Library",
+               },
+               {
+                       name:   "no arbitrary child reports",
+                       source: "Garden",
+                       target: "OneChat",
+               },
+               {
+                       name:   "no self messages",
+                       source: "OneChat",
+                       target: "OneChat",
+               },
+               {
+                       name:   "unknown sender",
+                       source: "Other",
+                       target: "Garden",
+               },
+               {
+                       name:   "unknown recipient",
+                       source: "OneChat",
+                       target: "Other",
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
+               })
+       }
+
+       for _, agent := range test.agents {
+               require.Zero(t, agent.memory.Len())
+       }
+}
+
+func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
+       server := newModelFake(t)
+       testCases := []struct {
+               name string
+               file string
+       }{
+               {
+                       name: "broken JSON",
+                       file: "{",
+               },
+               {
+                       name: "unknown sender",
+                       file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
+               },
+               {
+                       name: "sibling delivery",
+                       file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
+               },
+               {
+                       name: "invalid sequence",
+                       file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       dir := t.TempDir()
+                       test := newDaemonMessageTest(t, dir, server)
+                       path := test.messages.path
+                       test.Close(t)
+                       require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
+
+                       agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
+                       require.NoError(t, err)
+
+                       defer func() {
+                               require.NoError(t, agents.Close())
+                       }()
+
+                       var loops []*TurnLoop
+
+                       for _, agent := range agents {
+                               loop, err := agent.Loop(nil, func(Event) {
+                               })
+                               require.NoError(t, err)
+
+                               loops = append(loops, loop)
+                       }
+
+                       _, err = openDaemonMessages(loops)
+                       require.Error(t, err)
+
+                       for _, agent := range agents {
+                               require.Zero(t, agent.memory.Len())
+                       }
+               })
+       }
+}
+
+// DaemonMessageTest owns test agents and the loops selected by Start.
+type DaemonMessageTest struct {
+       agents   DaemonAgents
+       loops    []*TurnLoop
+       events   []*TurnEventRecorder
+       messages *DaemonMessages
+       logs     *SyncBuffer
+       cancel   context.CancelFunc
+       done     []chan error
+}
+
+func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) {
+       t.Helper()
+
+       ctx, cancel := context.WithCancel(t.Context())
+       r.cancel = cancel
+       for _, index := range indices {
+               done := make(chan error, 1)
+               r.done = append(r.done, done)
+               go func() {
+                       done <- r.loops[index].Run(ctx)
+               }()
+       }
+}
+
+func (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile {
+       t.Helper()
+
+       data, err := os.ReadFile(r.messages.path)
+       require.NoError(t, err)
+
+       var file DaemonMessageFile
+
+       require.NoError(t, json.Unmarshal(data, &file))
+
+       return file
+}
+
+func (r *DaemonMessageTest) Close(t *testing.T) {
+       t.Helper()
+
+       if r.cancel != nil {
+               r.cancel()
+
+               for _, done := range r.done {
+                       require.ErrorIs(t, <-done, context.Canceled)
+               }
+
+               r.cancel = nil
+               r.done = nil
+       }
+
+       require.NoError(t, r.agents.Close())
+
+       r.agents = nil
+}
+
+func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
+       t.Helper()
+
+       _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
+       if errors.Is(err, os.ErrNotExist) {
+               daemonChildConfig(t, dir, "Garden")
+               daemonChildConfig(t, dir, "Library")
+       } else {
+               require.NoError(t, err)
+       }
+
+       test := &DaemonMessageTest{
+               logs: &SyncBuffer{},
+       }
+       test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
+       require.NoError(t, err)
+       t.Cleanup(func() {
+               test.Close(t)
+       })
+
+       for _, agent := range test.agents {
+               events := &TurnEventRecorder{}
+               loop, err := agent.Loop(nil, events.Add)
+               require.NoError(t, err)
+
+               test.events = append(test.events, events)
+               test.loops = append(test.loops, loop)
+       }
+
+       test.messages, err = openDaemonMessages(test.loops)
+       require.NoError(t, err)
+
+       return test
+}

diff --git a/memory_delivery_test.go b/memory_delivery_test.go
index 6f5e20c68f9202e0dfd343844fccb1b5839debcf..05b74bcfccc8e9db166f5bc3ca585b73f3cff04e 100644 (file)
--- a/memory_delivery_test.go
+++ b/memory_delivery_test.go
@@ -32,7 +32,7 @@ func TestQueuedAppendResumesWithoutADuplicate(t *testing.T) {
        }()
 
        second, err := memory.AppendOnce("user", first.Text, saved, func(int) error {
        }()
 
        second, err := memory.AppendOnce("user", first.Text, saved, func(int) error {

-               t.Fatal("a recorded message must not be prepared again")
+               require.FailNow(t, "a recorded message must not be prepared again")

 
                return nil
        })
 
                return nil
        })


@@ -106,7 +106,7 @@ func TestQueuedAppendRefusesAnInvalidReceipt(t *testing.T) {
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        _, err := memory.AppendOnce("user", "Other text.", &tc.id, func(int) error {
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        _, err := memory.AppendOnce("user", "Other text.", &tc.id, func(int) error {

-                               t.Fatal("an invalid receipt must not be prepared")
+                               require.FailNow(t, "an invalid receipt must not be prepared")

 
                                return nil
                        })
 
                                return nil
                        })

diff --git a/turn.go b/turn.go
index 2e5e50eef577cd3da3656dbaaa3122569ed112da..e559950675925ccf3fb51d470c7ce8a2af2bb803 100644 (file)
--- a/turn.go
+++ b/turn.go
@@ -88,18 +88,22 @@ type Event struct {
 // turn reads the view of the turn before from the prompt cache. User messages
 // that come during a turn join the next request of its chain after the tool
 // results, and so do the reports of the subagents. The subagents run until the
 // turn reads the view of the turn before from the prompt cache. User messages
 // that come during a turn join the next request of its chain after the tool
 // results, and so do the reports of the subagents. The subagents run until the

-// loop ends.
+// loop ends. Parent and child messages stay queued until the turn finishes.
+// active belongs to the running turn. The mutex protects pending, cancel, and
+// failure.

 type TurnLoop struct {
 type TurnLoop struct {

-       agent  *Agent
-       memory *Memory
-       client ChainClient
-       tools  *Toolbox
-       runs   *SubagentRuns
-       layout *ViewLayout
-       clock  func() time.Time
-       logger *slog.Logger
-       notify func(Event)
-       wake   chan struct{}
+       agent    *Agent
+       memory   *Memory
+       client   ChainClient
+       tools    *Toolbox
+       runs     *SubagentRuns
+       layout   *ViewLayout
+       clock    func() time.Time
+       logger   *slog.Logger
+       notify   func(Event)
+       wake     chan struct{}
+       messages *DaemonMessages
+       active   []Message

 
        mutex   sync.Mutex
        pending []Message
 
        mutex   sync.Mutex
        pending []Message


@@ -236,10 +240,15 @@ func (l *TurnLoop) Run(ctx context.Context) error {
        }
 
        for {
        }
 
        for {

-               for l.waiting() {
-                       err := l.turn(ctx)
+               for ctx.Err() == nil && l.waiting() {
+                       err := l.failed()

                        if err != nil {
                        if err != nil {

-                               return err
+                               return fmt.Errorf("check turn loop failure: %w", err)
+                       }
+
+                       err = l.turn(ctx)
+                       if err != nil {
+                               return fmt.Errorf("run turn: %w", err)

                        }
                }
 
                        }
                }
 


@@ -251,7 +260,7 @@ func (l *TurnLoop) Run(ctx context.Context) error {
 
                err := l.failed()
                if err != nil {
 
                err := l.failed()
                if err != nil {

-                       return err
+                       return fmt.Errorf("check turn loop failure: %w", err)

                }
        }
 }
                }
        }
 }


@@ -315,8 +324,11 @@ func (l *TurnLoop) failed() error {
 // when no client is connected, but an interrupt does not. The messages of the
 // turn stay in the log and the view, and the next turn starts only with a new
 // message, so a lasting failure such as a needed login does not repeat. Only a
 // when no client is connected, but an interrupt does not. The messages of the
 // turn stay in the log and the view, and the next turn starts only with a new
 // message, so a lasting failure such as a needed login does not repeat. Only a

-// failed log write is an error of turn.
+// failed log or queue write is an error of turn. A daemon stop leaves taken
+// parent and child messages queued for restart. A user interrupt completes them
+// without a report.

 func (l *TurnLoop) turn(ctx context.Context) error {
 func (l *TurnLoop) turn(ctx context.Context) error {

+       runCtx := ctx

        ctx, cancel := context.WithCancel(ctx)
        defer cancel()
 
        ctx, cancel := context.WithCancel(ctx)
        defer cancel()
 


@@ -336,7 +348,19 @@ func (l *TurnLoop) turn(ctx context.Context) error {
                Kind: "busy",
        })
 
                Kind: "busy",
        })
 

-       err := l.chain(ctx)
+       if l.messages != nil {
+               l.logger.Info("turn started")
+       }
+
+       reply, err := l.chain(ctx)
+       if l.messages != nil && !errors.Is(err, errChainLogWrite) && (err == nil || runCtx.Err() == nil) {
+               finishErr := l.messages.Finish(l, l.active, reply)
+               if finishErr != nil {
+                       err = errors.Join(err, fmt.Errorf("%w: finish agent messages: %w", errChainLogWrite, finishErr))
+               }
+       }
+
+       l.active = nil

        if err != nil && !errors.Is(err, errChainLogWrite) {
                if ctx.Err() == nil {
                        l.logger.Error("turn failed", "err", err)
        if err != nil && !errors.Is(err, errChainLogWrite) {
                if ctx.Err() == nil {
                        l.logger.Error("turn failed", "err", err)


@@ -350,6 +374,10 @@ func (l *TurnLoop) turn(ctx context.Context) error {
                err = nil
        }
 
                err = nil
        }
 

+       if l.messages != nil {
+               l.logger.Info("turn done")
+       }
+

        l.notify(Event{
                Kind: "idle",
        })
        l.notify(Event{
                Kind: "idle",
        })


@@ -358,20 +386,20 @@ func (l *TurnLoop) turn(ctx context.Context) error {
 }
 
 // chain runs the request chain of a turn until the final answer.
 }
 
 // chain runs the request chain of a turn until the final answer.

-func (l *TurnLoop) chain(ctx context.Context) error {
+func (l *TurnLoop) chain(ctx context.Context) (string, error) {

        messages := l.take()
        if len(messages) == 0 {
        messages := l.take()
        if len(messages) == 0 {

-               return nil
+               return "", nil

        }
 
        err := l.memory.Wait(ctx)
        if err != nil {
        }
 
        err := l.memory.Wait(ctx)
        if err != nil {

-               return err
+               return "", fmt.Errorf("wait for turn memory: %w", err)

        }
 
        instructions, err := l.agent.Prompt()
        if err != nil {
        }
 
        instructions, err := l.agent.Prompt()
        if err != nil {

-               return err
+               return "", fmt.Errorf("read turn prompt: %w", err)

        }
 
        skip := make(map[int]bool, len(messages))
        }
 
        skip := make(map[int]bool, len(messages))


@@ -381,7 +409,7 @@ func (l *TurnLoop) chain(ctx context.Context) error {
 
        view, err := l.layout.Split(l.memory.ViewLines(skip))
        if err != nil {
 
        view, err := l.layout.Split(l.memory.ViewLines(skip))
        if err != nil {

-               return err
+               return "", fmt.Errorf("split turn view: %w", err)

        }
 
        chain := &Chain{
        }
 
        chain := &Chain{


@@ -396,9 +424,12 @@ func (l *TurnLoop) chain(ctx context.Context) error {
                take: l.take,
        }
 
                take: l.take,
        }
 

-       _, err = chain.Run(ctx, turnInput(view, l.state(), messages))
+       reply, err := chain.Run(ctx, turnInput(view, l.state(), messages))
+       if err != nil {
+               return "", fmt.Errorf("run turn chain: %w", err)
+       }

 
 

-       return err
+       return reply, nil

 }
 
 // take returns the user messages that wait for a request, and removes them.
 }
 
 // take returns the user messages that wait for a request, and removes them.


@@ -408,6 +439,9 @@ func (l *TurnLoop) take() []Message {
 
        messages := l.pending
        l.pending = nil
 
        messages := l.pending
        l.pending = nil

+       if l.messages != nil {
+               l.active = append(l.active, messages...)
+       }

 
        return messages
 }
 
        return messages
 }