mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-18 00:15:05 +00:00
feat(teams): improve task dispatch, concurrency, and tool ergonomics
- Move task dispatch from mid-turn to post-turn to prevent dependent tasks from completing before the current agent's run finishes - Add team create lock to serialize list→create flows across concurrent group chat sessions, preventing duplicate task creation - Require list-before-create gate: agents must call team_tasks(list) before creating tasks - Make assignee required on task creation - Add pagination (50 per page) to task list with offset support - Slim task list/get/search responses with dedicated structs to reduce context token usage - Add task board snapshot in announce messages to leader - Workspace: allow subdirectory paths in read/delete, show directories in list output - UI: reduce kanban card title font size for better visual balance
This commit is contained in:
1 parent
0d9906d40e
commit
8d6729e959
16 files changed
+392
-110
No files matched your search
@@ -563,10 +563,6 @@ func handleTeammateMessage(
|
||||
ActorID: toAgent,
|
||||
},
|
||||
})
|
||||
// FailTask also unblocks dependent tasks.
|
||||
if postTurn != nil {
|
||||
postTurn.DispatchUnblockedTasks(ctx, teamID)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
result := outcome.Result.Content
|
||||
@@ -591,13 +587,15 @@ func handleTeammateMessage(
|
||||
ActorID: toAgent,
|
||||
},
|
||||
})
|
||||
// Dispatch newly-unblocked dependent tasks.
|
||||
if postTurn != nil {
|
||||
postTurn.DispatchUnblockedTasks(ctx, teamID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// 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 postTurn != nil {
|
||||
postTurn.DispatchUnblockedTasks(ctx, teamID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -645,14 +643,25 @@ func handleTeammateMessage(
|
||||
}
|
||||
memberAgent := inMeta["to_agent"]
|
||||
|
||||
// Build task board snapshot scoped to this batch (same origin_trace_id).
|
||||
taskBoardSnapshot := ""
|
||||
if teamIDStr := inMeta["team_id"]; teamIDStr != "" {
|
||||
if teamUUID, err := uuid.Parse(teamIDStr); err == nil {
|
||||
taskBoardSnapshot = buildTaskBoardSnapshot(ctx, teamStore, teamUUID, inMeta["origin_chat_id"], inMeta["origin_trace_id"])
|
||||
}
|
||||
}
|
||||
|
||||
announceContent := fmt.Sprintf(
|
||||
"[System Message] Team member %q completed task.\n\nResult:\n%s\n\n"+
|
||||
"Present this result to the user. Any media files are forwarded automatically. Do NOT search for files — the result above contains all relevant information.",
|
||||
"[System Message] Team member %q completed task.\n\nResult:\n%s",
|
||||
memberAgent, outcome.Result.Content,
|
||||
)
|
||||
if taskBoardSnapshot != "" {
|
||||
announceContent += "\n\n" + taskBoardSnapshot
|
||||
}
|
||||
announceContent += "\n\nPresent this result to the user. Any media files are forwarded automatically. Do NOT search for files — the result above contains all relevant information."
|
||||
// Append team workspace path so lead can locate files without searching.
|
||||
if ws := inMeta["team_workspace"]; ws != "" {
|
||||
announceContent += fmt.Sprintf("\n[Team workspace: %s — use read_file with path relative to workspace root, e.g. read_file(path=\"teams/...\")]", ws)
|
||||
announceContent += fmt.Sprintf("\n[Team workspace: %s — use workspace_read to read files, e.g. workspace_read(file_name=\"filename.md\")]", ws)
|
||||
}
|
||||
|
||||
// Route to the lead's session on the original channel/chat.
|
||||
@@ -699,6 +708,9 @@ func handleTeammateMessage(
|
||||
announceOutCh := sched.Schedule(announceCtx, scheduler.LaneSubagent, announceReq)
|
||||
announceOutcome := <-announceOutCh
|
||||
|
||||
// Release team create lock — tasks already visible in DB, safe for other goroutines to list.
|
||||
announcePtd.ReleaseTeamLock()
|
||||
|
||||
// Post-turn: dispatch pending team tasks created during announce.
|
||||
if postTurn != nil {
|
||||
for tid, tIDs := range announcePtd.Drain() {
|
||||
@@ -845,3 +857,42 @@ func handleStopCommand(
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// buildTaskBoardSnapshot returns a formatted summary of batch task statuses
|
||||
// for inclusion in the announce message to the leader. Scoped by (teamID, chatID)
|
||||
// and filtered by origin_trace_id to show only tasks from the current batch.
|
||||
func buildTaskBoardSnapshot(ctx context.Context, teamStore store.TeamStore, teamID uuid.UUID, chatID, originTraceID string) string {
|
||||
if teamStore == nil || originTraceID == "" {
|
||||
return ""
|
||||
}
|
||||
allTasks, err := teamStore.ListTasks(ctx, teamID, "", store.TeamTaskFilterAll, "", "", chatID, 0)
|
||||
if err != nil || len(allTasks) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Filter to current batch by origin_trace_id stored in task metadata.
|
||||
var active, completed int
|
||||
var activeLines []string
|
||||
for _, t := range allTasks {
|
||||
tid, _ := t.Metadata["origin_trace_id"].(string)
|
||||
if tid != originTraceID {
|
||||
continue
|
||||
}
|
||||
switch t.Status {
|
||||
case store.TeamTaskStatusCompleted, store.TeamTaskStatusCancelled, store.TeamTaskStatusFailed:
|
||||
completed++
|
||||
default:
|
||||
active++
|
||||
activeLines = append(activeLines, fmt.Sprintf(" #%d %s — %s", t.TaskNumber, t.Subject, t.Status))
|
||||
}
|
||||
}
|
||||
total := active + completed
|
||||
if total == 0 {
|
||||
return ""
|
||||
}
|
||||
if active == 0 {
|
||||
return fmt.Sprintf("=== Task board (this batch) ===\nAll %d tasks completed.", total)
|
||||
}
|
||||
return fmt.Sprintf("=== Task board (this batch) ===\nTask progress: %d/%d completed, %d active:\n%s",
|
||||
completed, total, active, strings.Join(activeLines, "\n"))
|
||||
}
|
||||
@@ -333,6 +333,9 @@ func processNormalMessage(
|
||||
go func(agentKey, channel, chatID, session, rID string, meta map[string]string, blockReplyEnabled bool, ptd *tools.PendingTeamDispatch) {
|
||||
outcome := <-outCh
|
||||
|
||||
// Release team create lock — tasks already visible in DB, other goroutines can list.
|
||||
ptd.ReleaseTeamLock()
|
||||
|
||||
// Post-turn: dispatch pending team tasks created during this turn.
|
||||
if postTurn != nil {
|
||||
for teamID, taskIDs := range ptd.Drain() {
|
||||
|
||||
@@ -331,7 +331,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
|
||||
if recovered, err := l.teamStore.RecoverStaleTasks(ctx, team.ID); err == nil && recovered > 0 {
|
||||
slog.Info("recovered stale tasks", "team", team.ID, "count", recovered)
|
||||
}
|
||||
if tasks, err := l.teamStore.ListTasks(ctx, team.ID, "newest", "", req.UserID, "", ""); err == nil {
|
||||
if tasks, err := l.teamStore.ListTasks(ctx, team.ID, "newest", "", req.UserID, "", "", 0); err == nil {
|
||||
var stale []string
|
||||
var inProgress []string
|
||||
for _, t := range tasks {
|
||||
@@ -608,7 +608,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
|
||||
if teamTaskCreates > teamTaskSpawns && !teamTaskRetried {
|
||||
if l.teamStore != nil && l.agentUUID != uuid.Nil {
|
||||
if team, _ := l.teamStore.GetTeamForAgent(ctx, l.agentUUID); team != nil && tools.IsTeamV2(team) {
|
||||
if tasks, err := l.teamStore.ListTasks(ctx, team.ID, "newest", "", req.UserID, "", ""); err == nil {
|
||||
if tasks, err := l.teamStore.ListTasks(ctx, team.ID, "newest", "", req.UserID, "", "", 0); err == nil {
|
||||
var pendingIDs []string
|
||||
for _, t := range tasks {
|
||||
if t.Status == store.TeamTaskStatusPending {
|
||||
|
||||
@@ -83,6 +83,7 @@ func buildTeamMD(team *store.TeamData, members []store.TeamMemberData, selfID uu
|
||||
sb.WriteString("Do NOT use `spawn` for team delegation — `spawn` is only for self-clone subagent work.\n\n")
|
||||
sb.WriteString("Rules:\n")
|
||||
sb.WriteString("- Always specify `assignee` — match member expertise from the list above\n")
|
||||
sb.WriteString("- **Check task board first** — ALWAYS call `team_tasks(action=\"list\")` before creating tasks. The system blocks creation if you skip this step\n")
|
||||
sb.WriteString("- Create all tasks first, then briefly tell the user what you delegated\n")
|
||||
sb.WriteString("- Do NOT add confirmations (\"Done!\", \"Got it!\") — just state what was assigned\n")
|
||||
sb.WriteString("- Results arrive automatically — do NOT present partial results\n")
|
||||
|
||||
@@ -81,7 +81,7 @@ func (c *Channel) handleTasksList(ctx context.Context, chatID int64, isGroup boo
|
||||
return
|
||||
}
|
||||
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "")
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "", 0)
|
||||
if err != nil {
|
||||
slog.Warn("tasks command: ListTasks failed", "error", err)
|
||||
send("Failed to list tasks. Please try again.")
|
||||
@@ -173,7 +173,7 @@ func (c *Channel) handleTaskDetail(ctx context.Context, chatID int64, text strin
|
||||
return
|
||||
}
|
||||
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "")
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "", 0)
|
||||
if err != nil {
|
||||
slog.Warn("task_detail command: ListTasks failed", "error", err)
|
||||
send("Failed to list tasks. Please try again.")
|
||||
@@ -236,7 +236,7 @@ func (c *Channel) handleCallbackQuery(ctx context.Context, query *telego.Callbac
|
||||
return
|
||||
}
|
||||
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "")
|
||||
tasks, err := c.teamStore.ListTasks(ctx, team.ID, "newest", store.TeamTaskFilterAll, taskUserID(c.Name(), chatID, isGroup), "", "", 0)
|
||||
if err != nil {
|
||||
send("Failed to list tasks.")
|
||||
return
|
||||
|
||||
@@ -156,7 +156,7 @@ func (m *TeamsMethods) handleTaskList(ctx context.Context, client *gateway.Clien
|
||||
return
|
||||
}
|
||||
|
||||
tasks, err := m.teamStore.ListTasks(ctx, teamID, "newest", params.Status, "", params.Channel, params.ChatID)
|
||||
tasks, err := m.teamStore.ListTasks(ctx, teamID, "newest", params.Status, "", params.Channel, params.ChatID, 0)
|
||||
if err != nil {
|
||||
slog.Warn("teams.tasks.list failed", "team_id", teamID, "status_filter", params.Status, "error", err)
|
||||
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error()))
|
||||
|
||||
@@ -163,7 +163,7 @@ func (s *PGTeamStore) UpdateTask(ctx context.Context, taskID uuid.UUID, updates
|
||||
return execMapUpdate(ctx, s.db, "team_tasks", taskID, updates)
|
||||
}
|
||||
|
||||
func (s *PGTeamStore) ListTasks(ctx context.Context, teamID uuid.UUID, orderBy string, statusFilter string, userID string, channel string, chatID string) ([]store.TeamTaskData, error) {
|
||||
func (s *PGTeamStore) ListTasks(ctx context.Context, teamID uuid.UUID, orderBy string, statusFilter string, userID string, channel string, chatID string, offset int) ([]store.TeamTaskData, error) {
|
||||
orderClause := "t.priority DESC, t.created_at"
|
||||
if orderBy == "newest" {
|
||||
orderClause = "t.created_at DESC"
|
||||
@@ -187,7 +187,7 @@ func (s *PGTeamStore) ListTasks(ctx context.Context, teamID uuid.UUID, orderBy s
|
||||
`+taskJoinClause+`
|
||||
WHERE t.team_id = $1 AND ($2 = '' OR t.user_id = $2) `+statusWhere+` `+scopeWhere+`
|
||||
ORDER BY `+orderClause+`
|
||||
LIMIT $3`, teamID, userID, maxListTasksRows, channel, chatID)
|
||||
LIMIT $3 OFFSET $6`, teamID, userID, maxListTasksRows+1, channel, chatID, offset)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -320,7 +320,7 @@ type TeamStore interface {
|
||||
// statusFilter: "" = non-completed (default), "completed", "all".
|
||||
// userID: if non-empty, filter to tasks created by this user.
|
||||
// channel+chatID: if either is non-empty, filter to that exact scope.
|
||||
ListTasks(ctx context.Context, teamID uuid.UUID, orderBy string, statusFilter string, userID string, channel string, chatID string) ([]TeamTaskData, error)
|
||||
ListTasks(ctx context.Context, teamID uuid.UUID, orderBy string, statusFilter string, userID string, channel string, chatID string, offset int) ([]TeamTaskData, error)
|
||||
// GetTask returns a single task by ID with joined agent info.
|
||||
GetTask(ctx context.Context, taskID uuid.UUID) (*TeamTaskData, error)
|
||||
// GetTasksByIDs returns multiple tasks by IDs in a single query.
|
||||
|
||||
@@ -286,8 +286,10 @@ const ctxPendingDispatch toolContextKey = "tool_pending_team_dispatch"
|
||||
// After the turn ends, the consumer drains and dispatches them.
|
||||
// Thread-safe: tools may execute in parallel goroutines.
|
||||
type PendingTeamDispatch struct {
|
||||
mu sync.Mutex
|
||||
tasks map[uuid.UUID][]uuid.UUID // teamID → []taskID
|
||||
mu sync.Mutex
|
||||
tasks map[uuid.UUID][]uuid.UUID // teamID → []taskID
|
||||
listed bool // true after list called in this turn
|
||||
teamLock *sync.Mutex // acquired on list, released before post-turn dispatch
|
||||
}
|
||||
|
||||
func NewPendingTeamDispatch() *PendingTeamDispatch {
|
||||
@@ -310,6 +312,37 @@ func (p *PendingTeamDispatch) Drain() map[uuid.UUID][]uuid.UUID {
|
||||
return out
|
||||
}
|
||||
|
||||
// MarkListed records that list was called in this turn.
|
||||
func (p *PendingTeamDispatch) MarkListed() {
|
||||
p.mu.Lock()
|
||||
p.listed = true
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
// HasListed reports whether list was called in this turn.
|
||||
func (p *PendingTeamDispatch) HasListed() bool {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.listed
|
||||
}
|
||||
|
||||
// SetTeamLock stores the acquired team create lock so it can be released post-turn.
|
||||
func (p *PendingTeamDispatch) SetTeamLock(m *sync.Mutex) {
|
||||
p.mu.Lock()
|
||||
p.teamLock = m
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
// ReleaseTeamLock releases the held team create lock, if any.
|
||||
func (p *PendingTeamDispatch) ReleaseTeamLock() {
|
||||
p.mu.Lock()
|
||||
if p.teamLock != nil {
|
||||
p.teamLock.Unlock()
|
||||
p.teamLock = nil
|
||||
}
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
func WithPendingTeamDispatch(ctx context.Context, ptd *PendingTeamDispatch) context.Context {
|
||||
return context.WithValue(ctx, ctxPendingDispatch, ptd)
|
||||
}
|
||||
|
||||
@@ -87,10 +87,11 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]any
|
||||
ActorID: ownerKey,
|
||||
})
|
||||
|
||||
// Immediately dispatch any newly-unblocked tasks.
|
||||
t.manager.DispatchUnblockedTasks(ctx, team.ID)
|
||||
// Dependent tasks are dispatched by the consumer after this agent's turn ends
|
||||
// (post-turn), not mid-turn. This prevents dependent tasks from completing and
|
||||
// announcing to the leader before this agent's own run finishes.
|
||||
|
||||
return NewResult(fmt.Sprintf("Task %s completed. Dependent tasks have been unblocked.", taskID))
|
||||
return NewResult(fmt.Sprintf("Task %s completed. Dependent tasks will be dispatched after this turn ends.", taskID))
|
||||
}
|
||||
|
||||
func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]any) *Result {
|
||||
@@ -135,10 +136,9 @@ func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]any)
|
||||
ActorID: t.manager.agentKeyFromID(ctx, agentID),
|
||||
})
|
||||
|
||||
// Immediately dispatch any newly-unblocked tasks.
|
||||
t.manager.DispatchUnblockedTasks(ctx, team.ID)
|
||||
// Dependent tasks are dispatched by the consumer after this agent's turn ends (post-turn).
|
||||
|
||||
return NewResult(fmt.Sprintf("Task %s cancelled. Any running delegation has been stopped and dependent tasks unblocked.", taskID))
|
||||
return NewResult(fmt.Sprintf("Task %s cancelled. Dependent tasks will be unblocked after this turn ends.", taskID))
|
||||
}
|
||||
|
||||
func (t *TeamTasksTool) executeReview(ctx context.Context, args map[string]any) *Result {
|
||||
|
||||
@@ -22,6 +22,11 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
|
||||
return ErrorResult(err.Error())
|
||||
}
|
||||
|
||||
// Gate: must list tasks before creating to prevent duplicates in concurrent group chat.
|
||||
if ptd := PendingTeamDispatchFromCtx(ctx); ptd != nil && !ptd.HasListed() {
|
||||
return ErrorResult("You must check existing tasks first. Call team_tasks(action=\"list\") to review the current task board before creating new tasks — this prevents duplicates in concurrent sessions.")
|
||||
}
|
||||
|
||||
subject, _ := args["subject"].(string)
|
||||
if subject == "" {
|
||||
return ErrorResult("subject is required for create action")
|
||||
@@ -65,29 +70,29 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve optional assignee (agent key → UUID). Must be a team member.
|
||||
var assigneeID uuid.UUID
|
||||
if assigneeKey, _ := args["assignee"].(string); assigneeKey != "" {
|
||||
aid, err := t.manager.resolveAgentByKey(assigneeKey)
|
||||
if err != nil {
|
||||
return ErrorResult(fmt.Sprintf("assignee %q not found: %v", assigneeKey, err))
|
||||
// Resolve assignee (agent key → UUID). Required — every task must be assigned.
|
||||
assigneeKey, _ := args["assignee"].(string)
|
||||
if assigneeKey == "" {
|
||||
return ErrorResult("assignee is required — specify which team member should handle this task")
|
||||
}
|
||||
assigneeID, err := t.manager.resolveAgentByKey(assigneeKey)
|
||||
if err != nil {
|
||||
return ErrorResult(fmt.Sprintf("assignee %q not found: %v", assigneeKey, err))
|
||||
}
|
||||
// Verify assignee is a member of this team.
|
||||
members, err := t.manager.cachedListMembers(ctx, team.ID, agentID)
|
||||
if err != nil {
|
||||
return ErrorResult("failed to verify team membership: " + err.Error())
|
||||
}
|
||||
isMember := false
|
||||
for _, m := range members {
|
||||
if m.AgentID == assigneeID {
|
||||
isMember = true
|
||||
break
|
||||
}
|
||||
// Verify assignee is a member of this team.
|
||||
members, err := t.manager.cachedListMembers(ctx, team.ID, agentID)
|
||||
if err != nil {
|
||||
return ErrorResult("failed to verify team membership: " + err.Error())
|
||||
}
|
||||
isMember := false
|
||||
for _, m := range members {
|
||||
if m.AgentID == aid {
|
||||
isMember = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !isMember {
|
||||
return ErrorResult(fmt.Sprintf("agent %q is not a member of this team", assigneeKey))
|
||||
}
|
||||
assigneeID = aid
|
||||
}
|
||||
if !isMember {
|
||||
return ErrorResult(fmt.Sprintf("agent %q is not a member of this team", assigneeKey))
|
||||
}
|
||||
|
||||
requireApproval, _ := args["require_approval"].(bool)
|
||||
@@ -147,9 +152,7 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
|
||||
ChatID: chatID,
|
||||
Metadata: taskMeta,
|
||||
}
|
||||
if assigneeID != uuid.Nil {
|
||||
task.OwnerAgentID = &assigneeID
|
||||
}
|
||||
task.OwnerAgentID = &assigneeID
|
||||
|
||||
if err := t.manager.teamStore.CreateTask(ctx, task); err != nil {
|
||||
return ErrorResult("failed to create task: " + err.Error())
|
||||
@@ -168,23 +171,21 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
|
||||
ActorType: "agent",
|
||||
ActorID: agentKey,
|
||||
})
|
||||
if assigneeID != uuid.Nil {
|
||||
t.manager.broadcastTeamEvent(protocol.EventTeamTaskAssigned, protocol.TeamTaskEventPayload{
|
||||
TeamID: team.ID.String(),
|
||||
TaskID: task.ID.String(),
|
||||
Status: status,
|
||||
OwnerAgentKey: t.manager.agentKeyFromID(ctx, assigneeID),
|
||||
UserID: store.UserIDFromContext(ctx),
|
||||
Channel: ToolChannelFromCtx(ctx),
|
||||
ChatID: chatID,
|
||||
Timestamp: task.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"),
|
||||
ActorType: "agent",
|
||||
ActorID: agentKey,
|
||||
})
|
||||
}
|
||||
t.manager.broadcastTeamEvent(protocol.EventTeamTaskAssigned, protocol.TeamTaskEventPayload{
|
||||
TeamID: team.ID.String(),
|
||||
TaskID: task.ID.String(),
|
||||
Status: status,
|
||||
OwnerAgentKey: t.manager.agentKeyFromID(ctx, assigneeID),
|
||||
UserID: store.UserIDFromContext(ctx),
|
||||
Channel: ToolChannelFromCtx(ctx),
|
||||
ChatID: chatID,
|
||||
Timestamp: task.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"),
|
||||
ActorType: "agent",
|
||||
ActorID: agentKey,
|
||||
})
|
||||
|
||||
// Track for post-turn dispatch. If no post-turn hook (e.g. HTTP API), dispatch immediately.
|
||||
if assigneeID != uuid.Nil && status == store.TeamTaskStatusPending {
|
||||
if status == store.TeamTaskStatusPending {
|
||||
if ptd := PendingTeamDispatchFromCtx(ctx); ptd != nil {
|
||||
ptd.Add(team.ID, task.ID)
|
||||
} else {
|
||||
|
||||
@@ -4,13 +4,145 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
const listTasksLimit = 20
|
||||
const listPageSize = 50
|
||||
|
||||
// blockerSummary is a compact view of a blocker task for blocked_by resolution.
|
||||
type blockerSummary struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
Subject string `json:"subject"`
|
||||
Status string `json:"status"`
|
||||
OwnerAgentKey string `json:"owner_agent_key,omitempty"`
|
||||
}
|
||||
|
||||
// taskListItem is the slim view returned by list/search actions.
|
||||
type taskListItem struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
TaskNumber int `json:"task_number"`
|
||||
Identifier string `json:"identifier"`
|
||||
Subject string `json:"subject"`
|
||||
Status string `json:"status"`
|
||||
OwnerAgentID *uuid.UUID `json:"owner_agent_id,omitempty"`
|
||||
OwnerAgentKey string `json:"owner_agent_key,omitempty"`
|
||||
CreatedByAgentID *uuid.UUID `json:"created_by_agent_id,omitempty"`
|
||||
CreatedByAgentKey string `json:"created_by_agent_key,omitempty"`
|
||||
ProgressPercent int `json:"progress_percent,omitempty"`
|
||||
ProgressStep string `json:"progress_step,omitempty"`
|
||||
BlockedBy []blockerSummary `json:"blocked_by,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// taskDetailItem is the slim view returned by the get action.
|
||||
type taskDetailItem struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
TaskNumber int `json:"task_number"`
|
||||
Identifier string `json:"identifier"`
|
||||
Subject string `json:"subject"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Status string `json:"status"`
|
||||
Result *string `json:"result,omitempty"`
|
||||
OwnerAgentID *uuid.UUID `json:"owner_agent_id,omitempty"`
|
||||
OwnerAgentKey string `json:"owner_agent_key,omitempty"`
|
||||
CreatedByAgentID *uuid.UUID `json:"created_by_agent_id,omitempty"`
|
||||
CreatedByAgentKey string `json:"created_by_agent_key,omitempty"`
|
||||
ProgressPercent int `json:"progress_percent,omitempty"`
|
||||
ProgressStep string `json:"progress_step,omitempty"`
|
||||
BlockedBy []blockerSummary `json:"blocked_by,omitempty"`
|
||||
Priority int `json:"priority"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// slimComment is the slim comment view for get response.
|
||||
type slimComment struct {
|
||||
AgentKey string `json:"agent_key"`
|
||||
Content string `json:"content"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// slimEvent is the slim event view for get response.
|
||||
type slimEvent struct {
|
||||
EventType string `json:"event_type"`
|
||||
ActorID string `json:"actor_id"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// teamCreateLocks serializes list→create flows per (teamID:chatID) pair.
|
||||
var teamCreateLocks sync.Map // key: "teamID:chatID" → *sync.Mutex
|
||||
|
||||
func getTeamCreateLock(teamID, chatID string) *sync.Mutex {
|
||||
key := teamID + ":" + chatID
|
||||
v, _ := teamCreateLocks.LoadOrStore(key, &sync.Mutex{})
|
||||
return v.(*sync.Mutex)
|
||||
}
|
||||
|
||||
// resolveBlockers batch-loads blocker tasks and returns slim summaries.
|
||||
func (t *TeamTasksTool) resolveBlockers(ctx context.Context, blockedBy []uuid.UUID) []blockerSummary {
|
||||
if len(blockedBy) == 0 {
|
||||
return nil
|
||||
}
|
||||
tasks, err := t.manager.teamStore.GetTasksByIDs(ctx, blockedBy)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]blockerSummary, 0, len(tasks))
|
||||
for _, task := range tasks {
|
||||
out = append(out, blockerSummary{
|
||||
ID: task.ID,
|
||||
Subject: task.Subject,
|
||||
Status: task.Status,
|
||||
OwnerAgentKey: task.OwnerAgentKey,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (t *TeamTasksTool) toListItem(ctx context.Context, task store.TeamTaskData) taskListItem {
|
||||
return taskListItem{
|
||||
ID: task.ID,
|
||||
TaskNumber: task.TaskNumber,
|
||||
Identifier: task.Identifier,
|
||||
Subject: task.Subject,
|
||||
Status: task.Status,
|
||||
OwnerAgentID: task.OwnerAgentID,
|
||||
OwnerAgentKey: task.OwnerAgentKey,
|
||||
CreatedByAgentID: task.CreatedByAgentID,
|
||||
CreatedByAgentKey: task.CreatedByAgentKey,
|
||||
ProgressPercent: task.ProgressPercent,
|
||||
ProgressStep: task.ProgressStep,
|
||||
BlockedBy: t.resolveBlockers(ctx, task.BlockedBy),
|
||||
CreatedAt: task.CreatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *TeamTasksTool) toDetailItem(ctx context.Context, task *store.TeamTaskData) taskDetailItem {
|
||||
return taskDetailItem{
|
||||
ID: task.ID,
|
||||
TaskNumber: task.TaskNumber,
|
||||
Identifier: task.Identifier,
|
||||
Subject: task.Subject,
|
||||
Description: task.Description,
|
||||
Status: task.Status,
|
||||
Result: task.Result,
|
||||
OwnerAgentID: task.OwnerAgentID,
|
||||
OwnerAgentKey: task.OwnerAgentKey,
|
||||
CreatedByAgentID: task.CreatedByAgentID,
|
||||
CreatedByAgentKey: task.CreatedByAgentKey,
|
||||
ProgressPercent: task.ProgressPercent,
|
||||
ProgressStep: task.ProgressStep,
|
||||
BlockedBy: t.resolveBlockers(ctx, task.BlockedBy),
|
||||
Priority: task.Priority,
|
||||
CreatedAt: task.CreatedAt,
|
||||
UpdatedAt: task.UpdatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *TeamTasksTool) executeList(ctx context.Context, args map[string]any) *Result {
|
||||
team, _, err := t.manager.resolveTeam(ctx)
|
||||
@@ -20,34 +152,49 @@ func (t *TeamTasksTool) executeList(ctx context.Context, args map[string]any) *R
|
||||
|
||||
statusFilter, _ := args["status"].(string)
|
||||
|
||||
page := 1
|
||||
if p, ok := args["page"].(float64); ok && int(p) > 1 {
|
||||
page = int(p)
|
||||
}
|
||||
offset := (page - 1) * listPageSize
|
||||
|
||||
// Delegate/system channels see all tasks; end users only see their own.
|
||||
filterUserID := ""
|
||||
channel := ToolChannelFromCtx(ctx)
|
||||
if channel != ChannelDelegate && channel != ChannelSystem {
|
||||
filterUserID = store.UserIDFromContext(ctx)
|
||||
}
|
||||
chatID := ToolChatIDFromCtx(ctx)
|
||||
|
||||
tasks, err := t.manager.teamStore.ListTasks(ctx, team.ID, "priority", statusFilter, filterUserID, "", "")
|
||||
// Acquire team create lock to serialize list→create flows across concurrent goroutines.
|
||||
if ptd := PendingTeamDispatchFromCtx(ctx); ptd != nil && !ptd.HasListed() {
|
||||
lock := getTeamCreateLock(team.ID.String(), chatID)
|
||||
lock.Lock()
|
||||
ptd.SetTeamLock(lock)
|
||||
ptd.MarkListed()
|
||||
}
|
||||
|
||||
tasks, err := t.manager.teamStore.ListTasks(ctx, team.ID, "priority", statusFilter, filterUserID, "", chatID, offset)
|
||||
if err != nil {
|
||||
return ErrorResult("failed to list tasks: " + err.Error())
|
||||
}
|
||||
|
||||
// Strip results from list view — use action=get for full detail
|
||||
for i := range tasks {
|
||||
tasks[i].Result = nil
|
||||
hasMore := len(tasks) > listPageSize
|
||||
if hasMore {
|
||||
tasks = tasks[:listPageSize]
|
||||
}
|
||||
|
||||
hasMore := len(tasks) > listTasksLimit
|
||||
if hasMore {
|
||||
tasks = tasks[:listTasksLimit]
|
||||
items := make([]taskListItem, 0, len(tasks))
|
||||
for _, task := range tasks {
|
||||
items = append(items, t.toListItem(ctx, task))
|
||||
}
|
||||
|
||||
resp := map[string]any{
|
||||
"tasks": tasks,
|
||||
"count": len(tasks),
|
||||
"tasks": items,
|
||||
"count": len(items),
|
||||
"page": page,
|
||||
}
|
||||
if hasMore {
|
||||
resp["note"] = fmt.Sprintf("Showing first %d tasks. Use action=search with a query to find older tasks.", listTasksLimit)
|
||||
resp["has_more"] = true
|
||||
}
|
||||
|
||||
@@ -100,7 +247,7 @@ func (t *TeamTasksTool) executeGet(ctx context.Context, args map[string]any) *Re
|
||||
return ErrorResult("task does not belong to your team")
|
||||
}
|
||||
|
||||
// Truncate result for context protection (full result in DB)
|
||||
// Truncate result for context protection
|
||||
const maxResultRunes = 8000
|
||||
if task.Result != nil {
|
||||
r := []rune(*task.Result)
|
||||
@@ -110,21 +257,40 @@ func (t *TeamTasksTool) executeGet(ctx context.Context, args map[string]any) *Re
|
||||
}
|
||||
}
|
||||
|
||||
// Load comments, events, and attachments for full detail view.
|
||||
comments, _ := t.manager.teamStore.ListTaskComments(ctx, taskID)
|
||||
events, _ := t.manager.teamStore.ListTaskEvents(ctx, taskID)
|
||||
attachments, _ := t.manager.teamStore.ListTaskAttachments(ctx, taskID)
|
||||
detail := t.toDetailItem(ctx, task)
|
||||
|
||||
resp := map[string]any{
|
||||
"task": task,
|
||||
// Load and slim comments/events/attachments
|
||||
resp := map[string]any{"task": detail}
|
||||
|
||||
if comments, _ := t.manager.teamStore.ListTaskComments(ctx, taskID); len(comments) > 0 {
|
||||
slim := make([]slimComment, 0, len(comments))
|
||||
for _, c := range comments {
|
||||
key := ""
|
||||
if c.AgentID != nil {
|
||||
key = t.manager.agentKeyFromID(ctx, *c.AgentID)
|
||||
}
|
||||
slim = append(slim, slimComment{
|
||||
AgentKey: key,
|
||||
Content: c.Content,
|
||||
CreatedAt: c.CreatedAt,
|
||||
})
|
||||
}
|
||||
resp["comments"] = slim
|
||||
}
|
||||
if len(comments) > 0 {
|
||||
resp["comments"] = comments
|
||||
|
||||
if events, _ := t.manager.teamStore.ListTaskEvents(ctx, taskID); len(events) > 0 {
|
||||
slim := make([]slimEvent, 0, len(events))
|
||||
for _, e := range events {
|
||||
slim = append(slim, slimEvent{
|
||||
EventType: e.EventType,
|
||||
ActorID: e.ActorID,
|
||||
CreatedAt: e.CreatedAt,
|
||||
})
|
||||
}
|
||||
resp["events"] = slim
|
||||
}
|
||||
if len(events) > 0 {
|
||||
resp["events"] = events
|
||||
}
|
||||
if len(attachments) > 0 {
|
||||
|
||||
if attachments, _ := t.manager.teamStore.ListTaskAttachments(ctx, taskID); len(attachments) > 0 {
|
||||
resp["attachments"] = attachments
|
||||
}
|
||||
|
||||
@@ -150,26 +316,19 @@ func (t *TeamTasksTool) executeSearch(ctx context.Context, args map[string]any)
|
||||
filterUserID = store.UserIDFromContext(ctx)
|
||||
}
|
||||
|
||||
tasks, err := t.manager.teamStore.SearchTasks(ctx, team.ID, query, 20, filterUserID)
|
||||
tasks, err := t.manager.teamStore.SearchTasks(ctx, team.ID, query, listPageSize, filterUserID)
|
||||
if err != nil {
|
||||
return ErrorResult("failed to search tasks: " + err.Error())
|
||||
}
|
||||
|
||||
// Show result snippets in search results
|
||||
const maxSnippetRunes = 500
|
||||
for i := range tasks {
|
||||
if tasks[i].Result != nil {
|
||||
r := []rune(*tasks[i].Result)
|
||||
if len(r) > maxSnippetRunes {
|
||||
s := string(r[:maxSnippetRunes]) + "..."
|
||||
tasks[i].Result = &s
|
||||
}
|
||||
}
|
||||
items := make([]taskListItem, 0, len(tasks))
|
||||
for _, task := range tasks {
|
||||
items = append(items, t.toListItem(ctx, task))
|
||||
}
|
||||
|
||||
out, _ := json.Marshal(map[string]any{
|
||||
"tasks": tasks,
|
||||
"count": len(tasks),
|
||||
"tasks": items,
|
||||
"count": len(items),
|
||||
})
|
||||
return SilentResult(string(out))
|
||||
}
|
||||
@@ -83,7 +83,11 @@ func (t *TeamTasksTool) Parameters() map[string]any {
|
||||
},
|
||||
"assignee": map[string]any{
|
||||
"type": "string",
|
||||
"description": "Agent key to assign task to (for create). Auto-dispatches to that team member.",
|
||||
"description": "Agent key to assign task to (REQUIRED for create). Auto-dispatches to that team member.",
|
||||
},
|
||||
"page": map[string]any{
|
||||
"type": "number",
|
||||
"description": "Page number for list/search (default 1, 50 per page)",
|
||||
},
|
||||
},
|
||||
"required": []string{"action"},
|
||||
|
||||
@@ -52,6 +52,34 @@ func ResolveWorkspacePath(dataDir, path string) string {
|
||||
// validFileName matches alphanumeric + "-_." only.
|
||||
var validFileName = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]{0,253}[a-zA-Z0-9._]$`)
|
||||
|
||||
// sanitizeFilePath validates a relative file path for workspace read/delete.
|
||||
// Allows subdirectory access (e.g. "generated/2026-03-16/image.png") while
|
||||
// preventing path traversal and absolute paths.
|
||||
func sanitizeFilePath(name string) (string, error) {
|
||||
if name == "" {
|
||||
return "", fmt.Errorf("file_name is required")
|
||||
}
|
||||
if len(name) > 512 {
|
||||
return "", fmt.Errorf("file_name exceeds 512 characters")
|
||||
}
|
||||
if strings.Contains(name, "\x00") {
|
||||
return "", fmt.Errorf("file_name contains null bytes")
|
||||
}
|
||||
if strings.Contains(name, "\\") {
|
||||
return "", fmt.Errorf("file_name contains backslash")
|
||||
}
|
||||
cleaned := filepath.Clean(name)
|
||||
if filepath.IsAbs(cleaned) {
|
||||
return "", fmt.Errorf("file_name must be a relative path")
|
||||
}
|
||||
for _, part := range strings.Split(cleaned, string(filepath.Separator)) {
|
||||
if part == ".." {
|
||||
return "", fmt.Errorf("file_name contains path traversal")
|
||||
}
|
||||
}
|
||||
return cleaned, nil
|
||||
}
|
||||
|
||||
// sanitizeFileName validates file name: max 255 chars, no path separators,
|
||||
// no null bytes, no "..", alphanumeric + "-_." only.
|
||||
func sanitizeFileName(name string) (string, error) {
|
||||
|
||||
@@ -111,6 +111,7 @@ func (t *WorkspaceReadTool) executeList(team *store.TeamData, channel, chatID st
|
||||
var totalSize int64
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() {
|
||||
lines = append(lines, fmt.Sprintf("- %s/ (directory)", entry.Name()))
|
||||
continue
|
||||
}
|
||||
info, err := entry.Info()
|
||||
@@ -144,8 +145,8 @@ func (t *WorkspaceReadTool) executeRead(args map[string]any, team *store.TeamDat
|
||||
return ErrorResult("file_name is required for action=read")
|
||||
}
|
||||
|
||||
// Sanitize to prevent path traversal.
|
||||
name, err := sanitizeFileName(fileName)
|
||||
// Sanitize: allow subdirectory paths but prevent traversal.
|
||||
name, err := sanitizeFilePath(fileName)
|
||||
if err != nil {
|
||||
return ErrorResult(err.Error())
|
||||
}
|
||||
@@ -192,7 +193,8 @@ func (t *WorkspaceReadTool) executeDelete(args map[string]any, team *store.TeamD
|
||||
return ErrorResult("reviewers cannot delete workspace files")
|
||||
}
|
||||
|
||||
name, err := sanitizeFileName(fileName)
|
||||
// Sanitize: allow subdirectory paths but prevent traversal.
|
||||
name, err := sanitizeFilePath(fileName)
|
||||
if err != nil {
|
||||
return ErrorResult(err.Error())
|
||||
}
|
||||
|
||||
@@ -68,7 +68,7 @@ export const KanbanCard = memo(function KanbanCard({ task, isTeamV2, emojiLookup
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<p className="line-clamp-2 text-sm font-medium leading-snug">{task.subject}</p>
|
||||
<p className="line-clamp-2 text-xs font-medium leading-snug">{task.subject}</p>
|
||||
|
||||
{/* Bottom row: owner + type badge + priority */}
|
||||
<div className="mt-2 flex items-center gap-1.5">
|
||||
|
||||
Reference in new issue
Block a user