feat(teams): add task approval workflow with approve/reject actions

Tasks can be created with require_approval flag, starting in
pending_approval status. Users approve/reject via tool actions or
WS methods. Approval respects blocked_by dependencies — tasks with
unresolved blockers transition to blocked instead of pending.
Delegate agents are restricted from approving/rejecting.
This commit is contained in:
viettranx committed 2026-03-12 21:45:28 +07:00
1 parent 853725a6d1
commit 1a4c211539
8 files changed
+377 -13

No files matched your search

+2
View File
@@ -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)
@@ -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, &params); 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, &params); 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,
})
}
}
+25 -1
View File
@@ -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 {
+11 -6
View File
@@ -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.
+147 -5
View File
@@ -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))
}
+2
View File
@@ -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"
+2
View File
@@ -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"
+1 -1
View File
@@ -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"`