goclaw/internal/tools/team_tool_dispatch.go

package tools

import (
	"context"
	"fmt"
	"log/slog"
	"path/filepath"
	"strings"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/bus"
	"github.com/nextlevelbuilder/goclaw/internal/store"
	"github.com/nextlevelbuilder/goclaw/internal/tracing"
	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)

// ============================================================
// TeamToolBackend exported wrappers (dispatch layer)
// ============================================================

func (m *TeamToolManager) DispatchTaskToAgent(ctx context.Context, task *store.TeamTaskData, team *store.TeamData, agentID uuid.UUID) {
	m.dispatchTaskToAgent(ctx, task, team, agentID)
}
func (m *TeamToolManager) BuildBlockerResultsSummary(ctx context.Context, task *store.TeamTaskData) string {
	return m.buildBlockerResultsSummary(ctx, task)
}
func (m *TeamToolManager) BuildRecentCommentsSummary(ctx context.Context, taskID uuid.UUID) string {
	return m.buildRecentCommentsSummary(ctx, taskID)
}
func (m *TeamToolManager) RestoreTraceContext(ctx context.Context, task *store.TeamTaskData) context.Context {
	return m.restoreTraceContext(ctx, task)
}

// maxTaskDispatches is the max number of times a single task can be dispatched
// before it auto-fails. Prevents infinite loops when agents can't complete a task.
const maxTaskDispatches = 3

