--- /dev/null
+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"`
+}
--- /dev/null
+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
+}
projects / onechat / commitdiff