From: Anderson A. Giacomolli Date: Wed, 11 Nov 2026 19:17:04 +0000 (-0300) Subject: Bind external events to permanent agent IDs X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/6c7859cfea123479aba046614ee81028651ddaf5?ds=inline;hp=9301f77a5709fe7daea3a71f8dd072c6cd7df4fe Bind external events to permanent agent IDs Notification intake binds each accepted event to a permanent agent ID. Only that agent sees the event at its next turn boundary. Global retry keys reject a different target or text, while root receipts without an identity retain their root binding. Pending recovery resolves identity before log reservations and does not redirect work to a replacement with the same name. Delivery, completion, backup, and import guards preserve that binding without changing saved envelopes or adding a rename operation. Issue: 44d8703 --- diff --git a/backup_test.go b/backup_test.go index 1638ed0..81fefd9 100644 --- a/backup_test.go +++ b/backup_test.go @@ -125,9 +125,10 @@ func TestBackupKeepsPendingNotificationsAndCompletedRetryReceipts(t *testing.T) require.NoError(t, test.loops[0].turn(t.Context())) _, err = test.messages.Notify(&NotifyInput{ - Source: "check", - ID: "pending", - Text: "A pending result.", + Source: "check", + ID: "pending", + AgentID: test.agents[1].agent.config.ID, + Text: "A pending result.", }) require.NoError(t, err) @@ -149,6 +150,7 @@ func TestBackupKeepsPendingNotificationsAndCompletedRetryReceipts(t *testing.T) recovered := newDaemonMessageTest(t, restored, server) require.Len(t, recovered.Queue(t).Pending, 1) + require.Equal(t, test.agents[1].agent.config.ID, recovered.Queue(t).Pending[0].TargetID) require.Len(t, recovered.Queue(t).Notifications, 2) duplicate, err := recovered.messages.Notify(&NotifyInput{ diff --git a/daemon_children_test.go b/daemon_children_test.go index 8349586..8fcae36 100644 --- a/daemon_children_test.go +++ b/daemon_children_test.go @@ -99,6 +99,37 @@ func TestDaemonToolsCreateAndTellAChildDuringATurn(t *testing.T) { }, 5*time.Second, time.Millisecond) } +func TestRuntimeChildAcceptsNotificationsByItsPermanentID(t *testing.T) { + test := newDaemonChildrenTest(t) + test.server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5)) + } + + _, err := test.runtime.Create("Garden", "Keep garden plans.") + require.NoError(t, err) + + child := test.runtime.messages.Child(test.root.agent.config.Name, "Garden") + require.NotNil(t, child) + + receipt, err := test.runtime.messages.Notify(&NotifyInput{ + AgentID: child.agent.config.ID, + Source: "check", + ID: "runtime-child", + Text: "A new result.", + }) + require.NoError(t, err) + require.Equal(t, "accepted", receipt.Status) + require.Eventually(t, func() bool { + return child.memory.Len() == 1 + }, 10*time.Second, time.Millisecond) + require.Zero(t, test.root.memory.Len()) + + logged, err := child.memory.Message(0) + require.NoError(t, err) + require.Equal(t, "work", logged.Kind) + require.Contains(t, logged.Text, "Untrusted external data") +} + func TestChildCreationRefusesUsedNames(t *testing.T) { test := newDaemonChildrenTest(t) box := test.root.tools diff --git a/daemon_message.go b/daemon_message.go index f37a88f..8e36538 100644 --- a/daemon_message.go +++ b/daemon_message.go @@ -14,10 +14,10 @@ import ( ) // 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 @@ -42,8 +42,9 @@ type DaemonMessageFile struct { // 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"` @@ -53,6 +54,7 @@ type DaemonMessage struct { 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 @@ -123,16 +125,23 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 } @@ -141,6 +150,19 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 @@ -153,7 +175,8 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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: @@ -164,6 +187,10 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 } @@ -201,6 +228,18 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 { @@ -496,7 +535,12 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) 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 @@ -521,9 +565,10 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) 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() @@ -541,14 +586,28 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { 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{ @@ -562,21 +621,22 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { 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() @@ -585,7 +645,7 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { } select { - case root.wake <- struct{}{}: + case target.wake <- struct{}{}: default: } @@ -598,11 +658,11 @@ func (m *DaemonMessages) Notify(input *NotifyInput) (Event, error) { }, 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() @@ -615,7 +675,16 @@ func (m *DaemonMessages) Notifications(loop *TurnLoop) error { 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 } diff --git a/daemon_message_test.go b/daemon_message_test.go index cfb41c6..b401eb7 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -304,6 +304,7 @@ func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) { testCases := []struct { name string resume bool + child bool }{ { name: "keeps its reservation before a startup user message", @@ -312,10 +313,24 @@ func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) { 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) { @@ -324,9 +339,10 @@ func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) { 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) @@ -340,35 +356,35 @@ func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) { 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) @@ -398,6 +414,372 @@ func TestNotificationBatchMakesProgressWithEscapedControlCharacters(t *testing.T 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) { diff --git a/notify.go b/notify.go index 1036312..2071ce6 100644 --- a/notify.go +++ b/notify.go @@ -3,6 +3,7 @@ package main import ( "context" "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -49,20 +50,24 @@ Child targets, receipt expiry, and processing status are not supported. ) // NotifyInput is caller-supplied data. Its source label proves no identity or -// authority. Source and ID together identify a retry of the same text. +// authority. Source and ID together identify a retry of the same text. AgentID +// selects a permanent agent. An empty AgentID selects the root. type NotifyInput struct { - Source string `json:"source"` - ID string `json:"id"` - Text string `json:"text"` + AgentID string `json:"agent_id,omitempty"` + Source string `json:"source"` + ID string `json:"id"` + Text string `json:"text"` } // NotifyReceipt keeps a retry key after its queued delivery finishes. Digest -// binds the original text without retaining another copy of that text. +// binds the original text without retaining another copy of that text. TargetID +// binds the permanent agent identity. An empty TargetID selects the root. type NotifyReceipt struct { - Source string `json:"source"` - ID string `json:"id"` - Digest string `json:"digest"` - Number int `json:"number"` + TargetID string `json:"target_id,omitempty"` + Source string `json:"source"` + ID string `json:"id"` + Digest string `json:"digest"` + Number int `json:"number"` } // notify submits a local event and prints only its acceptance receipt. The @@ -202,6 +207,13 @@ func notifyParse(args []string) (NotifyInput, error) { // notifyValidate checks both CLI and direct socket inputs without echoing data. func notifyValidate(input NotifyInput) error { + if input.AgentID != "" { + id, err := hex.DecodeString(input.AgentID) + if err != nil || len(id) != agentIDBytes { + return errors.New("notification agent ID is invalid") + } + } + if !notifyIdentifier(input.Source) || !notifyIdentifier(input.ID) { return errors.New("notification source and ID must be 1 to 128 ASCII letters, digits, dots, underscores, or hyphens") } diff --git a/whatsapp_import.go b/whatsapp_import.go index f293ebd..cb77563 100644 --- a/whatsapp_import.go +++ b/whatsapp_import.go @@ -357,7 +357,7 @@ func whatsappImportIdle(root string, agent *Agent) error { } for _, message := range queue.Pending { - if message.Target == agent.config.Name { + if (message.Notification && message.TargetID != "" && message.TargetID == agent.config.ID) || (message.TargetID == "" && message.Target == agent.config.Name) { return fmt.Errorf("import refused: agent %q has pending deliveries", agent.config.Name) } } diff --git a/whatsapp_import_test.go b/whatsapp_import_test.go index 382d4bf..eed22fd 100644 --- a/whatsapp_import_test.go +++ b/whatsapp_import_test.go @@ -705,24 +705,46 @@ func TestWhatsAppImportRefusesPendingDeliveriesAndKeepsRestartSafe(t *testing.T) } func TestWhatsAppImportBlocksPendingNotificationsButNotCompletedReceipts(t *testing.T) { - server := newModelFake(t) - test := newDaemonMessageTest(t, t.TempDir(), server) - _, err := test.messages.Notify(&NotifyInput{ - Source: "check", - ID: "1", - Text: "A result.", - }) - require.NoError(t, err) + testCases := []struct { + name string + index int + }{ + { + name: "blocks only pending root notifications", + }, + { + name: "blocks only pending child notifications", + index: 1, + }, + } - root := test.agents[0].agent - directory := root.dir - require.ErrorContains(t, whatsappImportIdle(directory, root), "pending deliveries") - require.NoError(t, whatsappImportIdle(directory, test.agents[1].agent)) - require.NoError(t, test.messages.Notifications(test.loops[0])) - require.NoError(t, test.loops[0].turn(t.Context())) - require.Len(t, test.Queue(t).Notifications, 1) - require.Empty(t, test.Queue(t).Pending) - require.NoError(t, whatsappImportIdle(directory, root)) + 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("")}, 100, 0, 5)) + } + + test := newDaemonMessageTest(t, t.TempDir(), server) + _, err := test.messages.Notify(&NotifyInput{ + AgentID: test.agents[tc.index].agent.config.ID, + Source: "check", + ID: "1", + Text: "A result.", + }) + require.NoError(t, err) + + target := test.agents[tc.index].agent + directory := test.agents[0].agent.dir + require.ErrorContains(t, whatsappImportIdle(directory, target), "pending deliveries") + require.NoError(t, whatsappImportIdle(directory, test.agents[tc.index^1].agent)) + require.NoError(t, test.messages.Notifications(test.loops[tc.index])) + require.NoError(t, test.loops[tc.index].turn(t.Context())) + require.Len(t, test.Queue(t).Notifications, 1) + require.Empty(t, test.Queue(t).Pending) + require.NoError(t, whatsappImportIdle(directory, target)) + }) + } } func TestWhatsAppImportPendingDeliveriesBlockOnlyTheirTarget(t *testing.T) {