diff --git a/internal/gateway/methods/teams.go b/internal/gateway/methods/teams.go index 752a071c..5f31825e 100644 --- a/internal/gateway/methods/teams.go +++ b/internal/gateway/methods/teams.go @@ -44,6 +44,8 @@ func (m *TeamsMethods) Register(router *gateway.MethodRouter) { router.Register(protocol.MethodTeamsGet, m.handleGet) router.Register(protocol.MethodTeamsDelete, m.handleDelete) router.Register(protocol.MethodTeamsTaskList, m.handleTaskList) + router.Register(protocol.MethodTeamsTaskApprove, m.handleTaskApprove) + router.Register(protocol.MethodTeamsTaskReject, m.handleTaskReject) router.Register(protocol.MethodTeamsMembersAdd, m.handleAddMember) router.Register(protocol.MethodTeamsMembersRemove, m.handleRemoveMember) router.Register(protocol.MethodTeamsUpdate, m.handleUpdate) diff --git a/internal/gateway/methods/teams_tasks_approval.go b/internal/gateway/methods/teams_tasks_approval.go new file mode 100644 index 00000000..29f1f2a4 --- /dev/null +++ b/internal/gateway/methods/teams_tasks_approval.go @@ -0,0 +1,187 @@ +package methods + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/internal/i18n" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// --- Task Approve --- + +type teamsTaskApproveParams struct { + TaskID string `json:"task_id"` +} + +func (m *TeamsMethods) handleTaskApprove(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + if m.teamStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, i18n.T(locale, i18n.MsgTeamsNotConfigured))) + return + } + + var params teamsTaskApproveParams + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))) + return + } + if params.TaskID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "task_id"))) + return + } + + taskID, err := uuid.Parse(params.TaskID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "task_id"))) + return + } + + // Fetch task to get team_id, subject, and verify status + task, err := m.teamStore.GetTask(ctx, taskID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + if task.Status != store.TeamTaskStatusPendingApproval { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, + fmt.Sprintf("task is not pending approval (current status: %s)", task.Status))) + return + } + + // Atomic transition: pending_approval -> pending (or blocked if blockers exist) + if err := m.teamStore.ApproveTask(ctx, taskID, task.TeamID); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + // Re-fetch to get actual post-approval status (pending or blocked) + updated, err := m.teamStore.GetTask(ctx, taskID) + if err != nil { + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) + } else { + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"task": updated})) + } + + newStatus := store.TeamTaskStatusPending + if updated != nil { + newStatus = updated.Status + } + + // Broadcast event + if m.msgBus != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamTaskApproved, + Payload: protocol.TeamTaskEventPayload{ + TeamID: task.TeamID.String(), + TaskID: taskID.String(), + Subject: task.Subject, + Status: newStatus, + UserID: client.UserID(), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + }, + }) + } + + // Inject message to lead agent via mailbox + team, err := m.teamStore.GetTeam(ctx, task.TeamID) + if err == nil { + msg := fmt.Sprintf("Task '%s' (id=%s) has been approved by the user (status: %s).", task.Subject, task.ID, newStatus) + _ = m.teamStore.SendMessage(ctx, &store.TeamMessageData{ + TeamID: task.TeamID, + FromAgentID: team.LeadAgentID, + ToAgentID: &team.LeadAgentID, + Content: msg, + MessageType: store.TeamMessageTypeChat, + TaskID: &taskID, + }) + } +} + +// --- Task Reject --- + +type teamsTaskRejectParams struct { + TaskID string `json:"task_id"` + Reason string `json:"reason"` +} + +func (m *TeamsMethods) handleTaskReject(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + if m.teamStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, i18n.T(locale, i18n.MsgTeamsNotConfigured))) + return + } + + var params teamsTaskRejectParams + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))) + return + } + if params.TaskID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "task_id"))) + return + } + + taskID, err := uuid.Parse(params.TaskID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "task_id"))) + return + } + + reason := params.Reason + if reason == "" { + reason = "Rejected by user" + } + + // Fetch task to get team_id and subject for the lead message + task, err := m.teamStore.GetTask(ctx, taskID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + // Reuse CancelTask (handles unblocking dependents, guards against already-completed) + if err := m.teamStore.CancelTask(ctx, taskID, task.TeamID, reason); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) + + // Broadcast event + if m.msgBus != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamTaskRejected, + Payload: protocol.TeamTaskEventPayload{ + TeamID: task.TeamID.String(), + TaskID: taskID.String(), + Subject: task.Subject, + Status: "cancelled", + Reason: reason, + UserID: client.UserID(), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + }, + }) + } + + // Inject message to lead agent via mailbox + team, err := m.teamStore.GetTeam(ctx, task.TeamID) + if err == nil { + leadMsg := fmt.Sprintf("Task '%s' (id=%s) was rejected by the user. Reason: %s", task.Subject, task.ID, reason) + _ = m.teamStore.SendMessage(ctx, &store.TeamMessageData{ + TeamID: task.TeamID, + FromAgentID: team.LeadAgentID, + ToAgentID: &team.LeadAgentID, + Content: leadMsg, + MessageType: store.TeamMessageTypeChat, + TaskID: &taskID, + }) + } +} diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index 10badeeb..9e107265 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -51,7 +51,7 @@ func (s *PGTeamStore) ListTasks(ctx context.Context, teamID uuid.UUID, orderBy s orderClause = "t.created_at DESC" } - statusWhere := "AND t.status != 'completed'" // default: active only + statusWhere := "AND t.status IN ('pending', 'pending_approval', 'in_progress', 'blocked')" // default: active only switch statusFilter { case store.TeamTaskFilterAll: statusWhere = "" @@ -175,6 +175,30 @@ func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID, teamID uuid.UUID return tx.Commit() } +func (s *PGTeamStore) ApproveTask(ctx context.Context, taskID, teamID uuid.UUID) error { + // Transition pending_approval → pending or blocked (if blockers exist). + // Single atomic UPDATE: check blocked_by array to decide target status. + res, err := s.db.ExecContext(ctx, + `UPDATE team_tasks SET + status = CASE WHEN array_length(blocked_by, 1) > 0 THEN $1 ELSE $2 END, + updated_at = $3 + WHERE id = $4 AND status = $5 AND team_id = $6`, + store.TeamTaskStatusBlocked, store.TeamTaskStatusPending, time.Now(), + taskID, store.TeamTaskStatusPendingApproval, teamID, + ) + if err != nil { + return err + } + n, err := res.RowsAffected() + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("task not found, not pending approval, or wrong team") + } + return nil +} + func (s *PGTeamStore) CancelTask(ctx context.Context, taskID, teamID uuid.UUID, reason string) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { diff --git a/internal/store/team_store.go b/internal/store/team_store.go index 6b6ed50d..7e18a6bb 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -23,16 +23,17 @@ const ( // Team task status constants. const ( - TeamTaskStatusPending = "pending" - TeamTaskStatusInProgress = "in_progress" - TeamTaskStatusCompleted = "completed" - TeamTaskStatusBlocked = "blocked" - TeamTaskStatusFailed = "failed" + TeamTaskStatusPending = "pending" + TeamTaskStatusPendingApproval = "pending_approval" + TeamTaskStatusInProgress = "in_progress" + TeamTaskStatusCompleted = "completed" + TeamTaskStatusBlocked = "blocked" + TeamTaskStatusFailed = "failed" ) // Team task list filter constants (for ListTasks statusFilter parameter). const ( - TeamTaskFilterActive = "" // default: pending + in_progress + blocked + TeamTaskFilterActive = "" // default: pending + pending_approval + in_progress + blocked TeamTaskFilterCompleted = "completed" // only completed tasks TeamTaskFilterAll = "all" // all statuses ) @@ -199,6 +200,10 @@ type TeamStore interface { // teamID is validated in the WHERE clause to prevent cross-team task completion. CompleteTask(ctx context.Context, taskID, teamID uuid.UUID, result string) error + // ApproveTask atomically transitions a task from pending_approval to pending. + // Returns error if the task is not in pending_approval status or not found. + ApproveTask(ctx context.Context, taskID, teamID uuid.UUID) error + // CancelTask marks a non-completed task as cancelled (status=completed, result="CANCELLED: ..."), // unblocks dependent tasks, and transitions blocked→pending when all blockers are resolved. // Returns error if the task is already completed or not found. diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index 2edbe1f0..7f7e5224 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -25,7 +25,7 @@ func NewTeamTasksTool(manager *TeamToolManager) *TeamTasksTool { func (t *TeamTasksTool) Name() string { return "team_tasks" } func (t *TeamTasksTool) Description() string { - return "Manage the shared team task list. Actions: list (active tasks overview), get (full task detail with result), create, claim, complete, cancel, search. See TEAM.md for your team context." + return "Manage the shared team task list. Actions: list (active tasks overview), get (full task detail with result), create, claim, complete, cancel, approve, reject, search. See TEAM.md for your team context." } func (t *TeamTasksTool) Parameters() map[string]any { @@ -34,7 +34,7 @@ func (t *TeamTasksTool) Parameters() map[string]any { "properties": map[string]any{ "action": map[string]any{ "type": "string", - "description": "'list', 'get', 'create', 'claim', 'complete', 'cancel', or 'search'", + "description": "'list', 'get', 'create', 'claim', 'complete', 'cancel', 'approve', 'reject', or 'search'", }, "status": map[string]any{ "type": "string", @@ -71,7 +71,11 @@ func (t *TeamTasksTool) Parameters() map[string]any { }, "reason": map[string]any{ "type": "string", - "description": "Cancellation reason (optional for action=cancel)", + "description": "Cancellation or rejection reason (optional for action=cancel or reject)", + }, + "require_approval": map[string]any{ + "type": "boolean", + "description": "If true, task requires user approval before agents can claim it (optional for action=create, default false)", }, }, "required": []string{"action"}, @@ -94,10 +98,14 @@ func (t *TeamTasksTool) Execute(ctx context.Context, args map[string]any) *Resul return t.executeComplete(ctx, args) case "cancel": return t.executeCancel(ctx, args) + case "approve": + return t.executeApprove(ctx, args) + case "reject": + return t.executeReject(ctx, args) case "search": return t.executeSearch(ctx, args) default: - return ErrorResult(fmt.Sprintf("unknown action: %s (use list, get, create, claim, complete, cancel, or search)", action)) + return ErrorResult(fmt.Sprintf("unknown action: %s (use list, get, create, claim, complete, cancel, approve, reject, or search)", action)) } } @@ -256,8 +264,11 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any) } } + requireApproval, _ := args["require_approval"].(bool) status := store.TeamTaskStatusPending - if len(blockedBy) > 0 { + if requireApproval { + status = store.TeamTaskStatusPendingApproval + } else if len(blockedBy) > 0 { status = store.TeamTaskStatusBlocked } @@ -426,3 +437,134 @@ func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]any) return NewResult(fmt.Sprintf("Task %s cancelled. Any running delegation has been stopped and dependent tasks unblocked.", taskIDStr)) } + +func (t *TeamTasksTool) executeApprove(ctx context.Context, args map[string]any) *Result { + // Delegate agents cannot approve tasks — approval requires user authority. + if ToolChannelFromCtx(ctx) == "delegate" { + return ErrorResult("delegate agents cannot approve team tasks") + } + + team, _, err := t.manager.resolveTeam(ctx) + if err != nil { + return ErrorResult(err.Error()) + } + + taskIDStr, _ := args["task_id"].(string) + if taskIDStr == "" { + return ErrorResult("task_id is required for approve action") + } + taskID, err := uuid.Parse(taskIDStr) + if err != nil { + return ErrorResult("invalid task_id") + } + + // Fetch task for subject (used in lead message) and team ownership check + task, err := t.manager.teamStore.GetTask(ctx, taskID) + if err != nil { + return ErrorResult("task not found: " + err.Error()) + } + if task.TeamID != team.ID { + return ErrorResult("task does not belong to your team") + } + + // Atomic transition: pending_approval -> pending (or blocked if blockers exist) + if err := t.manager.teamStore.ApproveTask(ctx, taskID, team.ID); err != nil { + return ErrorResult("failed to approve task: " + err.Error()) + } + + // Re-fetch to get the actual post-approval status (pending or blocked) + approved, _ := t.manager.teamStore.GetTask(ctx, taskID) + newStatus := store.TeamTaskStatusPending + if approved != nil { + newStatus = approved.Status + } + + t.manager.broadcastTeamEvent(protocol.EventTeamTaskApproved, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskIDStr, + Subject: task.Subject, + Status: newStatus, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + }) + + // Inject message to lead agent via mailbox + msg := fmt.Sprintf("Task '%s' (id=%s) has been approved by the user (status: %s).", task.Subject, task.ID, newStatus) + _ = t.manager.teamStore.SendMessage(ctx, &store.TeamMessageData{ + TeamID: team.ID, + FromAgentID: team.LeadAgentID, + ToAgentID: &team.LeadAgentID, + Content: msg, + MessageType: store.TeamMessageTypeChat, + TaskID: &taskID, + }) + + return NewResult(fmt.Sprintf("Task %s approved (status: %s).", taskIDStr, newStatus)) +} + +func (t *TeamTasksTool) executeReject(ctx context.Context, args map[string]any) *Result { + // Delegate agents cannot reject tasks. + if ToolChannelFromCtx(ctx) == "delegate" { + return ErrorResult("delegate agents cannot reject team tasks") + } + + team, _, err := t.manager.resolveTeam(ctx) + if err != nil { + return ErrorResult(err.Error()) + } + + taskIDStr, _ := args["task_id"].(string) + if taskIDStr == "" { + return ErrorResult("task_id is required for reject action") + } + taskID, err := uuid.Parse(taskIDStr) + if err != nil { + return ErrorResult("invalid task_id") + } + + reason, _ := args["reason"].(string) + if reason == "" { + reason = "Rejected by user" + } + + // Fetch task to get subject for the lead message + task, err := t.manager.teamStore.GetTask(ctx, taskID) + if err != nil { + return ErrorResult("task not found: " + err.Error()) + } + if task.TeamID != team.ID { + 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()) + } + + t.manager.broadcastTeamEvent(protocol.EventTeamTaskRejected, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskIDStr, + Subject: task.Subject, + Status: "cancelled", + Reason: reason, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + }) + + // Inject message to lead agent via mailbox + leadMsg := fmt.Sprintf("Task '%s' (id=%s) was rejected by the user. Reason: %s", task.Subject, task.ID, reason) + _ = t.manager.teamStore.SendMessage(ctx, &store.TeamMessageData{ + TeamID: team.ID, + FromAgentID: team.LeadAgentID, + ToAgentID: &team.LeadAgentID, + Content: leadMsg, + MessageType: store.TeamMessageTypeChat, + TaskID: &taskID, + }) + + return NewResult(fmt.Sprintf("Task %s rejected. Dependent tasks have been unblocked.", taskIDStr)) +} diff --git a/pkg/protocol/events.go b/pkg/protocol/events.go index 887601d0..69dc2e17 100644 --- a/pkg/protocol/events.go +++ b/pkg/protocol/events.go @@ -43,6 +43,8 @@ const ( // Team task lifecycle events. EventTeamTaskClaimed = "team.task.claimed" EventTeamTaskCancelled = "team.task.cancelled" + EventTeamTaskApproved = "team.task.approved" + EventTeamTaskRejected = "team.task.rejected" // Team CRUD events (admin operations). EventTeamCreated = "team.created" diff --git a/pkg/protocol/methods.go b/pkg/protocol/methods.go index dc62406e..42759068 100644 --- a/pkg/protocol/methods.go +++ b/pkg/protocol/methods.go @@ -107,6 +107,8 @@ const ( MethodTeamsGet = "teams.get" MethodTeamsDelete = "teams.delete" MethodTeamsTaskList = "teams.tasks.list" + MethodTeamsTaskApprove = "teams.tasks.approve" + MethodTeamsTaskReject = "teams.tasks.reject" MethodTeamsMembersAdd = "teams.members.add" MethodTeamsMembersRemove = "teams.members.remove" MethodTeamsUpdate = "teams.update" diff --git a/pkg/protocol/team_events.go b/pkg/protocol/team_events.go index ce5b922c..8d0f36ec 100644 --- a/pkg/protocol/team_events.go +++ b/pkg/protocol/team_events.go @@ -103,7 +103,7 @@ type QualityGateRetryPayload struct { } // TeamTaskEventPayload is the typed payload for team task lifecycle events -// (created, claimed, completed, cancelled). +// (created, claimed, completed, cancelled, approved, rejected). type TeamTaskEventPayload struct { TeamID string `json:"team_id"` TaskID string `json:"task_id"`