// dispatchTaskToAgent publishes a teammate-style inbound message so the
// gateway consumer picks it up and runs the assigned agent, then auto-completes
// the task on success or auto-fails on error.
//
// Routing uses task.Channel/task.ChatID (set at creation time) as primary source,
// falling back to ctx only for initial dispatch when the task is created and dispatched
// in the same call. This ensures correct routing even when called from
// DispatchUnblockedTasks (where ctx is the member agent's context, not the lead's).
func (m *TeamToolManager) dispatchTaskToAgent(ctx context.Context, task *store.TeamTaskData, team *store.TeamData, agentID uuid.UUID) {
	if m.msgBus == nil {
		return
	}
	teamID := team.ID

	// Safety net: never dispatch to the lead agent — causes dual-session loop.
	// Self-assignment is blocked at create time, but catch edge cases
	// from retry, ticker recovery, or manual DB edits.
	if agentID == team.LeadAgentID {
		slog.Warn("team_tasks.dispatch: blocked dispatch to lead agent",
			"task_id", task.ID, "agent_id", agentID, "team_id", teamID)
		_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{
			"status": store.TeamTaskStatusFailed,
			"result": "Cannot dispatch task to the team lead — reassign to a team member",
		})
		return
	}

	// Circuit breaker: auto-fail tasks that have been dispatched too many times.
	dispatchCount := 0
	if dc, ok := task.Metadata["dispatch_count"].(float64); ok {
		dispatchCount = int(dc)
	}
	if dispatchCount >= maxTaskDispatches {
		slog.Warn("team_tasks.dispatch: max dispatch count reached, auto-failing task",
			"task_id", task.ID, "dispatch_count", dispatchCount)
		failReason := fmt.Sprintf("Task auto-failed after %d dispatch attempts", dispatchCount)
		_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{
			"status": store.TeamTaskStatusFailed,
			"result": failReason,
		})
		return
	}

	// Increment dispatch count in metadata.
	if task.Metadata == nil {
		task.Metadata = make(map[string]any)
	}
	task.Metadata["dispatch_count"] = dispatchCount + 1
	_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{"metadata": task.Metadata})

	ag, err := m.cachedGetAgentByID(ctx, agentID)
	if err != nil {
		slog.Warn("team_tasks.dispatch: cannot resolve agent", "agent_id", agentID, "error", err)
		return
	}

	var content strings.Builder
	content.WriteString(fmt.Sprintf("[Assigned task #%d (id: %s)]: %s", task.TaskNumber, task.ID, task.Subject))
	if task.Description != "" {
		content.WriteString("\n\n" + task.Description)
	}
	// Hint: tell the agent it's on a team task and where the shared workspace is.
	if ws := taskTeamWorkspace(task); ws != "" {
		content.WriteString(fmt.Sprintf("\n\n[Team workspace: %s — use read_file/write_file/list_files to access shared files. All files you write are visible to the team lead and other members.]", ws))
	}
	// List attached files so member knows what's available to read.
	if files, ok := task.Metadata["attached_files"].([]any); ok && len(files) > 0 {
		content.WriteString("\n\n[Attached files in team workspace — use read_file to access:]")
		for _, f := range files {
			if path, ok := f.(string); ok {
				content.WriteString("\n- attachments/" + filepath.Base(path))
			}
		}
	}

	// Hint: guide member on available team_tasks actions.
	content.WriteString("\n\n[Instructions]\n" +
		"- Use team_tasks(action=\"progress\", percent=N, text=\"...\") to report progress\n" +
		"- Use team_tasks(action=\"comment\", text=\"...\") to share findings\n" +
		"- Use team_tasks(action=\"comment\", type=\"blocker\", text=\"...\") when BLOCKED and need leader input — auto-fails task and notifies leader\n" +
		"- When done: team_tasks(action=\"complete\", result=\"summary of your work\")\n" +
		"- Write output files to team workspace so lead can review")

	// Use task's stored channel/chat as primary source for routing.
	// Falls back to ctx values for initial dispatch (task just created, fields match ctx).
	originChannel := task.Channel
	if originChannel == "" {
		originChannel = OriginChannelFromCtx(ctx)
	}
	originChatID := task.ChatID
	if originChatID == "" {
		originChatID = OriginChatIDFromCtx(ctx)
	}
	// Resolve lead agent key for completion announce routing.
	fromAgent := ToolAgentKeyFromCtx(ctx)
	if leadAg, err := m.cachedGetAgentByID(ctx, team.LeadAgentID); err == nil {
		fromAgent = leadAg.AgentKey
	}

	// Resolve user ID: prefer context (available during leader's turn),
	// fall back to task's chat ID (stable for dispatches from consumer/ticker context).
	originUserID := store.UserIDFromContext(ctx)
	if originUserID == "" {
		originUserID = originChatID
	}

	// Preserve real acting sender so permission checks on the teammate's
	// turn (e.g. write_file in group chat) attribute to the original user
	// rather than the synthetic "teammate:dashboard" sender (#915).
	// For deferred dispatches (ticker/unblock), context has no sender — fall back
	// to origin_sender_id stored in task metadata at creation time.
	originSenderID := store.SenderIDFromContext(ctx)
	if originSenderID == "" || bus.IsInternalSender(originSenderID) {
		if s, ok := task.Metadata["origin_sender_id"].(string); ok && s != "" && !bus.IsInternalSender(s) {
			originSenderID = s
		}
	}

	// Resolve peer kind from context; fallback to task metadata, then "direct".
	originPeerKind := ToolPeerKindFromCtx(ctx)
	if originPeerKind == "" {
		if pk, ok := task.Metadata[TaskMetaPeerKind].(string); ok && pk != "" {
			originPeerKind = pk
		} else {
			originPeerKind = "direct"
		}
	}

	meta := map[string]string{
		MetaOriginChannel:   originChannel,
		MetaOriginPeerKind:  originPeerKind,
		MetaOriginChatID:    originChatID,
		MetaOriginUserID:    originUserID,
		MetaFromAgent:       fromAgent,
		MetaToAgent:         ag.AgentKey,
		MetaToAgentDisplay:  ag.DisplayName,
		MetaTeamTaskID:      task.ID.String(),
		MetaTeamID:          teamID.String(),
	}
	if originSenderID != "" {
		meta[MetaOriginSenderID] = originSenderID
	}
	// Role propagation: prefer context, fall back to task metadata for deferred dispatches.
	originRole := store.RoleFromContext(ctx)
	if originRole == "" {
		if r, ok := task.Metadata["origin_role"].(string); ok && r != "" {
			originRole = r
		}
	}
	if originRole != "" {
		meta[MetaOriginRole] = originRole
	}
	// Resolve local key from context; fallback to task metadata for deferred dispatches.
	localKey := ToolLocalKeyFromCtx(ctx)
	if localKey == "" {
		if lk, ok := task.Metadata[TaskMetaLocalKey].(string); ok {
			localKey = lk
		}
	}
	if localKey != "" {
		meta[MetaOriginLocalKey] = localKey
	}
	// Resolve origin session key from context; fallback to task metadata for deferred dispatches.
	// WS sessions use non-standard key format that BuildScopedSessionKey() cannot reproduce.
	originSessionKey := ToolSessionKeyFromCtx(ctx)
	if originSessionKey == "" {
		if sk, ok := task.Metadata[TaskMetaOriginSession].(string); ok {
			originSessionKey = sk
		}
	}
	if originSessionKey != "" {
		meta[MetaOriginSessionKey] = originSessionKey
	}
	// Pass leader agent ID so member agents can fallback-read leader's memory.
	meta[MetaLeaderAgentID] = team.LeadAgentID.String()
	// Pass the team workspace dir so member agents write files to the shared folder.
	if ws := taskTeamWorkspace(task); ws != "" {
		meta[MetaTeamWorkspace] = ws
	}
	// Propagate trace context so member agent's trace links back to the lead's trace,
	// and the announce-back run nests under the lead's root span.
	if traceID := tracing.TraceIDFromContext(ctx); traceID != uuid.Nil {
		meta[MetaOriginTraceID] = traceID.String()
	}
	if rootSpanID := tracing.ParentSpanIDFromContext(ctx); rootSpanID != uuid.Nil {
		meta[MetaOriginRootSpanID] = rootSpanID.String()
	}

	if !m.msgBus.TryPublishInbound(bus.InboundMessage{
		Channel:  "system",
		SenderID: "teammate:dashboard",
		ChatID:   teamID.String(),
		Content:  content.String(),
		UserID:   originUserID,
		TenantID: store.TenantIDFromContext(ctx),
		AgentID:  ag.AgentKey,
		Metadata: meta,
	}) {
		slog.Warn("team_tasks.dispatch: inbound buffer full, task dispatch dropped — ticker will retry",
			"task_id", task.ID, "agent_key", ag.AgentKey)
		return
	}
	slog.Info("team_tasks.dispatch: sent task to agent",
		"task_id", task.ID,
		"agent_key", ag.AgentKey,
		"team_id", teamID,
	)
}

