mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 09:28:14 +00:00
feat: webui 消息重放双重防护(client_msg_id 去重 + agent 内容级去重)
回应群聊 08-19 消息轰炸诊断(GUI SSE 重连导致消息重放):
1. webui 层 client_msg_id 单飞去重(与 GUI c29abe9 配套):
- /api/v1/chat 解析 client_msg_id, 同 ID 重放等待首次结果直接复用
- 响应带 deduplicated=true 标记; 无 ID 旧客户端完全兼容
- FIFO 缓存上限 256 条防泄漏
2. agent 核心层内容级短窗口去重(兜底无 ID 客户端):
- isDuplicateInput: source+content 为 key, 10s 窗口内重复丢弃
- 持续轰炸时刷新时间戳保持拦截; 过期项自动清理
测试: webui 去重三场景 + agent 核心去重行为验证, 全项目 go test 通过
This commit is contained in:
JianFeeeee
@ -111,6 +111,11 @@ type Agent struct {
|
||||
noMergeMarkers map[string]int
|
||||
noMergeMu sync.Mutex
|
||||
|
||||
// 输入去重：防 webui/GUI 断线重连导致的消息重放
|
||||
// key=source+"|"+content, value=上次接收时间；短窗口内同内容丢弃
|
||||
lastInput map[string]time.Time
|
||||
lastInputMu sync.Mutex
|
||||
|
||||
// 词嵌入模型，用于实体语义相似度计算
|
||||
embedder *memory.StaticEmbedder
|
||||
}
|
||||
@ -221,6 +226,7 @@ func New(cfg AgentConfig) *Agent {
|
||||
inputCfg: cfg.InputProcessing,
|
||||
embedder: embedder,
|
||||
noMergeMarkers: make(map[string]int),
|
||||
lastInput: make(map[string]time.Time),
|
||||
}
|
||||
}
|
||||
|
||||
@ -240,6 +246,34 @@ func (a *Agent) Stop() {
|
||||
|
||||
func (a *Agent) ID() types.AgentID { return a.id }
|
||||
|
||||
// isDuplicateInput 判断是否为短窗口内的重复输入（防 webui/GUI 断线重连消息重放）。
|
||||
// key=source+"|"+content；窗口内重复返回 true 并刷新时间戳（持续轰炸时保持拦截）。
|
||||
const duplicateInputWindow = 10 * time.Second
|
||||
|
||||
func (a *Agent) isDuplicateInput(source, content string) bool {
|
||||
a.lastInputMu.Lock()
|
||||
defer a.lastInputMu.Unlock()
|
||||
now := time.Now()
|
||||
key := source + "|" + content
|
||||
if last, ok := a.lastInput[key]; ok && now.Sub(last) < duplicateInputWindow {
|
||||
a.lastInput[key] = now
|
||||
return true
|
||||
}
|
||||
a.lastInput[key] = now
|
||||
// 顺带清理过期项，防止 map 无限增长
|
||||
for k, t := range a.lastInput {
|
||||
if now.Sub(t) > duplicateInputWindow {
|
||||
delete(a.lastInput, k)
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// IsDuplicateInput 导出包装，供测试验证去重行为。
|
||||
func (a *Agent) IsDuplicateInput(source, content string) bool {
|
||||
return a.isDuplicateInput(source, content)
|
||||
}
|
||||
|
||||
// SelfInputChan 返回自循环输入通道（只读，供内部测试验证）
|
||||
func (a *Agent) SelfInputChan() <-chan string {
|
||||
return a.selfInputCh
|
||||
|
||||
@ -124,6 +124,11 @@ func (a *Agent) handleInput(evt *agentIO.InputEvent) {
|
||||
if input == "" {
|
||||
return
|
||||
}
|
||||
// 去重：webui/GUI 断线重连会重放未确认消息，短窗口内同来源同内容丢弃，避免轰炸
|
||||
if a.isDuplicateInput(evt.Source, input) {
|
||||
log.Printf("[agent] dropped duplicate input from %s: %s", evt.Source, truncateStr(input, 60))
|
||||
return
|
||||
}
|
||||
a.processTextInput(evt, input)
|
||||
|
||||
case "image", "audio":
|
||||
|
||||
@ -88,6 +88,10 @@ type Handler struct {
|
||||
chatMu sync.Mutex
|
||||
chatHistory []ChatMsg
|
||||
pendingIdx int // chatHistory 中正在进行的 assistant 消息索引，-1 表示无
|
||||
|
||||
chatMsgMu sync.Mutex
|
||||
chatMsgCache map[string]*chatMsgEntry // client_msg_id -> 首次处理结果
|
||||
chatMsgOrder []string // FIFO 淘汰序
|
||||
cmdMu sync.Mutex
|
||||
cmdHistory []CmdExec
|
||||
termMu sync.Mutex
|
||||
@ -135,6 +139,40 @@ const maxChatHistory = 200
|
||||
const maxCmdHistory = 100
|
||||
const maxTerminals = 50
|
||||
|
||||
// ===== client_msg_id 去重（防 GUI 断线重连/超时重试导致的消息重放）=====
|
||||
// GUI 端每条发送消息带唯一 client_msg_id；服务端按 ID 单飞（singleflight）：
|
||||
// 首次请求正常注入 agent，同 ID 重放等待首次结果并直接复用，不再重复处理。
|
||||
|
||||
const maxChatMsgCache = 256
|
||||
|
||||
type chatMsgEntry struct {
|
||||
done chan struct{}
|
||||
resp *agentIO.OutputEvent
|
||||
}
|
||||
|
||||
func (h *Handler) claimChatMsg(id string) (*chatMsgEntry, bool) {
|
||||
h.chatMsgMu.Lock()
|
||||
defer h.chatMsgMu.Unlock()
|
||||
if e, ok := h.chatMsgCache[id]; ok {
|
||||
return e, true
|
||||
}
|
||||
e := &chatMsgEntry{done: make(chan struct{})}
|
||||
h.chatMsgCache[id] = e
|
||||
h.chatMsgOrder = append(h.chatMsgOrder, id)
|
||||
if len(h.chatMsgOrder) > maxChatMsgCache {
|
||||
old := h.chatMsgOrder[0]
|
||||
h.chatMsgOrder = h.chatMsgOrder[1:]
|
||||
delete(h.chatMsgCache, old)
|
||||
}
|
||||
return e, false
|
||||
}
|
||||
|
||||
// completeChatMsg 记录首次处理结果并唤醒所有等待的同 ID 重放请求。
|
||||
func (h *Handler) completeChatMsg(e *chatMsgEntry, resp *agentIO.OutputEvent) {
|
||||
e.resp = resp
|
||||
close(e.done)
|
||||
}
|
||||
|
||||
func NewHandler(s *sdk.PluginSDK) *Handler {
|
||||
var (
|
||||
sup sdk.SupervisorAPI
|
||||
@ -175,6 +213,7 @@ func NewHandler(s *sdk.PluginSDK) *Handler {
|
||||
sessions: make(map[string]time.Time),
|
||||
termStates: make(map[string]*termState),
|
||||
pendingIdx: -1,
|
||||
chatMsgCache: make(map[string]*chatMsgEntry),
|
||||
}
|
||||
h.loadChatHistory()
|
||||
if s != nil {
|
||||
@ -1159,9 +1198,10 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
var body struct {
|
||||
Message string `json:"message"`
|
||||
DeviceID string `json:"device_id"` // 消息来源设备（GUI/受控设备），可选
|
||||
DeviceName string `json:"device_name"` // 设备显示名，可选
|
||||
Message string `json:"message"`
|
||||
DeviceID string `json:"device_id"` // 消息来源设备（GUI/受控设备），可选
|
||||
DeviceName string `json:"device_name"` // 设备显示名，可选
|
||||
ClientMsgID string `json:"client_msg_id"` // 客户端唯一消息 ID（防断线重放/超时重试）
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
|
||||
@ -1176,6 +1216,36 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
|
||||
return
|
||||
}
|
||||
|
||||
// client_msg_id 去重：同 ID 重放等待首次结果直接复用，不重复注入 agent。
|
||||
// 无 ID 的旧客户端走原路径（agent 核心层另有内容级短窗口去重兑底）。
|
||||
var entry *chatMsgEntry
|
||||
if body.ClientMsgID != "" {
|
||||
var replay bool
|
||||
entry, replay = h.claimChatMsg(body.ClientMsgID)
|
||||
if replay {
|
||||
log.Printf("[webui] duplicate chat msg %s: waiting for first request result", body.ClientMsgID)
|
||||
select {
|
||||
case <-entry.done:
|
||||
resp := entry.resp
|
||||
if resp == nil {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
|
||||
return
|
||||
}
|
||||
content, _ := resp.Payload["content"].(string)
|
||||
reasoning, _ := resp.Payload["reasoning_content"].(string)
|
||||
result := map[string]interface{}{"response": content, "deduplicated": true}
|
||||
if reasoning != "" {
|
||||
result["reasoning_content"] = reasoning
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// 来源编码：带设备身份时用 webui/{device_id}（agent 经 injectSourceContext 可见来源）；
|
||||
// 无设备时保持 webui（兼容旧调用）。device_name 一并注入便于 agent 识别。
|
||||
source := "webui"
|
||||
@ -1187,7 +1257,10 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
payload["device_id"] = body.DeviceID
|
||||
payload["device_name"] = body.DeviceName
|
||||
}
|
||||
// 带超时的上下文，防止 InjectTextSync 长时间阻塞 HTTP 请求
|
||||
if body.ClientMsgID != "" {
|
||||
payload["client_msg_id"] = body.ClientMsgID
|
||||
}
|
||||
// 带超时的上下文，防止 InjectInputSync 长时间阻塞 HTTP 请求
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@ -1200,10 +1273,17 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
select {
|
||||
case resp = <-respCh:
|
||||
case <-ctx.Done():
|
||||
if entry != nil {
|
||||
h.completeChatMsg(entry, nil)
|
||||
}
|
||||
writeJSON(w, http.StatusGatewayTimeout, map[string]string{"error": "agent timeout (60s)"})
|
||||
return
|
||||
}
|
||||
|
||||
if entry != nil {
|
||||
h.completeChatMsg(entry, resp)
|
||||
}
|
||||
|
||||
if resp == nil {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
|
||||
return
|
||||
|
||||
@ -841,3 +841,141 @@ func TestHandleCompletionsEndToEnd(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ===== client_msg_id 去重测试（防 GUI 断线重连消息重放）=====
|
||||
|
||||
func TestHandleChatClientMsgIDDedup(t *testing.T) {
|
||||
iom := agentIO.NewIOManager()
|
||||
|
||||
memDB, err := memory.NewGraphDB(t.TempDir() + "/graph.db")
|
||||
if err != nil {
|
||||
t.Fatalf("NewGraphDB: %v", err)
|
||||
}
|
||||
defer memDB.Close()
|
||||
|
||||
pm := agentAPI.NewProviderManager()
|
||||
pm.Register("echo", &echoProvider{name: "echo"})
|
||||
|
||||
agent := agentCore.New(agentCore.AgentConfig{
|
||||
ID: "test",
|
||||
SystemPrompt: "你是测试助手",
|
||||
Provider: &echoProvider{name: "echo"},
|
||||
ProviderManager: pm,
|
||||
IO: iom,
|
||||
Memory: memDB,
|
||||
})
|
||||
agent.Start()
|
||||
defer agent.Stop()
|
||||
|
||||
sup := supervisor.New(&types.Config{
|
||||
Daemon: types.DaemonConfig{
|
||||
CheckInterval: time.Minute,
|
||||
HeartbeatInterval: 30 * time.Second,
|
||||
},
|
||||
})
|
||||
sup.Start()
|
||||
defer sup.Shutdown()
|
||||
|
||||
s := testSDK(sdk.SDKConfig{
|
||||
Supervisor: supervisor.NewSDKAdapter(sup),
|
||||
IOManager: iom,
|
||||
Config: sdk.NewConfig(&types.Config{}),
|
||||
})
|
||||
h := NewHandler(s)
|
||||
|
||||
t.Run("same_client_msg_id_replay_returns_cached_response", func(t *testing.T) {
|
||||
body := `{"message":"你好","client_msg_id":"msg-abc-123"}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
h.handleChat(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("first request: expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var first map[string]interface{}
|
||||
json.NewDecoder(w.Body).Decode(&first)
|
||||
if first["response"] != "echo: 你好" {
|
||||
t.Fatalf("expected echo response, got %v", first["response"])
|
||||
}
|
||||
|
||||
// 同 ID 重放：应直接复用首次结果，不重复注入 agent
|
||||
req2 := httptest.NewRequest(http.MethodPost, "/api/v1/chat", strings.NewReader(body))
|
||||
w2 := httptest.NewRecorder()
|
||||
h.handleChat(w2, req2)
|
||||
if w2.Code != http.StatusOK {
|
||||
t.Fatalf("replay: expected 200, got %d: %s", w2.Code, w2.Body.String())
|
||||
}
|
||||
var second map[string]interface{}
|
||||
json.NewDecoder(w2.Body).Decode(&second)
|
||||
if second["response"] != "echo: 你好" {
|
||||
t.Fatalf("replay expected same response, got %v", second["response"])
|
||||
}
|
||||
if second["deduplicated"] != true {
|
||||
t.Fatalf("replay expected deduplicated=true, got %v", second["deduplicated"])
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("different_client_msg_id_processed_normally", func(t *testing.T) {
|
||||
body := `{"message":"第二条","client_msg_id":"msg-def-456"}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
h.handleChat(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var resp map[string]interface{}
|
||||
json.NewDecoder(w.Body).Decode(&resp)
|
||||
if resp["deduplicated"] == true {
|
||||
t.Fatal("new msg id should not be deduplicated")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("no_client_msg_id_backward_compatible", func(t *testing.T) {
|
||||
body := `{"message":"旧客户端消息"}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
h.handleChat(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ===== agent 核心层内容级去重测试 =====
|
||||
|
||||
func TestAgentDuplicateInputDedup(t *testing.T) {
|
||||
iom := agentIO.NewIOManager()
|
||||
|
||||
memDB, err := memory.NewGraphDB(t.TempDir() + "/graph.db")
|
||||
if err != nil {
|
||||
t.Fatalf("NewGraphDB: %v", err)
|
||||
}
|
||||
defer memDB.Close()
|
||||
|
||||
pm := agentAPI.NewProviderManager()
|
||||
pm.Register("echo", &echoProvider{name: "echo"})
|
||||
|
||||
agent := agentCore.New(agentCore.AgentConfig{
|
||||
ID: "test",
|
||||
SystemPrompt: "你是测试助手",
|
||||
Provider: &echoProvider{name: "echo"},
|
||||
ProviderManager: pm,
|
||||
IO: iom,
|
||||
Memory: memDB,
|
||||
})
|
||||
agent.Start()
|
||||
defer agent.Stop()
|
||||
|
||||
// 直接验证 isDuplicateInput 行为
|
||||
if agent.IsDuplicateInput("webui", "重复消息") {
|
||||
t.Fatal("first input should not be duplicate")
|
||||
}
|
||||
if !agent.IsDuplicateInput("webui", "重复消息") {
|
||||
t.Fatal("immediate same-content same-source should be duplicate")
|
||||
}
|
||||
if agent.IsDuplicateInput("webui", "不同消息") {
|
||||
t.Fatal("different content should not be duplicate")
|
||||
}
|
||||
if agent.IsDuplicateInput("qq", "重复消息") {
|
||||
t.Fatal("different source should not be duplicate")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user