From 014f74ec1545fbdbbd507bd645bbeb340f8da667 Mon Sep 17 00:00:00 2001 From: viettranx Date: Mon, 30 Mar 2026 15:20:01 +0700 Subject: [PATCH] fix(agent): group session unresponsive during team task execution (#266) Two fixes: 1. Remove assistant prefill from team task reminders. The injected [user]+[assistant]+[user] pattern caused LLMs to treat the canned ack as "turn complete", returning NO_REPLY for every user message in group sessions with active tasks. Reminders are now merged into the user message as prefix tags. 2. Add PeerKind propagation to team notification routing. TaskTicker and progress notifications were missing PeerKind on InboundMessage, causing them to route to phantom DM sessions instead of the correct group session. PeerKind is now carried through event payloads, notify queue metadata, and all inbound message publications. --- cmd/gateway.go | 2 ++ cmd/gateway_consumer_handlers.go | 11 ++++++----- cmd/gateway_consumer_post_turn.go | 10 ++++++++++ internal/agent/loop_team_reminders.go | 26 ++++++++++++-------------- internal/tasks/task_ticker.go | 9 +++++++++ internal/tools/team_event_helpers.go | 8 +++++++- internal/tools/team_notify_queue.go | 1 + internal/tools/team_tasks_blocker.go | 6 ++++++ internal/tools/team_tasks_create.go | 1 + internal/tools/team_tool_dispatch.go | 5 +++++ internal/tools/team_tool_validation.go | 6 ++++++ pkg/protocol/team_events.go | 1 + 12 files changed, 66 insertions(+), 20 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index dda07bf1..985a179b 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -677,6 +677,7 @@ func runGateway() { ChatID: meta.ChatID, AgentID: meta.LeadAgent, UserID: meta.UserID, + PeerKind: meta.PeerKind, Content: leaderContent, Metadata: map[string]string{"run_kind": tools.RunKindNotification}, }) @@ -825,6 +826,7 @@ func runGateway() { ChatID: payload.ChatID, UserID: payload.UserID, LeadAgent: leadAgentKey, + PeerKind: payload.PeerKind, }) }) slog.Info("team progress notification subscriber registered") diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index 3b28e09d..a0124b94 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/cmd/gateway_consumer_handlers.go @@ -309,11 +309,12 @@ func handleTeammateMessage( teamID, _ := uuid.Parse(inMeta[tools.MetaTeamID]) if teamTaskID != uuid.Nil { meta := teammateTaskMeta{ - TaskID: teamTaskID, - TeamID: teamID, - ToAgent: inMeta[tools.MetaToAgent], - Channel: inMeta[tools.MetaOriginChannel], - ChatID: inMeta[tools.MetaOriginChatID], + TaskID: teamTaskID, + TeamID: teamID, + ToAgent: inMeta[tools.MetaToAgent], + Channel: inMeta[tools.MetaOriginChannel], + ChatID: inMeta[tools.MetaOriginChatID], + PeerKind: inMeta[tools.MetaOriginPeerKind], } cachedTeam = resolveTeamTaskOutcome(ctx, deps, outcome, taskActionFlags, meta) } diff --git a/cmd/gateway_consumer_post_turn.go b/cmd/gateway_consumer_post_turn.go index 1ef590f9..8239e9c2 100644 --- a/cmd/gateway_consumer_post_turn.go +++ b/cmd/gateway_consumer_post_turn.go @@ -51,6 +51,7 @@ type teammateTaskMeta struct { ToAgent string Channel string ChatID string + PeerKind string // "group" or "direct" — for correct notification routing (#266) Subject string TaskNumber int } @@ -96,6 +97,7 @@ func resolveTeamTaskOutcome( taskNumber := meta.TaskNumber taskChannel := meta.Channel taskChatID := meta.ChatID + taskPeerKind := meta.PeerKind // Enrich with live task data if available. if currentTask != nil { @@ -111,6 +113,11 @@ func resolveTeamTaskOutcome( if currentTask.ChatID != "" { taskChatID = currentTask.ChatID } + if taskPeerKind == "" && currentTask.Metadata != nil { + if pk, ok := currentTask.Metadata[tools.TaskMetaPeerKind].(string); ok && pk != "" { + taskPeerKind = pk + } + } } // Smart post-turn decision based on action flags. @@ -129,6 +136,7 @@ func resolveTeamTaskOutcome( tools.WithReason(outcome.Err.Error()), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), + tools.WithPeerKind(taskPeerKind), tools.WithTimestamp(now), )) } @@ -162,6 +170,7 @@ func resolveTeamTaskOutcome( tools.WithReason("loop_detector_kill"), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), + tools.WithPeerKind(taskPeerKind), tools.WithTimestamp(now), )) } @@ -195,6 +204,7 @@ func resolveTeamTaskOutcome( tools.WithOwnerAgentKey(toAgent), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), + tools.WithPeerKind(taskPeerKind), tools.WithTimestamp(now), )) } diff --git a/internal/agent/loop_team_reminders.go b/internal/agent/loop_team_reminders.go index 3138882c..159e5f34 100644 --- a/internal/agent/loop_team_reminders.go +++ b/internal/agent/loop_team_reminders.go @@ -64,14 +64,14 @@ func (l *Loop) injectTeamTaskReminders(ctx context.Context, req *RunRequest, mes } if len(parts) > 0 { reminder := "[System] " + strings.Join(parts, "\n\n") - // Pop user message, inject reminder, push user message back + // Merge reminder into the user message as a prefix tag. + // Previous approach injected [user]+[assistant]+[user] which caused + // LLMs to treat the assistant ack as "turn complete" → NO_REPLY (#266). userMsg := messages[len(messages)-1] - messages = messages[:len(messages)-1] - messages = append(messages, - providers.Message{Role: "user", Content: reminder}, - providers.Message{Role: "assistant", Content: "I see the task status. Let me handle accordingly."}, - userMsg, - ) + messages[len(messages)-1] = providers.Message{ + Role: "user", + Content: "[Active team tasks]\n" + reminder + "\n[/Active team tasks]\n\n" + userMsg.Content, + } } } } @@ -90,14 +90,12 @@ func (l *Loop) injectTeamTaskReminders(ctx context.Context, req *RunRequest, mes "Stay focused on this task. Your final response becomes the task result — make it clear and complete. "+ "For long tasks, report progress: team_tasks(action=\"progress\", percent=50, text=\"status\").", task.TaskNumber, task.Subject) - // Pop user message, inject reminder, push user message back + // Merge reminder into user message as prefix tag (#266). userMsg := messages[len(messages)-1] - messages = messages[:len(messages)-1] - messages = append(messages, - providers.Message{Role: "user", Content: reminder}, - providers.Message{Role: "assistant", Content: "Understood. I'll focus on this task and report progress."}, - userMsg, - ) + messages[len(messages)-1] = providers.Message{ + Role: "user", + Content: "[Task context]\n" + reminder + "\n[/Task context]\n\n" + userMsg.Content, + } } } } diff --git a/internal/tasks/task_ticker.go b/internal/tasks/task_ticker.go index 1c086ddb..13ccf9a3 100644 --- a/internal/tasks/task_ticker.go +++ b/internal/tasks/task_ticker.go @@ -235,12 +235,21 @@ func (t *TaskTicker) notifyLeaders(ctx context.Context, tasks []store.RecoveredT chatID = scope.TeamID.String() } + // Resolve PeerKind from first task's metadata for correct session routing (#266). + var peerKind string + if fullTask, err := t.teams.GetTask(ctx, scopeTasks[0].ID); err == nil && fullTask != nil && fullTask.Metadata != nil { + if pk, ok := fullTask.Metadata["peer_kind"].(string); ok { + peerKind = pk + } + } + if !t.msgBus.TryPublishInbound(bus.InboundMessage{ Channel: channel, SenderID: "ticker:system", ChatID: chatID, AgentID: lead.AgentKey, UserID: team.CreatedBy, + PeerKind: peerKind, TenantID: scope.TenantID, Content: content, }) { diff --git a/internal/tools/team_event_helpers.go b/internal/tools/team_event_helpers.go index b277da29..a7a66fae 100644 --- a/internal/tools/team_event_helpers.go +++ b/internal/tools/team_event_helpers.go @@ -90,6 +90,11 @@ func WithChatID(id string) TaskEventOption { return func(p *protocol.TeamTaskEventPayload) { p.ChatID = id } } +// WithPeerKind sets PeerKind on the payload for correct session routing (#266). +func WithPeerKind(pk string) TaskEventOption { + return func(p *protocol.TeamTaskEventPayload) { p.PeerKind = pk } +} + // WithCommentText sets CommentText on the payload. func WithCommentText(t string) TaskEventOption { return func(p *protocol.TeamTaskEventPayload) { p.CommentText = t } @@ -103,13 +108,14 @@ func WithProgress(percent int, step string) TaskEventOption { } } -// WithContextInfo extracts UserID, Channel, and ChatID from the context +// WithContextInfo extracts UserID, Channel, ChatID, and PeerKind from the context // using standard tool context accessors. func WithContextInfo(ctx context.Context) TaskEventOption { return func(p *protocol.TeamTaskEventPayload) { p.UserID = store.UserIDFromContext(ctx) p.Channel = ToolChannelFromCtx(ctx) p.ChatID = ToolChatIDFromCtx(ctx) + p.PeerKind = ToolPeerKindFromCtx(ctx) } } diff --git a/internal/tools/team_notify_queue.go b/internal/tools/team_notify_queue.go index 3575bf03..fc720639 100644 --- a/internal/tools/team_notify_queue.go +++ b/internal/tools/team_notify_queue.go @@ -13,6 +13,7 @@ type NotifyRoutingMeta struct { ChatID string UserID string LeadAgent string // agent key (only used in leader mode) + PeerKind string // "group" or "direct" — routes to correct session (#266) } // TeamNotifyQueue batches team task notifications per chat with debounce, diff --git a/internal/tools/team_tasks_blocker.go b/internal/tools/team_tasks_blocker.go index 2b11b2bb..a994fd77 100644 --- a/internal/tools/team_tasks_blocker.go +++ b/internal/tools/team_tasks_blocker.go @@ -45,6 +45,10 @@ func (t *TeamTasksTool) handleBlockerComment( // 2. Notify subscriber → "❌ Task failed" → chat channel (direct outbound) // 3. WS broadcast → web UI dashboard real-time update memberKey := t.manager.AgentKeyFromID(ctx, agentID) + blockerPeerKind := "" + if pk, ok := task.Metadata[TaskMetaPeerKind].(string); ok { + blockerPeerKind = pk + } t.manager.BroadcastTeamEvent(ctx, protocol.EventTeamTaskFailed, BuildTaskEventPayload( team.ID.String(), taskID.String(), store.TeamTaskStatusFailed, @@ -55,6 +59,7 @@ func (t *TeamTasksTool) handleBlockerComment( WithUserID(store.UserIDFromContext(ctx)), WithChannel(task.Channel), WithChatID(task.ChatID), + WithPeerKind(blockerPeerKind), )) // Escalate to leader if enabled in team settings. @@ -74,6 +79,7 @@ func (t *TeamTasksTool) handleBlockerComment( ChatID: task.ChatID, Content: escalationMsg, UserID: store.UserIDFromContext(ctx), + PeerKind: blockerPeerKind, TenantID: store.TenantIDFromContext(ctx), AgentID: leadAg.AgentKey, }) { diff --git a/internal/tools/team_tasks_create.go b/internal/tools/team_tasks_create.go index 00b8bbe3..405c7b6c 100644 --- a/internal/tools/team_tasks_create.go +++ b/internal/tools/team_tasks_create.go @@ -290,6 +290,7 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any) WithOwnerAgentKey(t.manager.AgentKeyFromID(ctx, assigneeID)), WithChannel(task.Channel), WithChatID(task.ChatID), + WithPeerKind(ToolPeerKindFromCtx(ctx)), )) t.manager.DispatchTaskToAgent(ctx, task, team, assigneeID) } diff --git a/internal/tools/team_tool_dispatch.go b/internal/tools/team_tool_dispatch.go index 894ca50c..51e6f0be 100644 --- a/internal/tools/team_tool_dispatch.go +++ b/internal/tools/team_tool_dispatch.go @@ -359,6 +359,10 @@ func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uui continue } dispatched[ownerID] = true + taskPeerKind := "" + if pk, ok := task.Metadata[TaskMetaPeerKind].(string); ok { + taskPeerKind = pk + } m.broadcastTeamEvent(ctx, protocol.EventTeamTaskDispatched, BuildTaskEventPayload( teamID.String(), task.ID.String(), store.TeamTaskStatusInProgress, @@ -367,6 +371,7 @@ func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uui WithOwnerAgentKey(m.agentKeyFromID(ctx, ownerID)), WithChannel(task.Channel), WithChatID(task.ChatID), + WithPeerKind(taskPeerKind), )) // Append completed blocker results so the member agent has context. diff --git a/internal/tools/team_tool_validation.go b/internal/tools/team_tool_validation.go index f0c459bc..81f0dc57 100644 --- a/internal/tools/team_tool_validation.go +++ b/internal/tools/team_tool_validation.go @@ -98,6 +98,10 @@ func (m *TeamToolManager) ProcessPendingTasks(ctx context.Context, teamID uuid.U slog.Warn("post_turn: assign failed", "task_id", task.ID, "error", err) continue } + taskPeerKind := "" + if pk, ok := task.Metadata[TaskMetaPeerKind].(string); ok { + taskPeerKind = pk + } m.broadcastTeamEvent(ctx, protocol.EventTeamTaskDispatched, BuildTaskEventPayload( teamID.String(), task.ID.String(), store.TeamTaskStatusInProgress, @@ -106,6 +110,7 @@ func (m *TeamToolManager) ProcessPendingTasks(ctx context.Context, teamID uuid.U WithOwnerAgentKey(m.agentKeyFromID(ctx, *task.OwnerAgentID)), WithChannel(task.Channel), WithChatID(task.ChatID), + WithPeerKind(taskPeerKind), )) // Restore leader's trace context from task metadata (ctx here is the // consumer goroutine context which has no trace after the turn ends). @@ -177,6 +182,7 @@ func (m *TeamToolManager) notifyLeaderCycleError(ctx context.Context, teamID uui ChatID: chatID, AgentID: leadAgent.AgentKey, UserID: team.CreatedBy, + PeerKind: ToolPeerKindFromCtx(ctx), TenantID: store.TenantIDFromContext(ctx), Content: content, }) diff --git a/pkg/protocol/team_events.go b/pkg/protocol/team_events.go index 1a359388..5af4fcf0 100644 --- a/pkg/protocol/team_events.go +++ b/pkg/protocol/team_events.go @@ -100,6 +100,7 @@ type TeamTaskEventPayload struct { UserID string `json:"user_id"` Channel string `json:"channel"` ChatID string `json:"chat_id"` + PeerKind string `json:"peer_kind,omitempty"` // "group" or "direct" — for correct session routing (#266) Timestamp string `json:"timestamp"` // Comment text preview (for team.task.commented events, truncated).