diff --git a/cmd/gateway.go b/cmd/gateway.go index 4278288c..7bd49d55 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -512,16 +512,37 @@ func runGateway() { // Team progress notification subscriber — forwards task events to chat channels. // Reads team.settings.notifications config; direct mode sends outbound, leader mode - // injects into leader agent session. + // injects into leader agent session. Notifications are batched per chat + // with 2s debounce to avoid spamming users when multiple tasks dispatch at once. if pgStores.Teams != nil { notifyTeamStore := pgStores.Teams notifyAgentStore := pgStores.Agents + teamNotifyQueue := tools.NewTeamNotifyQueue(2000, func(items []string, meta tools.NotifyRoutingMeta) { + content := tools.FormatBatchedNotify(items) + if meta.Mode == "leader" { + leaderContent := fmt.Sprintf("[Auto-status — relay to user, NO task actions]\n%s\n\nBriefly inform the user. Do NOT create, retry, reassign, or modify any tasks.", content) + msgBus.TryPublishInbound(bus.InboundMessage{ + Channel: meta.Channel, + SenderID: "notification:progress", + ChatID: meta.ChatID, + AgentID: meta.LeadAgent, + UserID: meta.UserID, + Content: leaderContent, + }) + } else { + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: meta.Channel, + ChatID: meta.ChatID, + Content: content, + }) + } + }) msgBus.Subscribe("consumer.team-notify", func(evt bus.Event) { payload, ok := evt.Payload.(protocol.TeamTaskEventPayload) if !ok || payload.TeamID == "" || payload.Channel == "" { return } - // Only forward assigned/failed events (completed handled by announce-back). + // Only forward assigned/failed/progress events (completed handled by announce-back). var notifyType string switch evt.Name { case protocol.EventTeamTaskAssigned: @@ -588,9 +609,9 @@ func runGateway() { content = fmt.Sprintf("❌ Task #%d \"%s\" failed: %s", payload.TaskNumber, payload.Subject, reason) } + // Resolve leader agent for leader mode routing. + var leadAgent string if cfg.Mode == "leader" { - // Route through leader agent — model reformulates. - leadAgent := "" if notifyAgentStore != nil { if la, err := notifyAgentStore.GetByID(context.Background(), team.LeadAgentID); err == nil { leadAgent = la.AgentKey @@ -599,23 +620,16 @@ func runGateway() { if leadAgent == "" { return } - leaderContent := fmt.Sprintf("[Auto-status — relay to user, NO task actions]\n%s\n\nBriefly inform the user. Do NOT create, retry, reassign, or modify any tasks.", content) - msgBus.TryPublishInbound(bus.InboundMessage{ - Channel: payload.Channel, - SenderID: "notification:progress", - ChatID: payload.ChatID, - AgentID: leadAgent, - UserID: payload.UserID, - Content: leaderContent, - }) - } else { - // Direct mode — send outbound directly to channel. - msgBus.PublishOutbound(bus.OutboundMessage{ - Channel: payload.Channel, - ChatID: payload.ChatID, - Content: content, - }) } + + batchKey := payload.TeamID + ":" + payload.ChatID + teamNotifyQueue.Enqueue(batchKey, content, tools.NotifyRoutingMeta{ + Mode: cfg.Mode, + Channel: payload.Channel, + ChatID: payload.ChatID, + UserID: payload.UserID, + LeadAgent: leadAgent, + }) }) slog.Info("team progress notification subscriber registered") } diff --git a/internal/store/pg/teams_tasks_progress.go b/internal/store/pg/teams_tasks_progress.go index 4d863dfa..56e682ef 100644 --- a/internal/store/pg/teams_tasks_progress.go +++ b/internal/store/pg/teams_tasks_progress.go @@ -180,7 +180,8 @@ func scanRecoveredTaskInfoRows(rows interface { func (s *PGTeamStore) ResetTaskStatus(ctx context.Context, taskID, teamID uuid.UUID) error { now := time.Now() res, err := s.db.ExecContext(ctx, - `UPDATE team_tasks SET status = $1, locked_at = NULL, lock_expires_at = NULL, result = NULL, updated_at = $2 + `UPDATE team_tasks SET status = $1, locked_at = NULL, lock_expires_at = NULL, result = NULL, + progress_percent = NULL, progress_step = NULL, updated_at = $2 WHERE id = $3 AND team_id = $4 AND status IN ($5, $6)`, store.TeamTaskStatusPending, now, taskID, teamID, store.TeamTaskStatusStale, store.TeamTaskStatusFailed, diff --git a/internal/tools/team_notify_queue.go b/internal/tools/team_notify_queue.go new file mode 100644 index 00000000..3575bf03 --- /dev/null +++ b/internal/tools/team_notify_queue.go @@ -0,0 +1,102 @@ +package tools + +import ( + "strings" + "sync" + "time" +) + +// NotifyRoutingMeta carries routing info for batched team notifications. +type NotifyRoutingMeta struct { + Mode string // "direct" or "leader" + Channel string + ChatID string + UserID string + LeadAgent string // agent key (only used in leader mode) +} + +// TeamNotifyQueue batches team task notifications per chat with debounce, +// following the same pattern as AnnounceQueue for subagent results. +type TeamNotifyQueue struct { + mu sync.Mutex + batches map[string]*notifyBatch // key: "teamID:chatID" + debounce time.Duration + cap int // immediate drain threshold + onDrain func(items []string, meta NotifyRoutingMeta) +} + +type notifyBatch struct { + items []string + timer *time.Timer + meta NotifyRoutingMeta +} + +// NewTeamNotifyQueue creates a notification queue with debounce and drain callback. +func NewTeamNotifyQueue(debounceMs int, onDrain func(items []string, meta NotifyRoutingMeta)) *TeamNotifyQueue { + if debounceMs <= 0 { + debounceMs = 2000 + } + return &TeamNotifyQueue{ + batches: make(map[string]*notifyBatch), + debounce: time.Duration(debounceMs) * time.Millisecond, + cap: 20, + onDrain: onDrain, + } +} + +// Enqueue adds a formatted notification line to the batch for the given key. +// Resets debounce timer. Drains immediately if cap is reached. +func (q *TeamNotifyQueue) Enqueue(key string, content string, meta NotifyRoutingMeta) { + q.mu.Lock() + defer q.mu.Unlock() + + b, ok := q.batches[key] + if !ok { + b = ¬ifyBatch{meta: meta} + q.batches[key] = b + } + b.items = append(b.items, content) + + // Immediate drain if cap reached. + if len(b.items) >= q.cap { + if b.timer != nil { + b.timer.Stop() + } + items := b.items + bMeta := b.meta + delete(q.batches, key) + go q.drain(items, bMeta) + return + } + + // Reset debounce timer. + if b.timer != nil { + b.timer.Stop() + } + b.timer = time.AfterFunc(q.debounce, func() { + q.mu.Lock() + b, ok := q.batches[key] + if !ok { + q.mu.Unlock() + return + } + items := b.items + bMeta := b.meta + delete(q.batches, key) + q.mu.Unlock() + + q.drain(items, bMeta) + }) +} + +func (q *TeamNotifyQueue) drain(items []string, meta NotifyRoutingMeta) { + if len(items) == 0 || q.onDrain == nil { + return + } + q.onDrain(items, meta) +} + +// FormatBatchedNotify joins notification lines into a single message. +func FormatBatchedNotify(items []string) string { + return strings.Join(items, "\n") +} diff --git a/internal/tools/team_tasks_followup.go b/internal/tools/team_tasks_followup.go index 1da59268..422600f0 100644 --- a/internal/tools/team_tasks_followup.go +++ b/internal/tools/team_tasks_followup.go @@ -136,6 +136,8 @@ func (t *TeamTasksTool) executeRetry(ctx context.Context, args map[string]any) * t.manager.broadcastTeamEvent(protocol.EventTeamTaskAssigned, protocol.TeamTaskEventPayload{ TeamID: team.ID.String(), TaskID: taskID.String(), + TaskNumber: task.TaskNumber, + Subject: task.Subject, Status: store.TeamTaskStatusInProgress, OwnerAgentKey: t.manager.agentKeyFromID(ctx, *task.OwnerAgentID), UserID: store.UserIDFromContext(ctx), diff --git a/internal/tools/team_tasks_mutations.go b/internal/tools/team_tasks_mutations.go index 862cd478..5e41179b 100644 --- a/internal/tools/team_tasks_mutations.go +++ b/internal/tools/team_tasks_mutations.go @@ -176,21 +176,6 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any) ActorType: "agent", ActorID: agentKey, }) - t.manager.broadcastTeamEvent(protocol.EventTeamTaskAssigned, protocol.TeamTaskEventPayload{ - TeamID: team.ID.String(), - TaskID: task.ID.String(), - TaskNumber: task.TaskNumber, - Subject: task.Subject, - Status: status, - OwnerAgentKey: t.manager.agentKeyFromID(ctx, assigneeID), - UserID: store.UserIDFromContext(ctx), - Channel: ToolChannelFromCtx(ctx), - ChatID: chatID, - Timestamp: task.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"), - ActorType: "agent", - ActorID: agentKey, - }) - // Track for post-turn dispatch. If no post-turn hook (e.g. HTTP API), dispatch immediately. if status == store.TeamTaskStatusPending { if ptd := PendingTeamDispatchFromCtx(ctx); ptd != nil {