diff --git a/cmd/gateway.go b/cmd/gateway.go index 57c9a5f6..bf892adf 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -602,6 +602,7 @@ func runGateway() { AgentID: meta.LeadAgent, UserID: meta.UserID, Content: leaderContent, + Metadata: map[string]string{"run_kind": tools.RunKindNotification}, }) } else { msgBus.PublishOutbound(bus.OutboundMessage{ diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index 2c355fe9..7ebd65f7 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/cmd/gateway_consumer_handlers.go @@ -263,7 +263,12 @@ func handleTeammateMessage( taskRunSessions.Store(taskIDStr, sessionKey) } - outCh := sched.Schedule(ctx, scheduler.LaneTeam, agent.RunRequest{ + // Inject action flags into context so team_tasks tool calls record what happened. + // The post-turn goroutine reads these flags to decide auto-complete vs skip. + taskActionFlags := &tools.TaskActionFlags{} + schedCtx := tools.WithTaskActionFlags(ctx, taskActionFlags) + + outCh := sched.Schedule(schedCtx, scheduler.LaneTeam, agent.RunRequest{ SessionKey: sessionKey, Message: msg.Content, Channel: origChannel, @@ -357,7 +362,11 @@ func handleTeammateMessage( taskChatID = currentTask.ChatID } } - if outcome.Err != nil { + // Smart post-turn decision based on action flags. + // Priority: error > completed > escalated > reviewed > progress-only > no-action. + switch { + case outcome.Err != nil: + // Agent errored → auto-fail. if err := teamStore.FailTask(ctx, teamTaskID, teamID, outcome.Err.Error()); err != nil { slog.Warn("auto-complete: FailTask error", "task_id", teamTaskID, "error", err) } else { @@ -375,30 +384,51 @@ func handleTeammateMessage( ActorID: toAgent, }) } - } else 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 len(result) > 100_000 { - result = result[:100_000] + "\n[truncated]" - } - if err := teamStore.CompleteTask(ctx, teamTaskID, teamID, result); err != nil { - slog.Warn("auto-complete: CompleteTask error", "task_id", teamTaskID, "error", err) - } else { - bus.BroadcastForTenant(msgBus, protocol.EventTeamTaskCompleted, store.TenantIDFromContext(ctx), protocol.TeamTaskEventPayload{ - TeamID: teamID.String(), - TaskID: teamTaskID.String(), - TaskNumber: taskNumber, - Subject: taskSubject, - Status: store.TeamTaskStatusCompleted, - OwnerAgentKey: toAgent, - Channel: taskChannel, + + case taskActionFlags.Completed || taskActionFlags.Escalated: + // Tool already completed/failed the task — skip auto-complete. + slog.Info("post-turn: tool handled task", "task_id", teamTaskID, + "completed", taskActionFlags.Completed, "escalated", taskActionFlags.Escalated) + + case taskActionFlags.Reviewed: + // Task submitted for review — skip auto-complete, renew lock. + _ = teamStore.RenewTaskLock(ctx, teamTaskID, teamID) + slog.Info("post-turn: task submitted for review", "task_id", teamTaskID) + + case taskActionFlags.Progressed || taskActionFlags.Commented || taskActionFlags.Claimed: + // Member interacted but didn't take terminal action — renew lock. + _ = teamStore.RenewTaskLock(ctx, teamTaskID, teamID) + slog.Warn("post-turn: member did not take terminal action", + "task_id", teamTaskID, "progressed", taskActionFlags.Progressed, + "commented", taskActionFlags.Commented, "claimed", taskActionFlags.Claimed) + + default: + // No task action flags recorded — backward compat: auto-complete. + 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 len(result) > 100_000 { + result = result[:100_000] + "\n[truncated]" + } + if err := teamStore.CompleteTask(ctx, teamTaskID, teamID, result); err != nil { + slog.Warn("auto-complete: CompleteTask error", "task_id", teamTaskID, "error", err) + } else { + bus.BroadcastForTenant(msgBus, protocol.EventTeamTaskCompleted, store.TenantIDFromContext(ctx), protocol.TeamTaskEventPayload{ + TeamID: teamID.String(), + TaskID: teamTaskID.String(), + TaskNumber: taskNumber, + Subject: taskSubject, + Status: store.TeamTaskStatusCompleted, + OwnerAgentKey: toAgent, + Channel: taskChannel, ChatID: taskChatID, Timestamp: now, ActorType: "agent", ActorID: toAgent, - }) + }) + } } } } diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index 4a380628..2f1c0422 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -315,6 +315,11 @@ func processNormalMessage( ptd := tools.NewPendingTeamDispatch() schedCtx := tools.WithPendingTeamDispatch(ctx, ptd) + // Propagate run_kind from metadata (e.g. "notification" for team task status relays). + if rk := msg.Metadata["run_kind"]; rk != "" { + schedCtx = tools.WithRunKind(schedCtx, rk) + } + // Schedule through main lane (per-session concurrency controlled by maxConcurrent) outCh := sched.ScheduleWithOpts(schedCtx, "main", agent.RunRequest{ SessionKey: sessionKey, diff --git a/internal/bus/bus.go b/internal/bus/bus.go index 2c836ea7..a14a59b7 100644 --- a/internal/bus/bus.go +++ b/internal/bus/bus.go @@ -22,8 +22,8 @@ type MessageBus struct { func New() *MessageBus { return &MessageBus{ - inbound: make(chan InboundMessage, 500), - outbound: make(chan OutboundMessage, 500), + inbound: make(chan InboundMessage, 1000), + outbound: make(chan OutboundMessage, 1000), handlers: make(map[string]MessageHandler), subscribers: make(map[string]EventHandler), } diff --git a/internal/store/pg/teams_tasks_progress.go b/internal/store/pg/teams_tasks_progress.go index df0a97cb..181f476a 100644 --- a/internal/store/pg/teams_tasks_progress.go +++ b/internal/store/pg/teams_tasks_progress.go @@ -78,7 +78,7 @@ func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Recover now := time.Now() rows, err := s.db.QueryContext(ctx, `UPDATE team_tasks t - SET status = $1, owner_agent_id = NULL, locked_at = NULL, lock_expires_at = NULL, + SET status = $1, locked_at = NULL, lock_expires_at = NULL, followup_at = NULL, followup_count = 0, followup_message = NULL, followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm @@ -101,7 +101,7 @@ func (s *PGTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.Recover now := time.Now() rows, err := s.db.QueryContext(ctx, `UPDATE team_tasks t - SET status = $1, owner_agent_id = NULL, locked_at = NULL, lock_expires_at = NULL, + SET status = $1, locked_at = NULL, lock_expires_at = NULL, followup_at = NULL, followup_count = 0, followup_message = NULL, followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm @@ -162,6 +162,54 @@ func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time return scanRecoveredTaskInfoRows(rows) } +// MarkInReviewStaleTasks marks in_review tasks older than olderThan as stale across all v2 active teams. +func (s *PGTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan time.Time) ([]store.RecoveredTaskInfo, error) { + now := time.Now() + rows, err := s.db.QueryContext(ctx, + `UPDATE team_tasks t + SET status = $1, updated_at = $2 + FROM agent_teams tm + WHERE t.team_id = tm.id AND tm.status = 'active' + AND COALESCE((tm.settings->>'version')::int, 0) >= 2 + AND t.status = $3 AND t.updated_at < $4 + RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, + store.TeamTaskStatusStale, now, store.TeamTaskStatusInReview, olderThan, + ) + if err != nil { + return nil, err + } + defer rows.Close() + return scanRecoveredTaskInfoRows(rows) +} + +// FixOrphanedBlockedTasks unblocks blocked tasks where all blockers have reached terminal status. +// Safety net for cases where unblockDependentTasks() transaction rolled back. +func (s *PGTeamStore) FixOrphanedBlockedTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { + now := time.Now() + rows, err := s.db.QueryContext(ctx, + `UPDATE team_tasks t + SET blocked_by = '{}', status = $1, updated_at = $2 + FROM agent_teams tm + WHERE t.team_id = tm.id AND tm.status = 'active' + AND COALESCE((tm.settings->>'version')::int, 0) >= 2 + AND t.status = 'blocked' + AND array_length(t.blocked_by, 1) > 0 + AND NOT EXISTS ( + SELECT 1 FROM unnest(t.blocked_by) AS bid(id) + JOIN team_tasks bt ON bt.id = bid.id AND bt.tenant_id = t.tenant_id + WHERE bt.status NOT IN ($3, $4, $5) + ) + RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, + store.TeamTaskStatusPending, now, + store.TeamTaskStatusCompleted, store.TeamTaskStatusFailed, store.TeamTaskStatusCancelled, + ) + if err != nil { + return nil, err + } + defer rows.Close() + return scanRecoveredTaskInfoRows(rows) +} + func scanRecoveredTaskInfoRows(rows interface { Next() bool Scan(...any) error @@ -187,9 +235,10 @@ func (s *PGTeamStore) ResetTaskStatus(ctx context.Context, taskID, teamID uuid.U res, err := s.db.ExecContext(ctx, `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) AND tenant_id = $7`, + WHERE id = $3 AND team_id = $4 AND status IN ($5, $6, $7, $8) AND tenant_id = $9`, store.TeamTaskStatusPending, now, - taskID, teamID, store.TeamTaskStatusStale, store.TeamTaskStatusFailed, tid, + taskID, teamID, store.TeamTaskStatusStale, store.TeamTaskStatusFailed, + store.TeamTaskStatusCancelled, store.TeamTaskStatusInReview, tid, ) if err != nil { return err @@ -199,7 +248,7 @@ func (s *PGTeamStore) ResetTaskStatus(ctx context.Context, taskID, teamID uuid.U return err } if n == 0 { - return fmt.Errorf("task not available for reset (not stale/failed or wrong team)") + return fmt.Errorf("task not available for reset (not stale/failed/cancelled/in_review or wrong team)") } return nil } diff --git a/internal/store/team_store.go b/internal/store/team_store.go index c2976b3d..c64bf865 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -324,6 +324,10 @@ type TeamStore interface { ListRecoverableTasks(ctx context.Context, teamID uuid.UUID) ([]TeamTaskData, error) // MarkAllStaleTasks sets pending tasks older than olderThan to stale status across all v2 active teams. MarkAllStaleTasks(ctx context.Context, olderThan time.Time) ([]RecoveredTaskInfo, error) + // MarkInReviewStaleTasks sets in_review tasks older than olderThan to stale across all v2 active teams. + MarkInReviewStaleTasks(ctx context.Context, olderThan time.Time) ([]RecoveredTaskInfo, error) + // FixOrphanedBlockedTasks unblocks blocked tasks where all blockers reached terminal status. + FixOrphanedBlockedTasks(ctx context.Context) ([]RecoveredTaskInfo, error) // ResetTaskStatus resets a stale or failed task back to pending for retry. ResetTaskStatus(ctx context.Context, taskID, teamID uuid.UUID) error diff --git a/internal/tasks/task_ticker.go b/internal/tasks/task_ticker.go index 7665eabd..d9010e12 100644 --- a/internal/tasks/task_ticker.go +++ b/internal/tasks/task_ticker.go @@ -17,10 +17,11 @@ import ( ) const ( - defaultRecoveryInterval = 5 * time.Minute - defaultStaleThreshold = 2 * time.Hour - followupCooldown = 5 * time.Minute - defaultFollowupInterval = 30 * time.Minute + defaultRecoveryInterval = 5 * time.Minute + defaultStaleThreshold = 2 * time.Hour + defaultInReviewThreshold = 4 * time.Hour + followupCooldown = 5 * time.Minute + defaultFollowupInterval = 30 * time.Minute ) // TaskTicker periodically recovers stale tasks and re-dispatches pending work. @@ -136,7 +137,35 @@ func (t *TaskTicker) recoverAll(forceRecover bool) { t.broadcastStaleEvents(recoverCtx, stale) } - // Step 4: Prune old cooldown entries to prevent memory leak. + // Step 4: Mark in_review tasks stale after 4 hours. + inReviewThreshold := time.Now().Add(-defaultInReviewThreshold) + staleReview, err := t.teams.MarkInReviewStaleTasks(recoverCtx, inReviewThreshold) + if err != nil { + slog.Warn("task_ticker: batch mark in_review stale", "error", err) + } + if len(staleReview) > 0 { + slog.Info("task_ticker: marked in_review stale", "count", len(staleReview)) + t.notifyLeaders(recoverCtx, staleReview, "in review too long (4+ hours) — marked stale", + "These tasks have been waiting for approval too long.\n"+ + "To approve: use team_tasks(action=\"approve\", task_id=\"\").\n"+ + "To reject: use team_tasks(action=\"reject\", task_id=\"\", text=\"reason\").\n"+ + "To retry: use team_tasks(action=\"retry\", task_id=\"\").") + t.broadcastStaleEvents(recoverCtx, staleReview) + } + + // Step 5: Fix orphaned blocked tasks where all blockers are terminal. + fixed, err := t.teams.FixOrphanedBlockedTasks(recoverCtx) + if err != nil { + slog.Warn("task_ticker: fix orphaned blocked tasks", "error", err) + } + if len(fixed) > 0 { + slog.Info("task_ticker: auto-unblocked orphaned tasks", "count", len(fixed)) + t.notifyLeaders(recoverCtx, fixed, "auto-unblocked (all blockers resolved)", + "These blocked tasks were automatically unblocked because all their dependencies completed.\n"+ + "They are now pending and will be dispatched if assigned.") + } + + // Step 6: Prune old cooldown entries to prevent memory leak. t.pruneCooldowns() } diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index a6881ec7..95e96f4c 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -29,6 +29,7 @@ const ( ctxWorkspace toolContextKey = "tool_workspace" ctxAgentKey toolContextKey = "tool_agent_key" ctxSessionKey toolContextKey = "tool_session_key" // origin session key for announce routing + ctxRunKind toolContextKey = "tool_run_kind" // "notification", "announce", "delegation" ) // Well-known channel names used for routing and access control. @@ -142,6 +143,21 @@ func ToolSessionKeyFromCtx(ctx context.Context) string { return v } +// WithRunKind injects the run classification (e.g. "notification") into context. +func WithRunKind(ctx context.Context, kind string) context.Context { + return context.WithValue(ctx, ctxRunKind, kind) +} + +// RunKindFromCtx returns the run kind from context, or empty string. +func RunKindFromCtx(ctx context.Context) string { + v, _ := ctx.Value(ctxRunKind).(string) + return v +} + +// RunKindNotification is the run kind for team task notification runs. +// Leader agents in this mode can only relay status — mutations are blocked. +const RunKindNotification = "notification" + // --- Builtin tool settings (global DB overrides) --- const ctxBuiltinToolSettings toolContextKey = "tool_builtin_settings" diff --git a/internal/tools/team_task_action_flags.go b/internal/tools/team_task_action_flags.go new file mode 100644 index 00000000..6dc93621 --- /dev/null +++ b/internal/tools/team_task_action_flags.go @@ -0,0 +1,43 @@ +package tools + +import "context" + +// TaskActionFlags tracks which team_tasks actions a member agent took during a turn. +// The consumer reads these flags post-turn to decide whether to auto-complete, +// skip, or re-dispatch the task. +// +// No mutex needed: a single goroutine writes (agent loop), and the consumer +// reads only after the turn ends (sequential). +type TaskActionFlags struct { + Completed bool // team_tasks(action="complete") + Reviewed bool // team_tasks(action="review") + Escalated bool // team_tasks(action="comment", type="blocker") + Progressed bool // team_tasks(action="progress") + Commented bool // team_tasks(action="comment") + Claimed bool // team_tasks(action="claim") +} + +// HasAny returns true if the member interacted with the task system at all. +func (f *TaskActionFlags) HasAny() bool { + return f.Completed || f.Reviewed || f.Escalated || f.Progressed || f.Commented || f.Claimed +} + +const ctxTaskActionFlags toolContextKey = "tool_task_action_flags" + +// WithTaskActionFlags injects a TaskActionFlags into context. +func WithTaskActionFlags(ctx context.Context, flags *TaskActionFlags) context.Context { + return context.WithValue(ctx, ctxTaskActionFlags, flags) +} + +// TaskActionFlagsFromCtx returns the TaskActionFlags from context, or nil. +func TaskActionFlagsFromCtx(ctx context.Context) *TaskActionFlags { + v, _ := ctx.Value(ctxTaskActionFlags).(*TaskActionFlags) + return v +} + +// recordTaskAction is a helper that sets a flag if TaskActionFlags exist in context. +func recordTaskAction(ctx context.Context, setter func(*TaskActionFlags)) { + if flags := TaskActionFlagsFromCtx(ctx); flags != nil { + setter(flags) + } +} diff --git a/internal/tools/team_tasks_lifecycle.go b/internal/tools/team_tasks_lifecycle.go index f5b400d6..0eb02c20 100644 --- a/internal/tools/team_tasks_lifecycle.go +++ b/internal/tools/team_tasks_lifecycle.go @@ -3,6 +3,7 @@ package tools import ( "context" "fmt" + "log/slog" "time" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -23,6 +24,8 @@ func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]any) * if err := t.manager.teamStore.ClaimTask(ctx, taskID, agentID, team.ID); err != nil { return ErrorResult("failed to claim task: " + err.Error()) } + // Record action flag after successful store operation. + recordTaskAction(ctx, func(f *TaskActionFlags) { f.Claimed = true }) ownerKey := t.manager.agentKeyFromID(ctx, agentID) t.manager.broadcastTeamEvent(ctx, protocol.EventTeamTaskClaimed, protocol.TeamTaskEventPayload{ @@ -70,6 +73,8 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]any if err := t.manager.teamStore.CompleteTask(ctx, taskID, team.ID, result); err != nil { return ErrorResult("failed to complete task: " + err.Error()) } + // Record action flag after successful store operation. + recordTaskAction(ctx, func(f *TaskActionFlags) { f.Completed = true }) ownerKey := t.manager.agentKeyFromID(ctx, agentID) // Fetch task for TaskNumber/Subject needed by notification subscriber. @@ -173,6 +178,8 @@ func (t *TeamTasksTool) executeReview(ctx context.Context, args map[string]any) if err := t.manager.teamStore.ReviewTask(ctx, taskID, team.ID); err != nil { return ErrorResult("failed to submit for review: " + err.Error()) } + // Record action flag after successful store operation. + recordTaskAction(ctx, func(f *TaskActionFlags) { f.Reviewed = true }) ownerKey := t.manager.agentKeyFromID(ctx, agentID) t.manager.broadcastTeamEvent(ctx, protocol.EventTeamTaskReviewed, protocol.TeamTaskEventPayload{ @@ -294,16 +301,71 @@ func (t *TeamTasksTool) executeReject(ctx context.Context, args map[string]any) return ErrorResult("task does not belong to your team") } - // Reuse CancelTask (handles unblocking dependents, guards against completed) - if err := t.manager.teamStore.CancelTask(ctx, taskID, team.ID, reason); err != nil { - return ErrorResult("failed to reject task: " + err.Error()) + // Record rejection as a task comment for audit trail (before status change). + rejectMsg := fmt.Sprintf("Task rejected. Reason: %s", reason) + _ = t.manager.teamStore.AddTaskComment(ctx, &store.TeamTaskCommentData{ + TaskID: taskID, + AgentID: &agentID, + Content: rejectMsg, + }) + + // Auto re-dispatch if task has an owner: skip RejectTask (which unblocks dependents) + // and instead reset in_review → pending → in_progress → dispatch. + // Dependents stay blocked until this task actually completes or circuit breaker fails it. + if task.OwnerAgentID != nil { + // Reset in_review → pending (ResetTaskStatus accepts in_review via store guard). + if err := t.manager.teamStore.ResetTaskStatus(ctx, taskID, team.ID); err != nil { + // Fallback: use RejectTask (cancels + unblocks) if reset fails. + slog.Warn("reject: reset failed, falling back to RejectTask", "task_id", taskID, "error", err) + if err := t.manager.teamStore.RejectTask(ctx, taskID, team.ID, reason); err != nil { + return ErrorResult("failed to reject task: " + err.Error()) + } + t.manager.broadcastTeamEvent(ctx, protocol.EventTeamTaskRejected, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskID.String(), + Subject: task.Subject, + Status: store.TeamTaskStatusCancelled, + Reason: reason, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + ActorType: "agent", + ActorID: t.manager.agentKeyFromID(ctx, agentID), + }) + return NewResult(fmt.Sprintf("Task %s rejected (cancelled). Use retry to re-dispatch manually.", taskID)) + } + if err := t.manager.teamStore.AssignTask(ctx, taskID, *task.OwnerAgentID, team.ID); err != nil { + slog.Warn("reject: assign task failed", "task_id", taskID, "error", err) + return NewResult(fmt.Sprintf("Task %s rejected but could not assign. Use retry to re-dispatch manually.", taskID)) + } + t.manager.broadcastTeamEvent(ctx, protocol.EventTeamTaskRejected, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskID.String(), + Subject: task.Subject, + Status: store.TeamTaskStatusInProgress, + Reason: reason, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + ActorType: "agent", + ActorID: t.manager.agentKeyFromID(ctx, agentID), + }) + t.manager.dispatchTaskToAgent(ctx, task, team, *task.OwnerAgentID) + return NewResult(fmt.Sprintf("Task %s rejected and re-dispatched to %s with feedback.", + taskID, t.manager.agentKeyFromID(ctx, *task.OwnerAgentID))) } + // No owner — use RejectTask to cancel + unblock dependents. + if err := t.manager.teamStore.RejectTask(ctx, taskID, team.ID, reason); err != nil { + return ErrorResult("failed to reject task: " + err.Error()) + } t.manager.broadcastTeamEvent(ctx, protocol.EventTeamTaskRejected, protocol.TeamTaskEventPayload{ TeamID: team.ID.String(), TaskID: taskID.String(), Subject: task.Subject, - Status: "cancelled", + Status: store.TeamTaskStatusCancelled, Reason: reason, UserID: store.UserIDFromContext(ctx), Channel: ToolChannelFromCtx(ctx), @@ -312,14 +374,5 @@ func (t *TeamTasksTool) executeReject(ctx context.Context, args map[string]any) ActorType: "agent", ActorID: t.manager.agentKeyFromID(ctx, agentID), }) - - // Record rejection as a task comment for audit trail. - rejectMsg := fmt.Sprintf("Task rejected. Reason: %s", reason) - _ = t.manager.teamStore.AddTaskComment(ctx, &store.TeamTaskCommentData{ - TaskID: taskID, - AgentID: &agentID, - Content: rejectMsg, - }) - - return NewResult(fmt.Sprintf("Task %s rejected. Dependent tasks have been unblocked.", taskID)) + return NewResult(fmt.Sprintf("Task %s rejected.", taskID)) } diff --git a/internal/tools/team_tasks_mutations.go b/internal/tools/team_tasks_mutations.go index 9d8b38a6..26349ac3 100644 --- a/internal/tools/team_tasks_mutations.go +++ b/internal/tools/team_tasks_mutations.go @@ -343,6 +343,14 @@ func (t *TeamTasksTool) executeComment(ctx context.Context, args map[string]any) ActorID: t.manager.agentKeyFromID(ctx, agentID), }) + // Record action flag after successful store operation. + recordTaskAction(ctx, func(f *TaskActionFlags) { + f.Commented = true + if commentType == "blocker" { + f.Escalated = true + } + }) + // Blocker escalation: auto-fail task + notify leader. if commentType == "blocker" { return t.handleBlockerComment(ctx, team, task, taskID, agentID, text) @@ -404,6 +412,8 @@ func (t *TeamTasksTool) executeProgress(ctx context.Context, args map[string]any if err := t.manager.teamStore.UpdateTaskProgress(ctx, taskID, team.ID, percent, step); err != nil { return ErrorResult("failed to update progress: " + err.Error()) } + // Record action flag after successful store operation. + recordTaskAction(ctx, func(f *TaskActionFlags) { f.Progressed = true }) ownerKey := "" if task.OwnerAgentID != nil { diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index 67ec5d47..cb0c771f 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -101,6 +101,16 @@ func (t *TeamTasksTool) Parameters() map[string]any { func (t *TeamTasksTool) Execute(ctx context.Context, args map[string]any) *Result { action, _ := args["action"].(string) + // Block mutations during notification runs — leader may only relay status. + if RunKindFromCtx(ctx) == RunKindNotification { + switch action { + case "list", "get", "search": + // Read-only actions allowed. + default: + return ErrorResult("This is a notification run. Your role is to relay task status to the user in a natural, conversational style. Do not modify tasks.") + } + } + switch action { case "list": return t.executeList(ctx, args)