// buildBlockerResultsSummary fetches completed blocker tasks (stored in metadata
// during creation) and formats their results for inclusion in the dispatch content.
func (m *TeamToolManager) buildBlockerResultsSummary(ctx context.Context, task *store.TeamTaskData) string {
	if task.Metadata == nil {
		return ""
	}
	rawBlockers, ok := task.Metadata["original_blocked_by"]
	if !ok {
		return ""
	}
	blockerStrs, ok := rawBlockers.([]any)
	if !ok || len(blockerStrs) == 0 {
		return ""
	}

	var parts []string
	for _, raw := range blockerStrs {
		idStr, ok := raw.(string)
		if !ok {
			continue
		}
		blockerID, err := uuid.Parse(idStr)
		if err != nil {
			continue
		}
		bt, err := m.teamStore.GetTask(ctx, blockerID)
		if err != nil || bt == nil || bt.Result == nil {
			continue
		}
		parts = append(parts, fmt.Sprintf("--- Result from blocker task #%d (%s) ---\n%s",
			bt.TaskNumber, bt.Subject, *bt.Result))
	}
	if len(parts) == 0 {
		return ""
	}
	return "=== Completed blocker task results ===\n\n" + strings.Join(parts, "\n\n")
}

// truncatePreview returns s truncated to maxRunes runes with "..." appended.
// Uses rune slicing to avoid splitting multi-byte UTF-8 characters.
func truncatePreview(s string, maxRunes int) string {
	runes := []rune(s)
	if len(runes) <= maxRunes {
		return s
	}
	return string(runes[:maxRunes]) + "..."
}

// buildRecentCommentsSummary fetches the 3 most recent comments on a task and
// formats them for inclusion in dispatch content. This ensures re-dispatched
// tasks (after reject, retry, stale recovery) include relevant context like
// rejection reasons or progress notes without mutating the task description.
func (m *TeamToolManager) buildRecentCommentsSummary(ctx context.Context, taskID uuid.UUID) string {
	comments, err := m.teamStore.ListRecentTaskComments(ctx, taskID, 3)
	if err != nil || len(comments) == 0 {
		return ""
	}

	var parts []string
	for _, c := range comments {
		author := "system"
		if c.AgentID != nil {
			author = m.agentKeyFromID(ctx, *c.AgentID)
		} else if c.UserID != "" {
			author = c.UserID
		}
		parts = append(parts, fmt.Sprintf("- [%s]: %s", author, truncatePreview(c.Content, 500)))
	}
	return "[Recent comments]\n" + strings.Join(parts, "\n")
}

