+func TestNotificationWaitsForTheNextTurnAfterEveryActiveToolRequest(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)
+
+ _, err := test.loops[0].Submit(terminalSource, "Start a tool chain.")
+ require.NoError(t, err)
+ <-arrived
+
+ before := test.agents[0].memory.Len()
+ input := &NotifyInput{
+ Source: "test-check",
+ ID: "event-1",
+ Text: "Unique queued external data.",
+ }
+
+ receipt, err := test.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "accepted", receipt.Status)
+ require.Equal(t, before, test.agents[0].memory.Len())
+ require.True(t, test.loops[0].Busy())
+ close(release)
+ test.events[0].WaitFor(t, "idle", 2)
+ require.Equal(t, 3, server.Calls())
+ require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), input.Text)
+ require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), input.Text)
+ require.Equal(t, notifySource, test.events[0].LastEvent("work").Source)
+ require.Empty(t, test.Queue(t).Pending)
+ require.Len(t, test.Queue(t).Notifications, 1)
+ require.Zero(t, test.agents[1].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.NotContains(t, test.logs.String(), input.Source)
+ require.NotContains(t, test.logs.String(), input.Text)
+ require.NotContains(t, test.logs.String(), input.ID)
+}
+
+func TestNotificationRetriesSurviveRestartBeforeAndAfterDelivery(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ input := &NotifyInput{
+ Source: "check",
+ ID: "result-1",
+ Text: "A new result.",
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+ require.Zero(t, first.agents[0].memory.Len())
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Zero(t, second.agents[0].memory.Len())
+ second.Start(t, 0)
+ second.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, second.Queue(t).Pending)
+ 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.Equal(t, receipt.ID, duplicate.ID)
+ require.False(t, third.loops[0].waiting())
+ require.Equal(t, 1, server.Calls())
+
+ changed := *input
+ changed.Text = "A different result."
+ _, err = third.messages.Notify(&changed)
+ require.ErrorContains(t, err, "conflicts")
+ require.Len(t, third.Queue(t).Notifications, 1)
+}
+
+func TestNotificationAcceptanceIsSerializedForConcurrentRetries(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ input := &NotifyInput{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ }
+
+ results := make(chan Event, 20)
+ failures := make(chan error, 20)
+
+ var workers sync.WaitGroup
+
+ for range 20 {
+ workers.Add(1)
+
+ go func() {
+ defer workers.Done()
+
+ receipt, err := test.messages.Notify(input)
+ results <- receipt
+
+ failures <- err
+ }()
+ }
+
+ workers.Wait()
+ close(results)
+ close(failures)
+
+ accepted := 0
+ for receipt := range results {
+ require.Equal(t, 1, receipt.ID)
+
+ if receipt.Status == "accepted" {
+ accepted++
+ } else {
+ require.Equal(t, "duplicate", receipt.Status)
+ }
+ }
+
+ for err := range failures {
+ require.NoError(t, err)
+ }
+
+ require.Equal(t, 1, accepted)
+ require.Len(t, test.Queue(t).Pending, 1)
+ require.Len(t, test.Queue(t).Notifications, 1)
+}
+
+func TestNotificationStorageFailureReturnsNoAcceptedReceipt(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.messages.path = filepath.Join(t.TempDir(), "missing", "messages.json")
+ receipt, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ })
+ require.ErrorContains(t, err, "accept external event")
+ require.Empty(t, receipt.Kind)
+ require.False(t, test.loops[0].waiting())
+ require.Zero(t, test.agents[0].memory.Len())
+
+ _, err = test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "2",
+ Text: "Another result.",
+ })
+ require.ErrorContains(t, err, "queue stopped")
+}
+
+func TestNotificationFullStoreRejectsNewKeysWithoutRemovingReceipts(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+
+ for number := range notifyReceipts {
+ test.messages.state.Notifications = append(test.messages.state.Notifications, NotifyReceipt{
+ Source: "check",
+ ID: fmt.Sprint(number),
+ Digest: notifyDigest("data"),
+ Number: number + 1,
+ })
+ }
+
+ test.messages.state.Next = notifyReceipts
+ receipt, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "new-key",
+ Text: "data",
+ })
+ require.ErrorContains(t, err, "store is full")
+ require.Empty(t, receipt.Kind)
+ require.Len(t, test.messages.state.Notifications, notifyReceipts)
+ require.Empty(t, test.messages.state.Pending)
+
+ receipt, err = test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "0",
+ Text: "data",
+ })
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", receipt.Status)
+}
+
+func TestNotificationBatchesKeepAcceptanceOrder(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[0].memory.summarizer = newFakeSummarizer()
+ for number := range 21 {
+ _, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: fmt.Sprint(number),
+ Text: fmt.Sprintf("result-%02d", number),
+ })
+ require.NoError(t, err)
+ }
+
+ test.Start(t, 0)
+ test.events[0].WaitFor(t, "idle", 2)
+ require.Equal(t, 2, server.Calls())
+
+ first := turnInputTexts(t, server.Call(0))
+ require.Len(t, first, 22)
+
+ for number := range 20 {
+ require.Contains(t, first[number+2], fmt.Sprintf("result-%02d", number))
+ }
+
+ second := turnInputTexts(t, server.Call(1))
+ require.Contains(t, second[len(second)-1], "result-20")
+ require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationRecoversItsPreparedLogIDWithoutAnotherAppend(t *testing.T) {
+ for _, appended := range []bool{false, true} {
+ t.Run(fmt.Sprintf("log append completed %t", appended), func(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{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ })
+ require.NoError(t, err)
+
+ if appended {
+ require.NoError(t, first.messages.Notifications(first.loops[0]))
+ } else {
+ id := 0
+ first.messages.state.Pending[0].LogID = &id
+ require.NoError(t, first.messages.save())
+ }
+
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ second.Start(t, 0)
+ second.events[0].WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, 1, second.agents[0].memory.Len())
+ require.Empty(t, second.Queue(t).Pending)
+ })
+ }
+}
+
+func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) {
+ testCases := []struct {
+ name string
+ resume bool
+ }{
+ {
+ name: "keeps its reservation before a startup user message",
+ },
+ {
+ name: "keeps its reservation before a worker recovery note",
+ resume: true,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(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{
+ Source: "check",
+ ID: "reserved",
+ Text: "A result with a prepared log ID.",
+ }
+
+ 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())
+
+ expected := first.Queue(t).Pending[0].Text
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ require.False(t, second.loops[0].waiting())
+ require.Empty(t, second.events[0].Kinds())
+
+ if tc.resume {
+ path := filepath.Join(second.agents[0].agent.dir, subagentWorkFileName)
+ require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
+ require.NoError(t, second.loops[0].runs.Resume(t.Context()))
+
+ note, err := second.agents[0].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.")
+ require.NoError(t, err)
+ require.Equal(t, 1, submitted.ID)
+ }
+
+ restored, err := second.agents[0].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)
+ require.Empty(t, second.Queue(t).Pending)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, id, second.events[0].LastEvent("work").ID)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Len(t, second.Queue(t).Notifications, 1)
+ })
+ }
+}
+
+func TestNotificationBatchMakesProgressWithEscapedControlCharacters(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{
+ Source: "check",
+ ID: "escaped",
+ Text: "data" + strings.Repeat("\x00", notifyTextBytes-4),
+ })
+ require.NoError(t, err)
+ require.Greater(t, len(test.Queue(t).Pending[0].Text), 64<<10)
+ test.Start(t, 0)
+ test.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, test.Queue(t).Pending)
+}
+
projects / onechat / blobdiff