From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 05:19:49 +0000 (-0300) Subject: Deliver messages between parent and child agents X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/2ad578a24f252706976cddb41bee6cc969f251cd?ds=sidebyside 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 473d1e5..b60f984 100644 --- 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 -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 diff --git a/daemon.go b/daemon.go index 3a54b1a..958d6f7 100644 --- 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) } + 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 { - return daemonSnapshot(ctx, loops, snapshot) + return messages.Snapshot(func() error { + return daemonSnapshot(ctx, loops, snapshot) + }) } var workers sync.WaitGroup diff --git a/daemon_agent_test.go b/daemon_agent_test.go index 7b1407b..3fceb34 100644 --- a/daemon_agent_test.go +++ b/daemon_agent_test.go @@ -3,6 +3,7 @@ package main import ( "context" "io" + "net/http" "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()) + 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) @@ -56,7 +66,13 @@ func TestDaemonOpensAndRunsTheRootAndTwoChildren(t *testing.T) { 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 } } diff --git a/daemon_message.go b/daemon_message.go new file mode 100644 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 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 6f5e20c..05b74bc 100644 --- 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 { - t.Fatal("a recorded message must not be prepared again") + require.FailNow(t, "a recorded message must not be prepared again") 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 { - t.Fatal("an invalid receipt must not be prepared") + require.FailNow(t, "an invalid receipt must not be prepared") return nil }) diff --git a/turn.go b/turn.go index 2e5e50e..e559950 100644 --- 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 -// 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 { - 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 @@ -236,10 +240,15 @@ func (l *TurnLoop) Run(ctx context.Context) error { } for { - for l.waiting() { - err := l.turn(ctx) + for ctx.Err() == nil && l.waiting() { + err := l.failed() 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 { - 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 -// 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 { + runCtx := ctx ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -336,7 +348,19 @@ func (l *TurnLoop) turn(ctx context.Context) error { 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) @@ -350,6 +374,10 @@ func (l *TurnLoop) turn(ctx context.Context) error { err = nil } + if l.messages != nil { + l.logger.Info("turn done") + } + 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. -func (l *TurnLoop) chain(ctx context.Context) error { +func (l *TurnLoop) chain(ctx context.Context) (string, error) { messages := l.take() if len(messages) == 0 { - return nil + return "", 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 { - return err + return "", fmt.Errorf("read turn prompt: %w", err) } 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 { - return err + return "", fmt.Errorf("split turn view: %w", err) } chain := &Chain{ @@ -396,9 +424,12 @@ func (l *TurnLoop) chain(ctx context.Context) error { 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. @@ -408,6 +439,9 @@ func (l *TurnLoop) take() []Message { messages := l.pending l.pending = nil + if l.messages != nil { + l.active = append(l.active, messages...) + } return messages }