refactor(teams): batch teammate announce via queue with dedup

Extract announce logic from handleTeammateMessage into a dedicated
announce queue that batches concurrent task completions. Handles
failure reporting and deduplicates rapid-fire announces.

cmd/gateway_announce_queue.go

@@ -0,0 +1,264 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+	"sync"
+
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/agent"
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
+	"github.com/nextlevelbuilder/goclaw/internal/config"
+	"github.com/nextlevelbuilder/goclaw/internal/scheduler"
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+	"github.com/nextlevelbuilder/goclaw/internal/tools"
+)
+
+// announceEntry holds one teammate completion result waiting to be announced.
+type announceEntry struct {
+	MemberAgent string
+	Content     string
+	Media       []agent.MediaResult
+}
+
+// announceQueueState tracks the per-session announce queue.
+// Producer-consumer: multiple goroutines add entries, one loops to drain+announce.
+type announceQueueState struct {
+	mu      sync.Mutex
+	running bool
+	entries []announceEntry
+}
+
+// announceQueues maps leadSessionKey → queue. Cleaned up when queue finishes.
+var announceQueues sync.Map
+
+func getOrCreateAnnounceQueue(key string) *announceQueueState {
+	v, _ := announceQueues.LoadOrStore(key, &announceQueueState{})
+	return v.(*announceQueueState)
+}
+
+// enqueueAnnounce adds a result to the queue. Returns (queue, isProcessor).
+// If isProcessor=true, the caller must run processAnnounceLoop.
+func enqueueAnnounce(key string, entry announceEntry) (*announceQueueState, bool) {
+	q := getOrCreateAnnounceQueue(key)
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.entries = append(q.entries, entry)
+	if q.running {
+		return q, false
+	}
+	q.running = true
+	return q, true
+}
+
+func (q *announceQueueState) drain() []announceEntry {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	out := q.entries
+	q.entries = nil
+	return out
+}
+
+// tryFinish atomically checks for pending entries and marks the queue idle.
+// Returns true if the processor should exit (no pending entries).
+// Prevents TOCTOU race between hasPending() and finish().
+func (q *announceQueueState) tryFinish(key string) bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if len(q.entries) > 0 {
+		return false // more work arrived — keep processing
+	}
+	q.running = false
+	announceQueues.Delete(key)
+	return true
+}
+
+// announceRouting holds the shared routing info captured by the first goroutine.
+type announceRouting struct {
+	LeadAgent        string
+	LeadSessionKey   string
+	OrigChannel      string
+	OrigChatID       string
+	OrigPeerKind     string
+	OrigLocalKey     string
+	OriginUserID     string
+	TeamID           string
+	TeamWorkspace    string
+	OriginTraceID    string
+	ParentTraceID    uuid.UUID
+	ParentRootSpanID uuid.UUID
+	OutMeta          map[string]string
+}
+
+// processAnnounceLoop drains entries, builds merged announce, schedules to leader.
+// Loops until queue is empty.
+func processAnnounceLoop(
+	ctx context.Context,
+	q *announceQueueState,
+	r announceRouting,
+	sched *scheduler.Scheduler,
+	msgBus *bus.MessageBus,
+	teamStore store.TeamStore,
+	postTurn tools.PostTurnProcessor,
+	cfg *config.Config,
+) {
+	for {
+		entries := q.drain()
+		if len(entries) == 0 {
+			if q.tryFinish(r.LeadSessionKey) {
+				return
+			}
+			continue // entries arrived between drain and tryFinish
+		}
+
+		// Build task board snapshot (latest state, after all completions in this batch).
+		snapshot := ""
+		if r.TeamID != "" && r.OriginTraceID != "" {
+			if teamUUID, err := uuid.Parse(r.TeamID); err == nil {
+				snapshot = buildTaskBoardSnapshot(ctx, teamStore, teamUUID, r.OrigChatID, r.OriginTraceID)
+			}
+		}
+
+		content := buildMergedAnnounceContent(entries, snapshot, r.TeamWorkspace)
+
+		req := agent.RunRequest{
+			SessionKey:       r.LeadSessionKey,
+			Message:          content,
+			Channel:          r.OrigChannel,
+			ChatID:           r.OrigChatID,
+			PeerKind:         r.OrigPeerKind,
+			LocalKey:         r.OrigLocalKey,
+			UserID:           r.OriginUserID,
+			RunID:            fmt.Sprintf("teammate-announce-%s-%d", r.LeadAgent, len(entries)),
+			RunKind:          "announce",
+			HideInput:        true,
+			Stream:           false,
+			TeamID:           r.TeamID,
+			ParentTraceID:    r.ParentTraceID,
+			ParentRootSpanID: r.ParentRootSpanID,
+		}
+		// Collect all media from entries.
+		for _, e := range entries {
+			for _, mr := range e.Media {
+				req.ForwardMedia = append(req.ForwardMedia, bus.MediaFile{
+					Path:     mr.Path,
+					MimeType: mr.ContentType,
+				})
+			}
+		}
+		// WS channel has no outbound media handler — deliver via ContentSuffix.
+		if r.OrigChannel == "ws" && len(req.ForwardMedia) > 0 {
+			req.ContentSuffix = mediaToMarkdownFromPaths(req.ForwardMedia, cfg)
+			req.ForwardMedia = nil
+		}
+
+		// Inject post-turn tracker (leader may create new tasks during announce).
+		ptd := tools.NewPendingTeamDispatch()
+		defer ptd.ReleaseTeamLock() // ensure lock released even on panic
+		schedCtx := tools.WithPendingTeamDispatch(ctx, ptd)
+		outCh := sched.Schedule(schedCtx, scheduler.LaneSubagent, req)
+		outcome := <-outCh
+
+		ptd.ReleaseTeamLock()
+		if postTurn != nil {
+			for tid, tIDs := range ptd.Drain() {
+				if err := postTurn.ProcessPendingTasks(ctx, tid, tIDs); err != nil {
+					slog.Warn("post_turn(announce): failed", "team_id", tid, "error", err)
+				}
+			}
+		}
+
+		if outcome.Err != nil {
+			slog.Error("teammate announce: lead run failed", "error", outcome.Err, "batch_size", len(entries))
+		} else {
+			isSilent := outcome.Result.Content == "" || agent.IsSilentReply(outcome.Result.Content)
+			if !(isSilent && len(outcome.Result.Media) == 0) {
+				out := outcome.Result.Content
+				if isSilent {
+					out = ""
+				}
+				outMsg := bus.OutboundMessage{
+					Channel:  r.OrigChannel,
+					ChatID:   r.OrigChatID,
+					Content:  out,
+					Metadata: r.OutMeta,
+				}
+				appendMediaToOutbound(&outMsg, outcome.Result.Media)
+				msgBus.PublishOutbound(outMsg)
+			}
+		}
+
+		slog.Info("teammate announce: batch processed",
+			"batch_size", len(entries), "session", r.LeadSessionKey)
+
+		// Loop back — tryFinish at top will exit when queue is truly empty.
+	}
+}
+
+// buildMergedAnnounceContent creates the announce message for one or more completed/failed tasks.
+func buildMergedAnnounceContent(entries []announceEntry, taskBoardSnapshot, teamWorkspace string) string {
+	var sb strings.Builder
+
+	if len(entries) == 1 {
+		e := entries[0]
+		if strings.HasPrefix(e.Content, "[FAILED]") {
+			fmt.Fprintf(&sb, "[System Message] Team member %q failed to complete task.\n\nError: %s", e.MemberAgent, strings.TrimPrefix(e.Content, "[FAILED] "))
+			sb.WriteString("\n\nInform the user about the failure. You may suggest retrying with team_tasks(action=\"retry\", task_id=\"...\") if appropriate.")
+		} else {
+			fmt.Fprintf(&sb, "[System Message] Team member %q completed task.\n\nResult:\n%s", e.MemberAgent, e.Content)
+		}
+	} else {
+		// Count successes vs failures for header.
+		var failed, succeeded int
+		for _, e := range entries {
+			if strings.HasPrefix(e.Content, "[FAILED]") {
+				failed++
+			} else {
+				succeeded++
+			}
+		}
+		if failed > 0 && succeeded > 0 {
+			fmt.Fprintf(&sb, "[System Message] %d task(s) completed, %d task(s) failed.\n", succeeded, failed)
+		} else if failed > 0 {
+			fmt.Fprintf(&sb, "[System Message] %d task(s) failed.\n", failed)
+		} else {
+			fmt.Fprintf(&sb, "[System Message] %d team tasks completed.\n", succeeded)
+		}
+		for _, e := range entries {
+			if strings.HasPrefix(e.Content, "[FAILED]") {
+				fmt.Fprintf(&sb, "\n--- FAILED: %q ---\nError: %s\n", e.MemberAgent, strings.TrimPrefix(e.Content, "[FAILED] "))
+			} else {
+				fmt.Fprintf(&sb, "\n--- Result from %q ---\n%s\n", e.MemberAgent, e.Content)
+			}
+		}
+		if failed > 0 {
+			sb.WriteString("\nFor failed tasks, you may suggest retrying with team_tasks(action=\"retry\", task_id=\"...\").")
+		}
+	}
+
+	if taskBoardSnapshot != "" {
+		sb.WriteString("\n\n")
+		sb.WriteString(taskBoardSnapshot)
+	}
+
+	// Batch-aware prompting: guide leader to summarize vs acknowledge.
+	allDone := strings.Contains(taskBoardSnapshot, "All ") && strings.Contains(taskBoardSnapshot, " completed")
+	if allDone {
+		sb.WriteString("\n\nAll tasks in this batch are completed. Present a comprehensive summary of ALL results to the user.")
+	} else if taskBoardSnapshot != "" {
+		sb.WriteString("\n\nSome tasks are still in progress. Briefly acknowledge this result (1-2 sentences). A full summary will come when all tasks complete.")
+	} else {
+		sb.WriteString("\n\nPresent this result to the user.")
+	}
+
+	sb.WriteString(" Any media files are forwarded automatically. Do NOT search for files — the results above contain all relevant information.")
+
+	if teamWorkspace != "" {
+		fmt.Fprintf(&sb, "\n[Team workspace: %s — use read_file/list_files to access shared files]", teamWorkspace)
+	}
+
+	return sb.String()
+}

