diff --git a/cmd/gateway_announce_queue.go b/cmd/gateway_announce_queue.go new file mode 100644 index 00000000..52e2c66d --- /dev/null +++ b/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() +} diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index 409e46c3..4435e3c2 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/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