From 64ca26e611dedb8dd1099f0a54e361192a582f1e Mon Sep 17 00:00:00 2001 From: viettranx Date: Thu, 5 Mar 2026 14:20:20 +0700 Subject: [PATCH] fix(delegation): serialize announces per session to prevent stale history reads Concurrent announce runs (delegate + subagent) on the same leader session could read stale session history because the agent loop reads at start and writes at end. Added per-session mutex via sync.Map to serialize announces, ensuring each reads up-to-date history. Also removed unnecessary task preview truncation in delegation events (WS clients handle display formatting). Co-Authored-By: Claude Opus 4.6 --- cmd/gateway_consumer.go | 45 ++++++++++++++++++++++++------- internal/tools/delegate_events.go | 6 +---- 2 files changed, 36 insertions(+), 15 deletions(-) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 4e3a51b8..e44fa017 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "strings" + "sync" "time" "github.com/google/uuid" @@ -49,6 +50,17 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // TTL=20min, max=5000 entries — prevents webhook retries / double-taps from duplicating agent runs. dedupe := bus.NewDedupeCache(20*time.Minute, 5000) + // Per-session announce serialization: prevents concurrent announce runs from + // reading stale session history. Without this, Announce #2 can start while + // Announce #1 is still running, read history that doesn't include Announce #1's + // messages (written only after agent loop completes), and generate responses + // with wrong context (e.g. "waiting for Tiểu La" when Tiểu La already finished). + var announceMu sync.Map // sessionKey → *sync.Mutex + getAnnounceMu := func(key string) *sync.Mutex { + v, _ := announceMu.LoadOrStore(key, &sync.Mutex{}) + return v.(*sync.Mutex) + } + // processNormalMessage handles routing, scheduling, and response delivery for a single // (possibly merged) inbound message. Called directly by the debouncer's flush callback. processNormalMessage := func(msg bus.InboundMessage) { @@ -393,8 +405,8 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Build outbound metadata for topic/thread routing. outMeta := buildAnnounceOutMeta(origLocalKey) - // Schedule through subagent lane - outCh := sched.Schedule(ctx, scheduler.LaneSubagent, agent.RunRequest{ + // Build request before goroutine to capture msg fields. + announceReq := agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, ForwardMedia: msg.Media, @@ -407,10 +419,17 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Stream: false, ParentTraceID: parentTraceID, ParentRootSpanID: parentRootSpanID, - }) + } - // Handle result asynchronously to not block the consumer loop - go func(origCh, chatID, senderID, label string, meta map[string]string) { + // Handle announce asynchronously with per-session serialization. + // The mutex ensures concurrent announces for the same session wait for + // each other, so each reads up-to-date session history. + go func(sessionKey, origCh, chatID, senderID, label string, meta map[string]string, req agent.RunRequest) { + mu := getAnnounceMu(sessionKey) + mu.Lock() + defer mu.Unlock() + + outCh := sched.Schedule(ctx, scheduler.LaneSubagent, req) outcome := <-outCh if outcome.Err != nil { if errors.Is(outcome.Err, context.Canceled) { @@ -455,7 +474,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"], outMeta) + }(sessionKey, origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"], outMeta, announceReq) continue } @@ -506,7 +525,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Build outbound metadata for topic/thread routing. outMeta := buildAnnounceOutMeta(origLocalKey) - outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ + announceReq := agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, ForwardMedia: msg.Media, @@ -519,9 +538,15 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Stream: false, ParentTraceID: parentTraceID, ParentRootSpanID: parentRootSpanID, - }) + } - go func(origCh, chatID, senderID string, meta map[string]string) { + // Same per-session serialization as subagent announce above. + go func(sessionKey, origCh, chatID, senderID string, meta map[string]string, req agent.RunRequest) { + mu := getAnnounceMu(sessionKey) + mu.Lock() + defer mu.Unlock() + + outCh := sched.Schedule(ctx, scheduler.LaneDelegate, req) outcome := <-outCh if outcome.Err != nil { if errors.Is(outcome.Err, context.Canceled) { @@ -559,7 +584,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(origChannel, msg.ChatID, msg.SenderID, outMeta) + }(sessionKey, origChannel, msg.ChatID, msg.SenderID, outMeta, announceReq) continue } diff --git a/internal/tools/delegate_events.go b/internal/tools/delegate_events.go index 76ae224c..c9902c18 100644 --- a/internal/tools/delegate_events.go +++ b/internal/tools/delegate_events.go @@ -42,10 +42,6 @@ func (dm *DelegateManager) emitDelegationEventWithError(task *DelegationTask, er // buildDelegationPayload creates a DelegationEventPayload from a DelegationTask. func buildDelegationPayload(task *DelegationTask) protocol.DelegationEventPayload { - taskPreview := task.Task - if len(taskPreview) > 200 { - taskPreview = taskPreview[:200] + "..." - } payload := protocol.DelegationEventPayload{ DelegationID: task.ID, SourceAgentID: task.SourceAgentID.String(), @@ -58,7 +54,7 @@ func buildDelegationPayload(task *DelegationTask) protocol.DelegationEventPayloa Channel: task.OriginChannel, ChatID: task.OriginChatID, Mode: task.Mode, - Task: taskPreview, + Task: task.Task, Status: task.Status, CreatedAt: task.CreatedAt.UTC().Format(time.RFC3339), }