cmd/gateway_consumer_handlers.go

@@ -542,27 +542,32 @@ func handleTeammateMessage(
 			}
 		}
 
+		// Determine announce content: success result or failure error.
+		var announceContent string
+		var announceMedia []agent.MediaResult
 		if outcome.Err != nil {
 			slog.Error("teammate message: agent run failed", "error", outcome.Err)
-			return
-		}
-		if (outcome.Result.Content == "" && len(outcome.Result.Media) == 0) || agent.IsSilentReply(outcome.Result.Content) {
+			errMsg := outcome.Err.Error()
+			if len(errMsg) > 500 {
+				errMsg = errMsg[:500] + "..."
+			}
+			announceContent = fmt.Sprintf("[FAILED] %s", errMsg)
+		} else if (outcome.Result.Content == "" && len(outcome.Result.Media) == 0) || agent.IsSilentReply(outcome.Result.Content) {
 			slog.Info("teammate message: suppressed silent/empty reply", "from", senderID)
 			return
+		} else {
+			announceContent = outcome.Result.Content
+			announceMedia = outcome.Result.Media
 		}
 
-		// Announce result to lead agent (same pattern as subagent announce).
-		// The lead reformulates the result and presents to the user.
-		// Extract parent trace context so the announce run nests under the lead's trace.
-		var announceParentTraceID, announceParentRootSpanID uuid.UUID
-		if tid := inMeta["origin_trace_id"]; tid != "" {
-			announceParentTraceID, _ = uuid.Parse(tid)
-		}
-		if sid := inMeta["origin_root_span_id"]; sid != "" {
-			announceParentRootSpanID, _ = uuid.Parse(sid)
+		// Announce result (or failure) to lead agent via announce queue.
+		// Queue merges concurrent completions into a single batched announce.
+		if origChatID == "" {
+			slog.Warn("teammate announce: no origin_chat_id, cannot announce to lead")
+			return
 		}
 
-		// Resolve lead from team — reuse cachedTeam to avoid duplicate DB call.
+		// Resolve lead agent.
 		leadAgent := ""
 		if cachedTeam != nil {
 			if leadAg, err := agentStore.GetByID(ctx, cachedTeam.LeadAgentID); err == nil {
@@ -583,36 +588,7 @@ func handleTeammateMessage(
 		if leadAgent == "" {
 			leadAgent = cfg.ResolveDefaultAgentID()
 		}
-		memberAgent := inMeta["to_agent"]
 
-		// Build task board snapshot scoped to this batch (same origin_trace_id).
-		taskBoardSnapshot := ""
-		if teamIDStr := inMeta["team_id"]; teamIDStr != "" {
-			if teamUUID, err := uuid.Parse(teamIDStr); err == nil {
-				taskBoardSnapshot = buildTaskBoardSnapshot(ctx, teamStore, teamUUID, inMeta["origin_chat_id"], inMeta["origin_trace_id"])
-			}
-		}
-
-		announceContent := fmt.Sprintf(
-			"[System Message] Team member %q completed task.\n\nResult:\n%s",
-			memberAgent, outcome.Result.Content,
-		)
-		if taskBoardSnapshot != "" {
-			announceContent += "\n\n" + taskBoardSnapshot
-		}
-		announceContent += "\n\nPresent this result to the user. Any media files are forwarded automatically. Do NOT search for files — the result above contains all relevant information."
-		// Append team workspace path so lead can locate files without searching.
-		if ws := inMeta["team_workspace"]; ws != "" {
-			announceContent += fmt.Sprintf("\n[Team workspace: %s — use read_file/list_files to access shared files, e.g. list_files(path=\".\") then read_file(path=\"filename.md\")]", ws)
-		}
-
-		// Route to the lead's session on the original channel/chat.
-		if origChatID == "" {
-			slog.Warn("teammate announce: no origin_chat_id, cannot announce to lead")
-			return
-		}
-		// Use the original peer kind so announce writes to the correct session
-		// (group session for group chats, direct session for DMs).
 		origPeerKind := inMeta["origin_peer_kind"]
 		if origPeerKind == "" {
 			origPeerKind = string(sessions.PeerDirect)
@@ -621,69 +597,44 @@ func handleTeammateMessage(
 		leadSessionKey := sessions.BuildScopedSessionKey(leadAgent, origCh, sessions.PeerKind(origPeerKind), origChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey)
 		leadSessionKey = overrideSessionKeyFromLocalKey(leadSessionKey, origLocalKey, leadAgent, origCh, origChatID, origPeerKind)
 
-		announceReq := agent.RunRequest{
-			SessionKey:       leadSessionKey,
-			Message:          announceContent,
-			Channel:          origCh,
-			ChatID:           origChatID,
-			PeerKind:         origPeerKind,
-			LocalKey:         origLocalKey,
-			UserID:           inMeta["origin_user_id"],
-			RunID:            fmt.Sprintf("teammate-announce-%s", memberAgent),
-			RunKind:          "announce",
-			HideInput:        true,
-			Stream:           false,
+		// Extract trace context for announce linking.
+		var parentTraceID, parentRootSpanID uuid.UUID
+		if tid := inMeta["origin_trace_id"]; tid != "" {
+			parentTraceID, _ = uuid.Parse(tid)
+		}
+		if sid := inMeta["origin_root_span_id"]; sid != "" {
+			parentRootSpanID, _ = uuid.Parse(sid)
+		}
+
+		// Enqueue result. If we become the processor, run the announce loop.
+		entry := announceEntry{
+			MemberAgent: inMeta["to_agent"],
+			Content:     announceContent,
+			Media:       announceMedia,
+		}
+		q, isProcessor := enqueueAnnounce(leadSessionKey, entry)
+		if !isProcessor {
+			slog.Info("teammate announce: merged into pending batch",
+				"member", entry.MemberAgent, "session", leadSessionKey)
+			return
+		}
+
+		routing := announceRouting{
+			LeadAgent:        leadAgent,
+			LeadSessionKey:   leadSessionKey,
+			OrigChannel:      origCh,
+			OrigChatID:       origChatID,
+			OrigPeerKind:     origPeerKind,
+			OrigLocalKey:     origLocalKey,
+			OriginUserID:     inMeta["origin_user_id"],
 			TeamID:           inMeta["team_id"],
-			ParentTraceID:    announceParentTraceID,
-			ParentRootSpanID: announceParentRootSpanID,
+			TeamWorkspace:    inMeta["team_workspace"],
+			OriginTraceID:    inMeta["origin_trace_id"],
+			ParentTraceID:    parentTraceID,
+			ParentRootSpanID: parentRootSpanID,
+			OutMeta:          meta,
 		}
-		for _, mr := range outcome.Result.Media {
-			announceReq.ForwardMedia = append(announceReq.ForwardMedia, bus.MediaFile{
-				Path:     mr.Path,
-				MimeType: mr.ContentType,
-			})
-		}
-
-		// Inject post-turn tracker for announce run (leader may create new tasks).
-		announcePtd := tools.NewPendingTeamDispatch()
-		announceCtx := tools.WithPendingTeamDispatch(ctx, announcePtd)
-		announceOutCh := sched.Schedule(announceCtx, scheduler.LaneSubagent, announceReq)
-		announceOutcome := <-announceOutCh
-
-		// Release team create lock — tasks already visible in DB, safe for other goroutines to list.
-		announcePtd.ReleaseTeamLock()
-
-		// Post-turn: dispatch pending team tasks created during announce.
-		if postTurn != nil {
-			for tid, tIDs := range announcePtd.Drain() {
-				if err := postTurn.ProcessPendingTasks(ctx, tid, tIDs); err != nil {
-					slog.Warn("post_turn(announce): failed", "team_id", tid, "error", err)
-				}
-			}
-		}
-
-		if announceOutcome.Err != nil {
-			slog.Error("teammate announce: lead run failed", "error", announceOutcome.Err)
-			return
-		}
-
-		isSilent := announceOutcome.Result.Content == "" || agent.IsSilentReply(announceOutcome.Result.Content)
-		if isSilent && len(announceOutcome.Result.Media) == 0 {
-			return
-		}
-
-		announceOut := announceOutcome.Result.Content
-		if isSilent {
-			announceOut = ""
-		}
-		outMsg := bus.OutboundMessage{
-			Channel:  origCh,
-			ChatID:   origChatID,
-			Content:  announceOut,
-			Metadata: meta,
-		}
-		appendMediaToOutbound(&outMsg, announceOutcome.Result.Media)
-		msgBus.PublishOutbound(outMsg)
+		processAnnounceLoop(ctx, q, routing, sched, msgBus, teamStore, postTurn, cfg)
 	}(origChannel, origChatID, msg.SenderID, taskIDStr, outMeta, msg.Metadata)
 
 	return true