package cmd import ( "context" "log/slog" "strings" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/scheduler" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/internal/tools" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // startTaskLockRenewal extends the task lock every 5 min while the agent works. // Returns a stop func — call it when the run finishes. Returns nil if taskID is // invalid or teamStore is nil. func startTaskLockRenewal(ctx context.Context, teamStore store.TeamStore, taskID, teamID uuid.UUID) (stop func()) { if taskID == uuid.Nil || teamStore == nil { return nil } ch := make(chan struct{}) go func() { ticker := time.NewTicker(5 * time.Minute) defer ticker.Stop() for { select { case <-ticker.C: if err := teamStore.RenewTaskLock(ctx, taskID, teamID); err != nil { slog.Warn("teammate lock renewal failed", "task_id", taskID, "error", err) return } slog.Debug("teammate lock renewed", "task_id", taskID) case <-ch: return case <-ctx.Done(): return } } }() return func() { close(ch) } } // teammateTaskMeta holds resolved task context for post-turn processing. type teammateTaskMeta struct { TaskID uuid.UUID TeamID uuid.UUID ToAgent string Channel string ChatID string PeerKind string // "group" or "direct" — for correct notification routing (#266) Subject string TaskNumber int } // resolveTeamTaskOutcome determines the correct task lifecycle action // after a teammate run: error→fail, completed/escalated→skip, reviewed→renew, // loopKilled→fail, default→auto-complete. // Returns the resolved team (for announce routing) and dispatches unblocked tasks. func resolveTeamTaskOutcome( ctx context.Context, deps *ConsumerDeps, outcome scheduler.RunOutcome, flags *tools.TaskActionFlags, meta teammateTaskMeta, ) *store.TeamData { if meta.TaskID == uuid.Nil || deps.TeamStore == nil { return nil } cachedTeam, _ := deps.TeamStore.GetTeam(ctx, meta.TeamID) if cachedTeam == nil { return nil } // Check current task status — agent may have already updated it via tool. currentTask, taskErr := deps.TeamStore.GetTask(ctx, meta.TaskID) alreadyTerminal := taskErr == nil && currentTask != nil && (currentTask.Status == store.TeamTaskStatusCompleted || currentTask.Status == store.TeamTaskStatusFailed || currentTask.Status == store.TeamTaskStatusCancelled) if alreadyTerminal { // Always dispatch unblocked tasks even if already terminal. if deps.PostTurn != nil { deps.PostTurn.DispatchUnblockedTasks(ctx, meta.TeamID) } return cachedTeam } toAgent := meta.ToAgent now := time.Now().UTC().Format("2006-01-02T15:04:05Z") taskSubject := meta.Subject taskNumber := meta.TaskNumber taskChannel := meta.Channel taskChatID := meta.ChatID taskPeerKind := meta.PeerKind var taskLocalKey string // Enrich with live task data if available. if currentTask != nil { if currentTask.Subject != "" { taskSubject = currentTask.Subject } if currentTask.TaskNumber != 0 { taskNumber = currentTask.TaskNumber } if currentTask.Channel != "" { taskChannel = currentTask.Channel } if currentTask.ChatID != "" { taskChatID = currentTask.ChatID } if currentTask.Metadata != nil { if taskPeerKind == "" { if pk, ok := currentTask.Metadata[tools.TaskMetaPeerKind].(string); ok && pk != "" { taskPeerKind = pk } } // local_key is required for forum topic routing (e.g. Telegram supergroup threads). if lk, ok := currentTask.Metadata[tools.TaskMetaLocalKey].(string); ok { taskLocalKey = lk } } } // Smart post-turn decision based on action flags. // Only error, completed/escalated, and reviewed block auto-complete. runFailure := outcome.Failure() switch { case runFailure != nil: // Agent errored or the pipeline stopped the run → auto-fail. if err := deps.TeamStore.FailTask(ctx, meta.TaskID, meta.TeamID, runFailure.Error()); err != nil { slog.Warn("auto-complete: FailTask error", "task_id", meta.TaskID, "error", err) } else { bus.BroadcastForTenant(deps.MsgBus, protocol.EventTeamTaskFailed, store.TenantIDFromContext(ctx), tools.BuildTaskEventPayload( meta.TeamID.String(), meta.TaskID.String(), store.TeamTaskStatusFailed, "agent", toAgent, tools.WithTaskInfo(taskNumber, taskSubject), tools.WithReason(runFailure.Error()), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), tools.WithPeerKind(taskPeerKind), tools.WithLocalKey(taskLocalKey), tools.WithTimestamp(now), )) } case flags.Completed || flags.Escalated: // Tool already completed/failed the task — skip auto-complete. slog.Info("post-turn: tool handled task", "task_id", meta.TaskID, "completed", flags.Completed, "escalated", flags.Escalated) case flags.Reviewed: // Task submitted for review — skip auto-complete, renew lock. _ = deps.TeamStore.RenewTaskLock(ctx, meta.TaskID, meta.TeamID) slog.Info("post-turn: task submitted for review", "task_id", meta.TaskID) case outcome.Result != nil && outcome.Result.LoopKilled: // Loop detector killed the run → auto-fail instead of auto-complete. // The agent was terminated (not stuck in_progress) — mark as failed // so the leader sees clear failure signal and can retry directly. failMsg := outcome.Result.Content if failMsg == "" { failMsg = "Agent run terminated by loop detector" } if err := deps.TeamStore.FailTask(ctx, meta.TaskID, meta.TeamID, failMsg); err != nil { slog.Warn("auto-fail: FailTask error (loop kill)", "task_id", meta.TaskID, "error", err) } else { bus.BroadcastForTenant(deps.MsgBus, protocol.EventTeamTaskFailed, store.TenantIDFromContext(ctx), tools.BuildTaskEventPayload( meta.TeamID.String(), meta.TaskID.String(), store.TeamTaskStatusFailed, "system", "loop_detector", tools.WithTaskInfo(taskNumber, taskSubject), tools.WithReason("loop_detector_kill"), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), tools.WithPeerKind(taskPeerKind), tools.WithLocalKey(taskLocalKey), tools.WithTimestamp(now), )) } slog.Warn("post-turn: loop detector killed member run", "task_id", meta.TaskID, "agent", toAgent) default: // Agent turn ended without terminal action — auto-complete. // Covers: Progressed, Commented, Claimed, or no flags at all. result := "" if outcome.Result != nil { result = outcome.Result.Content if len(outcome.Result.Deliverables) > 0 { result = strings.Join(outcome.Result.Deliverables, "\n\n---\n\n") } } if result == "" { result = "Agent run ended without explicit result" } if len(result) > 100_000 { result = result[:100_000] + "\n[truncated]" } if err := deps.TeamStore.CompleteTask(ctx, meta.TaskID, meta.TeamID, result); err != nil { slog.Warn("auto-complete: CompleteTask error", "task_id", meta.TaskID, "error", err) } else { bus.BroadcastForTenant(deps.MsgBus, protocol.EventTeamTaskCompleted, store.TenantIDFromContext(ctx), tools.BuildTaskEventPayload( meta.TeamID.String(), meta.TaskID.String(), store.TeamTaskStatusCompleted, "agent", toAgent, tools.WithTaskInfo(taskNumber, taskSubject), tools.WithOwnerAgentKey(toAgent), tools.WithChannel(taskChannel), tools.WithChatID(taskChatID), tools.WithPeerKind(taskPeerKind), tools.WithLocalKey(taskLocalKey), tools.WithTimestamp(now), )) } } // Always dispatch unblocked tasks after member turn ends, // regardless of whether the task was already completed by the tool. // This ensures dependent tasks start only after the member's run finishes. if deps.PostTurn != nil { deps.PostTurn.DispatchUnblockedTasks(ctx, meta.TeamID) } return cachedTeam }