)
// DaemonMessages owns the durable queue between the root and its children and
-// the routed inputs and root notifications. 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.
+// the routed inputs and notifications for permanent agents. 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
// DaemonMessage is one queued delivery. LogID is set before the log append. A
// nil LogID means that delivery has not started. WhatsAppBatch marks a routed
// external batch, whose event source is WhatsApp rather than its queue sender.
-// Notification marks root external data held outside memory until a turn
-// starts.
+// Notification marks external data held outside memory until a turn starts.
+// TargetID binds a notification to its permanent agent across name changes. An
+// empty TargetID identifies a root notification.
type DaemonMessage struct {
ID int `json:"id"`
Source string `json:"source"`
LogID *int `json:"log_id,omitempty"`
WhatsAppBatch int `json:"whatsapp_batch,omitempty"`
Notification bool `json:"notification,omitempty"`
+ TargetID string `json:"target_id,omitempty"`
}
// openDaemonMessages connects the root and its children. It restores queued
return nil, errors.New("too many notification receipts")
}
- receipts := make(map[int]bool)
+ receipts := make(map[int]NotifyReceipt)
keys := make(map[[2]string]bool)
for _, receipt := range m.state.Notifications {
key := [2]string{receipt.Source, receipt.ID}
digest, err := hex.DecodeString(receipt.Digest)
- if err != nil || len(digest) != 32 || receipt.Number <= 0 || receipt.Number > m.state.Next || receipts[receipt.Number] || keys[key] || !notifyIdentifier(receipt.Source) || !notifyIdentifier(receipt.ID) {
+ if err != nil || len(digest) != 32 || receipt.Number <= 0 || receipt.Number > m.state.Next || receipts[receipt.Number].Number != 0 || keys[key] || !notifyIdentifier(receipt.Source) || !notifyIdentifier(receipt.ID) {
return nil, errors.New("invalid notification receipt")
}
- receipts[receipt.Number] = true
+ if receipt.TargetID != "" {
+ id, err := hex.DecodeString(receipt.TargetID)
+ if err != nil || len(id) != agentIDBytes {
+ return nil, errors.New("invalid notification target identity")
+ }
+ }
+
+ receipts[receipt.Number] = receipt
keys[key] = true
}
message := &m.state.Pending[i]
source := m.loops[message.Source]
target := m.loops[message.Target]
+ if message.Notification {
+ target = loops[0]
+ if message.TargetID != "" {
+ target = m.notificationTarget(message.TargetID)
+ }
+
+ if target == nil {
+ return nil, errors.New("queued notification target is missing")
+ }
+
+ message.Target = target.agent.config.Name
+ }
+
if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
message.Target = message.Source
message.LogID = nil
switch {
case message.Notification:
- if source != loops[0] || target != source || message.Kind != "work" || message.WhatsAppBatch != 0 || !receipts[message.ID] {
+ receipt := receipts[message.ID]
+ if source != loops[0] || (target != source && target.agent.config.Parent != source.agent.config.Name) || message.Kind != "work" || message.WhatsAppBatch != 0 || receipt.Number == 0 || receipt.TargetID != message.TargetID {
return nil, errors.New("invalid queued notification")
}
case message.WhatsAppBatch != 0:
return nil, errors.New("queued agents are not parent and child")
}
+ if !message.Notification && message.TargetID != "" {
+ return nil, errors.New("agent message has a notification target identity")
+ }
+
previous = message.ID
}
return m, nil
}
+// notificationTarget resolves a stored identity without accepting names or
+// task-worker identities. The caller holds the queue lock or owns startup.
+func (m *DaemonMessages) notificationTarget(id string) *TurnLoop {
+ for _, loop := range m.ordered {
+ if strings.EqualFold(loop.agent.config.ID, id) {
+ return loop
+ }
+ }
+
+ return nil
+}
+
// 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 {
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]
+ target := message.Target == name
+ if message.Notification && message.TargetID != "" {
+ target = message.TargetID == loop.agent.config.ID
+ }
+
+ return target && message.LogID != nil && ids[*message.LogID]
})
parent := loop.agent.config.Parent
return nil
}
-// Notify accepts root external data without adding it to memory or an active
-// chain. A receipt is durable before success. Exact retries return the same
-// sequence even after completion. New keys fail when the receipt store is full.
+// Notify accepts external data for the root or a permanent child without adding
+// it to memory or an active chain. A receipt is durable before success. Exact
+// retries return the same sequence even after completion. Retry keys bind text
+// and permanent agent ID. New keys fail when the receipt store is full.
func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) {
m.mutex.Lock()
defer m.mutex.Unlock()
return Event{}, fmt.Errorf("validate external event: %w", err)
}
+ root := m.ordered[0]
+ target := root
+ if input.AgentID != "" {
+ target = m.notificationTarget(input.AgentID)
+ if target == nil {
+ return Event{}, errors.New("notification target identity is unknown")
+ }
+ }
+
digest := notifyDigest(input.Text)
for _, receipt := range m.state.Notifications {
if receipt.Source != input.Source || receipt.ID != input.ID {
continue
}
- if receipt.Digest != digest {
- return Event{}, errors.New("notification key conflicts with accepted text")
+ targetID := receipt.TargetID
+ if targetID == "" {
+ targetID = root.agent.config.ID
+ }
+
+ if receipt.Digest != digest || targetID != target.agent.config.ID {
+ return Event{}, errors.New("notification key conflicts with accepted target or text")
}
return Event{
return Event{}, errors.New("notification receipt store is full")
}
- root := m.ordered[0]
text, err := notifyText(*input, root.clock())
if err != nil {
return Event{}, fmt.Errorf("prepare external event: %w", err)
}
name := root.agent.config.Name
- m.add(name, name, "work", text)
+ m.add(name, target.agent.config.Name, "work", text)
m.state.Pending[len(m.state.Pending)-1].Notification = true
+ m.state.Pending[len(m.state.Pending)-1].TargetID = target.agent.config.ID
m.state.Notifications = append(m.state.Notifications, NotifyReceipt{
- Source: input.Source,
- ID: input.ID,
- Digest: digest,
- Number: m.state.Next,
+ TargetID: target.agent.config.ID,
+ Source: input.Source,
+ ID: input.ID,
+ Digest: digest,
+ Number: m.state.Next,
})
err = m.save()
}
select {
- case root.wake <- struct{}{}:
+ case target.wake <- struct{}{}:
default:
}
}, nil
}
-// Notifications delivers at most twenty queued events to a root turn. It stops
-// before 64 KiB of envelope text, except that the first event always fits. Call
-// only between chains, never from take. Events stay queued until Finish, so
-// daemon shutdown can repeat model calls and tool effects, but AppendOnce does
-// not create duplicate delivery log entries.
+// Notifications delivers at most twenty queued events to a recipient turn. It
+// stops before 64 KiB of envelope text, except that the first event always
+// fits. Call only between chains, never from take. Events stay queued until
+// Finish, so daemon shutdown can repeat model calls and tool effects, but
+// AppendOnce does not create duplicate delivery log entries.
func (m *DaemonMessages) Notifications(loop *TurnLoop) error {
m.mutex.Lock()
defer m.mutex.Unlock()
size := 0
for i := range m.state.Pending {
message := &m.state.Pending[i]
- if !message.Notification || message.Target != loop.agent.config.Name {
+ if !message.Notification {
+ continue
+ }
+
+ target := message.Target == loop.agent.config.Name
+ if message.TargetID != "" {
+ target = message.TargetID == loop.agent.config.ID
+ }
+
+ if !target {
continue
}
testCases := []struct {
name string
resume bool
+ child bool
}{
{
name: "keeps its reservation before a startup user message",
name: "keeps its reservation before a worker recovery note",
resume: true,
},
+ {
+ name: "keeps a child reservation before startup input",
+ child: true,
+ },
+ {
+ name: "keeps a child reservation before a worker recovery note",
+ child: true,
+ resume: true,
+ },
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
+ index := 0
+ if tc.child {
+ index = 1
+ }
+
directory := t.TempDir()
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
first := newDaemonMessageTest(t, directory, server)
input := &NotifyInput{
- Source: "check",
- ID: "reserved",
- Text: "A result with a prepared log ID.",
+ AgentID: first.agents[index].agent.config.ID,
+ Source: "check",
+ ID: "reserved",
+ Text: "A result with a prepared log ID.",
}
receipt, err := first.messages.Notify(input)
first.Close(t)
second := newDaemonMessageTest(t, directory, server)
- require.False(t, second.loops[0].waiting())
- require.Empty(t, second.events[0].Kinds())
+ require.False(t, second.loops[index].waiting())
+ require.Empty(t, second.events[index].Kinds())
if tc.resume {
- path := filepath.Join(second.agents[0].agent.dir, subagentWorkFileName)
+ path := filepath.Join(second.agents[index].agent.dir, subagentWorkFileName)
require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
- require.NoError(t, second.loops[0].runs.Resume(t.Context()))
+ require.NoError(t, second.loops[index].runs.Resume(t.Context()))
- note, err := second.agents[0].memory.Message(1)
+ note, err := second.agents[index].memory.Message(1)
require.NoError(t, err)
require.Equal(t, "note", note.Kind)
require.Contains(t, note.Text, "The daemon restarted, cutting BlueFox")
} else {
- submitted, err := second.loops[0].Submit(terminalSource, "A user message during startup.")
+ submitted, err := second.loops[index].Submit(terminalSource, "A user message during startup.")
require.NoError(t, err)
require.Equal(t, 1, submitted.ID)
}
- restored, err := second.agents[0].memory.Message(id)
+ restored, err := second.agents[index].memory.Message(id)
require.NoError(t, err)
require.Equal(t, "work", restored.Kind)
require.Equal(t, expected, restored.Text)
require.Len(t, second.Queue(t).Pending, 1)
- second.Start(t, 0)
- second.events[0].WaitFor(t, "idle", 1)
+ second.Start(t, index)
+ second.events[index].WaitFor(t, "idle", 1)
require.Empty(t, second.Queue(t).Pending)
require.Equal(t, 1, server.Calls())
- require.Equal(t, id, second.events[0].LastEvent("work").ID)
+ require.Equal(t, id, second.events[index].LastEvent("work").ID)
duplicate, err := second.messages.Notify(input)
require.NoError(t, err)
require.Empty(t, test.Queue(t).Pending)
}
+func TestNotificationSelectorsBindOnePermanentIdentity(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ input := &NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "targeted",
+ Text: "A result for one child.",
+ }
+
+ accepted, err := test.messages.Notify(input)
+ require.NoError(t, err)
+
+ queue := test.Queue(t)
+ require.Equal(t, test.agents[1].agent.config.ID, queue.Pending[0].TargetID)
+ require.Equal(t, queue.Pending[0].TargetID, queue.Notifications[0].TargetID)
+ require.Equal(t, "Garden", queue.Pending[0].Target)
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Zero(t, test.agents[1].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+
+ input.AgentID = strings.ToUpper(test.agents[1].agent.config.ID)
+ duplicate, err := test.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, accepted.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = test.agents[2].agent.config.ID
+ _, err = test.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+
+ input.AgentID = ""
+ _, err = test.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+ require.Equal(t, queue, test.Queue(t))
+}
+
+func TestNotificationRejectsUnknownWorkerAndMalformedIdentities(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.loops[0].runs.subagents.draw = func() string {
+ return "BlueFox"
+ }
+
+ worker, err := test.loops[0].runs.subagents.Create(t.Context())
+ require.NoError(t, err)
+
+ testCases := []NotifyInput{
+ {AgentID: worker.name},
+ {AgentID: "Missing"},
+ {AgentID: "garden"},
+ {AgentID: "../Garden"},
+ {AgentID: strings.Repeat("f", 2*agentIDBytes)},
+ {AgentID: "invalid"},
+ }
+
+ for _, tc := range testCases {
+ tc.Source = "check"
+ tc.ID = "1"
+ tc.Text = "Unchanged data."
+ receipt, err := test.messages.Notify(&tc)
+ require.Error(t, err)
+ require.Empty(t, receipt.Kind)
+ }
+
+ require.Empty(t, test.messages.state.Pending)
+ require.Empty(t, test.messages.state.Notifications)
+ require.Zero(t, server.Calls())
+}
+
+func TestChildNotificationWaitsForTheWholeActiveChain(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{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+
+ return
+ }
+
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.Start(t, 0, 1, 2)
+ require.NoError(t, test.messages.Tell("OneChat", "Garden", "Start the first chain."))
+ <-arrived
+
+ _, err := test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "next-turn",
+ Text: "Only the next child turn sees this.",
+ })
+ require.NoError(t, err)
+ require.Equal(t, 1, test.agents[1].memory.Len())
+ close(release)
+ test.events[1].WaitFor(t, "idle", 2)
+ require.Equal(t, 3, server.Calls())
+ require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), "Only the next child turn")
+ require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), "Only the next child turn")
+ require.Equal(t, notifySource, test.events[1].LastEvent("work").Source)
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationStableIdentitySurvivesOfflineNameChange(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ input := &NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "rename-proof",
+ Text: "A saved child result.",
+ }
+
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+
+ id := 0
+ first.messages.state.Pending[0].LogID = &id
+ require.NoError(t, first.messages.save())
+
+ envelope := first.Queue(t).Pending[0].Text
+ config := first.agents[1].agent.config
+ oldPath := first.agents[1].agent.dir
+ first.Close(t)
+
+ // Change only test fixtures to exercise recovery under the same
+ // identity.
+ newPath := filepath.Join(filepath.Dir(oldPath), "Orchard")
+ require.NoError(t, os.Rename(oldPath, newPath))
+
+ config.Name = "Orchard"
+ agentWriteTestConfig(t, newPath, config)
+
+ second := newDaemonMessageTest(t, directory, server)
+ child := second.messages.Child("OneChat", "Orchard")
+ require.NotNil(t, child)
+ require.False(t, child.waiting())
+
+ restored, err := child.memory.Message(id)
+ require.NoError(t, err)
+ require.Equal(t, envelope, restored.Text)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.NoError(t, second.messages.Notifications(child))
+ require.NoError(t, child.turn(t.Context()))
+ require.Empty(t, second.Queue(t).Pending)
+ require.Zero(t, second.agents[0].memory.Len())
+
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, directory, server)
+ duplicate, err = third.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Empty(t, third.Queue(t).Pending)
+ require.Equal(t, 1, server.Calls())
+}
+
+func TestMissingNotificationIdentityStopsStartupWithoutFallback(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, directory, server)
+ _, err := first.messages.Notify(&NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "missing",
+ Text: "Do not redirect this result.",
+ })
+ require.NoError(t, err)
+
+ path := first.messages.path
+ before, err := os.ReadFile(path)
+ require.NoError(t, err)
+
+ childPath := first.agents[1].agent.dir
+ first.Close(t)
+
+ require.NoError(t, os.RemoveAll(childPath))
+ // A new agent with the same name must not receive the old identity's
+ // work.
+ daemonChildConfig(t, directory, "Garden")
+
+ agents, err := openDaemonAgents(t.Context(), directory, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
+ require.NoError(t, err)
+
+ 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.ErrorContains(t, err, "notification target is missing")
+
+ after, err := os.ReadFile(path)
+ require.NoError(t, err)
+ require.Equal(t, before, after)
+ require.Zero(t, agents[0].memory.Len())
+ require.Zero(t, agents[1].memory.Len())
+ require.NoError(t, agents.Close())
+}
+
+func TestCompletedNotificationReceiptDoesNotRequireItsTargetAtStartup(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ _, err := first.messages.Notify(&NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "completed-child",
+ Text: "A completed result.",
+ })
+ require.NoError(t, err)
+ require.NoError(t, first.messages.Notifications(first.loops[1]))
+ require.NoError(t, first.loops[1].turn(t.Context()))
+ require.Empty(t, first.Queue(t).Pending)
+
+ childPath := first.agents[1].agent.dir
+ first.Close(t)
+ require.NoError(t, os.RemoveAll(childPath))
+
+ second := newDaemonMessageTest(t, directory, server)
+ require.Empty(t, second.Queue(t).Pending)
+ require.Len(t, second.Queue(t).Notifications, 1)
+ require.Zero(t, second.agents[0].memory.Len())
+}
+
+func TestLegacyNotificationReceiptsStillBindTheRoot(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, directory, server)
+ input := &NotifyInput{
+ Source: "check",
+ ID: "legacy",
+ Text: "A root result.",
+ }
+
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+
+ first.messages.state.Pending[0].TargetID = ""
+ first.messages.state.Notifications[0].TargetID = ""
+ require.NoError(t, first.messages.save())
+
+ before, err := os.ReadFile(first.messages.path)
+ require.NoError(t, err)
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ after, err := os.ReadFile(second.messages.path)
+ require.NoError(t, err)
+ require.Equal(t, before, after)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = second.agents[0].agent.config.ID
+ duplicate, err = second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = second.agents[1].agent.config.ID
+ _, err = second.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+}
+
+func TestChildNotificationRunsWhileTheRootIsBusy(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)
+
+ _, err := test.loops[0].Submit(terminalSource, "Keep the root busy.")
+ require.NoError(t, err)
+ <-arrived
+
+ _, err = test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "independent",
+ Text: "A result for the idle child.",
+ })
+ require.NoError(t, err)
+ test.events[1].WaitFor(t, "idle", 1)
+ require.True(t, test.loops[0].Busy())
+ require.Equal(t, 1, test.agents[0].memory.Len())
+ require.Empty(t, test.events[0].Last("work"))
+ close(release)
+ test.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, test.Queue(t).Pending)
+ require.Equal(t, 2, server.Calls())
+}
+
+func TestNotificationBatchesStayOrderedForEachChild(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.agents[1].memory.summarizer = newFakeSummarizer()
+ test.agents[2].memory.summarizer = newFakeSummarizer()
+
+ for number := range 21 {
+ for index := 1; index <= 2; index++ {
+ _, err := test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[index].agent.config.ID,
+ Source: "check",
+ ID: fmt.Sprintf("%d-%d", index, number),
+ Text: fmt.Sprintf("target-%d-result-%02d", index, number),
+ })
+ require.NoError(t, err)
+ }
+ }
+
+ test.Start(t, 0, 1, 2)
+ test.events[1].WaitFor(t, "idle", 2)
+ test.events[2].WaitFor(t, "idle", 2)
+ require.Equal(t, 4, server.Calls())
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Empty(t, test.Queue(t).Pending)
+
+ for index := 1; index <= 2; index++ {
+ for number := range 21 {
+ logged, err := test.agents[index].memory.Message(number)
+ require.NoError(t, err)
+ require.Contains(t, logged.Text, fmt.Sprintf("target-%d-result-%02d", index, number))
+ }
+ }
+}
+
func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
projects / onechat / commitdiff