修雨
5e9a920b76
fix(connect): prevent agent mid-turn blocking Co-authored-by: 修雨 <47820304+PeterGuy326@users.noreply.github.com>
94 lines
2.7 KiB
Go
94 lines
2.7 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 "sync"
|
|
|
|
// convQueue serialises agent turns per conversation while coalescing bursts.
|
|
// Same-chat agent calls must never run in parallel — two turns racing on one
|
|
// session corrupt the transcript. But a pure FIFO is also bad UX: if a user
|
|
// keeps sending clarifications while a long turn is running, the bot should not
|
|
// spend the next several minutes answering stale intermediate prompts. Instead,
|
|
// messages arriving during the active turn accumulate in one pending batch; when
|
|
// the active turn finishes, that whole batch is processed as a single follow-up.
|
|
// Different conversations still run in parallel.
|
|
type convQueue struct {
|
|
mu sync.Mutex
|
|
states map[string]*convQueueState
|
|
}
|
|
|
|
type convQueueState struct {
|
|
running bool
|
|
pending []connectQueuedTurn
|
|
process func([]connectQueuedTurn)
|
|
}
|
|
|
|
func newConvQueue() *convQueue {
|
|
return &convQueue{states: map[string]*convQueueState{}}
|
|
}
|
|
|
|
// submit schedules turn handling and returns immediately (the Stream callback
|
|
// must ack fast). If the conversation is already running, the turn is appended
|
|
// to that conversation's pending batch instead of creating another queued worker.
|
|
func (q *convQueue) submit(turn connectQueuedTurn, process func([]connectQueuedTurn)) {
|
|
convID := turn.convID
|
|
q.mu.Lock()
|
|
st := q.states[convID]
|
|
if st == nil {
|
|
st = &convQueueState{}
|
|
q.states[convID] = st
|
|
}
|
|
if st.running {
|
|
st.pending = append(st.pending, turn)
|
|
if process != nil {
|
|
st.process = process
|
|
}
|
|
q.mu.Unlock()
|
|
return
|
|
}
|
|
st.running = true
|
|
st.process = process
|
|
q.mu.Unlock()
|
|
go q.drain(convID, []connectQueuedTurn{turn})
|
|
}
|
|
|
|
func (q *convQueue) drain(convID string, batch []connectQueuedTurn) {
|
|
for {
|
|
process := q.processFor(convID)
|
|
if process != nil {
|
|
process(batch)
|
|
}
|
|
|
|
q.mu.Lock()
|
|
st := q.states[convID]
|
|
if st == nil || len(st.pending) == 0 {
|
|
delete(q.states, convID)
|
|
q.mu.Unlock()
|
|
return
|
|
}
|
|
batch = append([]connectQueuedTurn(nil), st.pending...)
|
|
st.pending = nil
|
|
q.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (q *convQueue) processFor(convID string) func([]connectQueuedTurn) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if st := q.states[convID]; st != nil {
|
|
return st.process
|
|
}
|
|
return nil
|
|
}
|