修雨
5e9a920b76
fix(connect): prevent agent mid-turn blocking Co-authored-by: 修雨 <47820304+PeterGuy326@users.noreply.github.com>
102 lines
3.0 KiB
Go
102 lines
3.0 KiB
Go
// Copyright 2026 Alibaba Group
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package helpers
|
|
|
|
import (
|
|
"reflect"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// TestConvQueueCoalescesSameConversationBurst: while one conversation is
|
|
// running, later messages are batched into one follow-up turn instead of forming
|
|
// a long FIFO of stale prompts.
|
|
func TestConvQueueCoalescesSameConversationBurst(t *testing.T) {
|
|
q := newConvQueue()
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
batches := make(chan []string, 2)
|
|
var once sync.Once
|
|
process := func(batch []connectQueuedTurn) {
|
|
var got []string
|
|
for _, turn := range batch {
|
|
got = append(got, turn.text)
|
|
}
|
|
batches <- got
|
|
once.Do(func() {
|
|
close(started)
|
|
<-release
|
|
})
|
|
}
|
|
|
|
q.submit(connectQueuedTurn{convID: "conv-1", text: "first"}, process)
|
|
<-started
|
|
q.submit(connectQueuedTurn{convID: "conv-1", text: "second"}, process)
|
|
q.submit(connectQueuedTurn{convID: "conv-1", text: "third"}, process)
|
|
close(release)
|
|
|
|
if got := <-batches; !reflect.DeepEqual(got, []string{"first"}) {
|
|
t.Fatalf("first batch = %v, want [first]", got)
|
|
}
|
|
if got := <-batches; !reflect.DeepEqual(got, []string{"second", "third"}) {
|
|
t.Fatalf("pending batch = %v, want [second third]", got)
|
|
}
|
|
}
|
|
|
|
// TestConvQueueParallelAcrossConversations: a slow conversation must not
|
|
// block another conversation.
|
|
func TestConvQueueParallelAcrossConversations(t *testing.T) {
|
|
q := newConvQueue()
|
|
slowRelease := make(chan struct{})
|
|
slowStarted := make(chan struct{})
|
|
q.submit(connectQueuedTurn{convID: "conv-slow", text: "slow"}, func([]connectQueuedTurn) {
|
|
close(slowStarted)
|
|
<-slowRelease
|
|
})
|
|
<-slowStarted
|
|
|
|
fastDone := make(chan struct{})
|
|
q.submit(connectQueuedTurn{convID: "conv-fast", text: "fast"}, func([]connectQueuedTurn) { close(fastDone) })
|
|
select {
|
|
case <-fastDone:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("fast conversation blocked behind slow one")
|
|
}
|
|
close(slowRelease)
|
|
}
|
|
|
|
// TestConvQueueDrainsEntries: once a conversation's queue drains, its chain
|
|
// entry is dropped (idle chats hold no memory).
|
|
func TestConvQueueDrainsEntries(t *testing.T) {
|
|
q := newConvQueue()
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
q.submit(connectQueuedTurn{convID: "conv-1", text: "done"}, func([]connectQueuedTurn) { wg.Done() })
|
|
wg.Wait()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for {
|
|
q.mu.Lock()
|
|
n := len(q.states)
|
|
q.mu.Unlock()
|
|
if n == 0 {
|
|
return
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("queue entries not cleaned up: %d left", n)
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|