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 <noreply@anthropic.com>

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
 		}
 

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),
 	}