// restoreTraceContext returns a context with the leader's trace IDs restored
// from task metadata. This is needed because DispatchUnblockedTasks runs in
// the member agent's context (during executeComplete), but the dispatch should
// link back to the leader's trace, not the member's.
func (m *TeamToolManager) restoreTraceContext(ctx context.Context, task *store.TeamTaskData) context.Context {
	if task.Metadata == nil {
		return ctx
	}
	if traceIDStr, ok := task.Metadata[TaskMetaOriginTrace].(string); ok {
		if traceID, err := uuid.Parse(traceIDStr); err == nil {
			ctx = tracing.WithTraceID(ctx, traceID)
		}
	}
	if spanIDStr, ok := task.Metadata[TaskMetaOriginRootSpan].(string); ok {
		if spanID, err := uuid.Parse(spanIDStr); err == nil {
			ctx = tracing.WithParentSpanID(ctx, spanID)
		}
	}
	return ctx
}

// DispatchUnblockedTasks finds pending tasks with an assigned owner, claims the
// highest-priority task per owner (pending → in_progress), and dispatches it.
// Only one task per owner is dispatched per round — remaining tasks stay pending
// and get dispatched after the current one completes. This ensures priority
// ordering and prevents the cancellation bug where CancelSession kills innocent
// queued tasks sharing the same session.
// Called after task completion/cancellation to start newly-unblocked work
// instead of waiting for the ticker (up to 5 min delay).
func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uuid.UUID) {
	team, err := m.teamStore.GetTeam(ctx, teamID)
	if err != nil || team == nil {
		return
	}
	tasks, err := m.teamStore.ListRecoverableTasks(ctx, teamID)
	if err != nil {
		return
	}
	// Track which owners already have a task dispatched this round.
	// ListRecoverableTasks orders by priority DESC, created_at — so the first
	// pending task per owner is automatically the highest priority.
	dispatched := make(map[uuid.UUID]bool)
	for i := range tasks {
		task := &tasks[i]
		if task.Status != store.TeamTaskStatusPending || task.OwnerAgentID == nil {
			continue
		}
		ownerID := *task.OwnerAgentID
		// Auto-fail tasks assigned to the lead agent — would cause self-dispatch loop.
		// Don't just skip: pending lead-owned tasks would stay stuck until stale timeout.
		if ownerID == team.LeadAgentID {
			slog.Warn("DispatchUnblockedTasks: auto-failing lead-owned task",
				"task_id", task.ID, "team_id", teamID)
			_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{
				"status": store.TeamTaskStatusFailed,
				"result": "Cannot dispatch task to the team lead — reassign to a team member",
			})
			continue
		}
		if dispatched[ownerID] {
			continue // skip — this owner already has a higher-priority task dispatched
		}
		// Assign (pending → in_progress + lock) so consumer can auto-complete.
		if err := m.teamStore.AssignTask(ctx, task.ID, ownerID, teamID); err != nil {
			slog.Warn("DispatchUnblockedTasks: assign failed", "task_id", task.ID, "error", err)
			continue
		}
		dispatched[ownerID] = true
		taskPeerKind := ""
		if pk, ok := task.Metadata[TaskMetaPeerKind].(string); ok {
			taskPeerKind = pk
		}
		taskLocalKey := ""
		if lk, ok := task.Metadata[TaskMetaLocalKey].(string); ok {
			taskLocalKey = lk
		}
		m.broadcastTeamEvent(ctx, protocol.EventTeamTaskDispatched, BuildTaskEventPayload(
			teamID.String(), task.ID.String(),
			store.TeamTaskStatusInProgress,
			"system", "dispatch_unblocked",
			WithTaskInfo(task.TaskNumber, task.Subject),
			WithOwnerAgentKey(m.agentKeyFromID(ctx, ownerID)),
			WithChannel(task.Channel),
			WithChatID(task.ChatID),
			WithPeerKind(taskPeerKind),
			WithLocalKey(taskLocalKey),
		))

		// Append completed blocker results so the member agent has context.
		if summary := m.buildBlockerResultsSummary(ctx, task); summary != "" {
			task.Description += "\n\n" + summary
		}

		// Append recent comments so re-dispatched tasks include rejection reasons,
		// progress notes, or other context from previous runs.
		if commentCtx := m.buildRecentCommentsSummary(ctx, task.ID); commentCtx != "" {
			task.Description += "\n\n" + commentCtx
		}

		// Restore leader's trace context (stored in task metadata during creation)
		// so the member agent's trace links back to the leader, not the completing member.
		dispatchCtx := m.restoreTraceContext(ctx, task)
		m.dispatchTaskToAgent(dispatchCtx, task, team, ownerID)
	}
}