mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-18 12:14:24 +00:00
feat(teams): team system v2 with bug fixes, workspace scope, versioning, and prompt optimization (#183)
* feat(workspace): add team shared workspace for file collaboration
- Add workspace_write and workspace_read tools for agents to share files across team members
- Create team_workspaces DB table with migration 000017 (file metadata, pinning, tags)
- Implement PostgreSQL store layer for workspace CRUD operations
- Add RPC handlers for workspace list/read/delete from web UI
- Build React workspace tab with file listing, content preview, and delete
- Propagate workspace channel/chatID scope through delegation chain
- Auto-allow workspace tools in agent tool policy when agent belongs to a team
- Inject team workspace guidance into system prompt for team agents
- Add /reset command handler for clearing session history
- Harden MCP bridge context middleware to reject headers when no gateway token
- Add i18n strings for workspace UI in en/vi/zh locales
* feat(teams): add comprehensive task management with followup reminders and recovery
- Add task followup/reminder system with auto-set on lead agent reply and auto-clear when user responds on channel
- Add task recovery ticker to re-dispatch stale/pending tasks periodically
- Add task scopes, filtering by status/channel/chatID, and task events
- Add WS RPC handlers for task CRUD, assignments, comments, events, and bulk operations (teams_tasks.go)
- Add task detail dialog, settings UI for followup config, and scope filtering in web dashboard
- Add migrations 000018 (team_tasks_v2) and 000019 (task_followup)
- Extend team_tasks_tool with await_reply, clear_followup actions
- Auto-complete/fail team tasks when delegate agent finishes
- Add workspace file listing and team tool manager enhancements
* docs(teams): add team system architecture and playbook ideas documentation
- Add TEAM_SYSTEM.md with full architecture design covering task management, shared workspace, and delegation engine subsystems
- Add TEAM_PLAYBOOK_IDEAS.md outlining future team coordination layers (playbook, member capabilities, auto-learned patterns)
- Document data models, status flows, tool actions, followup reminder system, task ticker, execution locking, and workspace scope model
* fix(teams): resolve 6 critical bugs in team task system
- Fix unblock SQL: check array_length after array_remove (not before)
- Enforce single-team leadership in team creation
- Add requireLead() for approve/reject tool actions
- Validate cross-team dependency references in blocked_by
- Add team_id to handoff route for multi-team isolation
- Set blocked_by DEFAULT '{}' to prevent NULL array issues
* refactor(workspace): use stable userID as scope key instead of connection UUID
Workspace scope changed from (team_id, channel, chat_id) to (team_id, userID).
Fixes workspace fragmentation across WS tab refreshes and reconnections.
* feat(teams): add V1/V2 versioning with feature gating and optimized prompts
- IsTeamV2() helper gates advanced features (locking, followup, review, audit)
- V2 tool actions rejected for V1 teams with clear error message
- Ticker, gateway consumer, delegation hooks respect version flag
- TEAM.md renders v1/v2 sections conditionally
- Tool descriptions and params optimized (~38% token reduction)
- UI: version toggle in settings, V2 Beta badge, conditional rendering
- i18n: version modal keys for en/vi/zh
* fix(migration): use VARCHAR(255) for user ID columns and add metadata JSONB
- assignee_user_id, user_id, actor_id: TEXT → VARCHAR(255)
- Add metadata JSONB to team_task_comments and team_task_attachments
---------
Co-authored-by: Nam Nguyen Ngoc <namnn.0911@gmail.com>
This commit is contained in:
1 parent
7bde3f2f7c
commit
1a42dc93a6
71 files changed
+7908
-790
No files matched your search
+12
-1
@@ -36,6 +36,7 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/skills"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store/pg"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tasks"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tracing"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/browser"
|
||||
@@ -1027,6 +1028,13 @@ func runGateway() {
|
||||
|
||||
go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, delegateMgr, pgStores.Sessions, pgStores.Agents, contactCollector)
|
||||
|
||||
// Task recovery ticker: re-dispatches stale/pending team tasks on startup and periodically.
|
||||
var taskTicker *tasks.TaskTicker
|
||||
if pgStores.Teams != nil {
|
||||
taskTicker = tasks.NewTaskTicker(pgStores.Teams, pgStores.Agents, msgBus, cfg.Gateway.TaskRecoveryIntervalSec)
|
||||
taskTicker.Start()
|
||||
}
|
||||
|
||||
go func() {
|
||||
sig := <-sigCh
|
||||
slog.Info("graceful shutdown initiated", "signal", sig)
|
||||
@@ -1034,9 +1042,12 @@ func runGateway() {
|
||||
// Broadcast shutdown event
|
||||
server.BroadcastEvent(*protocol.NewEvent(protocol.EventShutdown, nil))
|
||||
|
||||
// Stop channels and cron
|
||||
// Stop channels, cron, and task ticker
|
||||
channelMgr.StopAll(context.Background())
|
||||
pgStores.Cron.Stop()
|
||||
if taskTicker != nil {
|
||||
taskTicker.Stop()
|
||||
}
|
||||
|
||||
// Drain audit log queue before closing DB
|
||||
if auditCh != nil {
|
||||
|
||||
@@ -124,6 +124,12 @@ func builtinToolSeedData() []store.BuiltinToolDef {
|
||||
{Name: "team_message", DisplayName: "Team Message", Description: "Send a direct message or broadcast to teammates in the agent team", Category: "teams", Enabled: true,
|
||||
Requires: []string{"managed_mode", "teams"},
|
||||
},
|
||||
{Name: "workspace_write", DisplayName: "Workspace Write", Description: "Write files to the team shared workspace visible to all team members", Category: "teams", Enabled: true,
|
||||
Requires: []string{"managed_mode", "teams"},
|
||||
},
|
||||
{Name: "workspace_read", DisplayName: "Workspace Read", Description: "Read, list, delete, pin, and tag files in the team shared workspace", Category: "teams", Enabled: true,
|
||||
Requires: []string{"managed_mode", "teams"},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+173
-9
@@ -2,6 +2,7 @@ package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
@@ -16,6 +17,7 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/sessions"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
@@ -159,6 +161,18 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
quotaChecker.Increment(userID)
|
||||
}
|
||||
|
||||
// Auto-clear followup reminders when user sends a message on a real channel.
|
||||
// Fire-and-forget: don't block message processing.
|
||||
if teamStore != nil && msg.Channel != tools.ChannelSystem && msg.Channel != tools.ChannelDelegate && msg.Channel != tools.ChannelDashboard {
|
||||
go func(ch, cid string) {
|
||||
if n, err := teamStore.ClearFollowupByScope(ctx, ch, cid); err != nil {
|
||||
slog.Warn("auto-clear followup failed", "channel", ch, "chat_id", cid, "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info("auto-clear followup: cleared", "channel", ch, "chat_id", cid, "count", n)
|
||||
}
|
||||
}(msg.Channel, msg.ChatID)
|
||||
}
|
||||
|
||||
slog.Info("inbound: scheduling message (main lane)",
|
||||
"channel", msg.Channel,
|
||||
"chat_id", msg.ChatID,
|
||||
@@ -324,7 +338,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
})
|
||||
|
||||
// Handle result asynchronously to not block the flush callback.
|
||||
go func(channel, chatID, session, rID string, meta map[string]string, blockReplyEnabled bool) {
|
||||
go func(agentKey, channel, chatID, session, rID string, meta map[string]string, blockReplyEnabled bool) {
|
||||
outcome := <-outCh
|
||||
|
||||
// Clean up run tracking (in case HandleAgentEvent didn't fire for terminal events)
|
||||
@@ -398,7 +412,12 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
appendMediaToOutbound(&outMsg, outcome.Result.Media)
|
||||
|
||||
msgBus.PublishOutbound(outMsg)
|
||||
}(msg.Channel, msg.ChatID, sessionKey, runID, outMeta, blockReply)
|
||||
|
||||
// Auto-set followup when lead agent replies on a real channel with in_progress tasks.
|
||||
if teamStore != nil && channel != tools.ChannelSystem && channel != tools.ChannelDelegate && channel != tools.ChannelDashboard {
|
||||
go autoSetFollowup(ctx, teamStore, agentStore, agentKey, channel, chatID, outcome.Result.Content)
|
||||
}
|
||||
}(agentID, msg.Channel, msg.ChatID, sessionKey, runID, outMeta, blockReply)
|
||||
}
|
||||
|
||||
// Inbound debounce: merge rapid messages from the same sender before processing.
|
||||
@@ -432,7 +451,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
|
||||
// --- Subagent announce: bypass debounce, inject into parent agent session ---
|
||||
if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "subagent:") {
|
||||
if msg.Channel == tools.ChannelSystem && strings.HasPrefix(msg.SenderID, "subagent:") {
|
||||
origChannel := msg.Metadata["origin_channel"]
|
||||
origPeerKind := msg.Metadata["origin_peer_kind"]
|
||||
origLocalKey := msg.Metadata["origin_local_key"]
|
||||
@@ -566,7 +585,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
|
||||
// --- Delegate announce: bypass debounce, inject into parent agent session ---
|
||||
// Same pattern as subagent announce above, using "delegate" lane.
|
||||
if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "delegate:") {
|
||||
if msg.Channel == tools.ChannelSystem && strings.HasPrefix(msg.SenderID, "delegate:") {
|
||||
origChannel := msg.Metadata["origin_channel"]
|
||||
origPeerKind := msg.Metadata["origin_peer_kind"]
|
||||
origLocalKey := msg.Metadata["origin_local_key"]
|
||||
@@ -690,7 +709,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
|
||||
// --- Handoff announce: route initial message to target agent session ---
|
||||
// Same pattern as teammate message routing, using "delegate" lane.
|
||||
if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "handoff:") {
|
||||
if msg.Channel == tools.ChannelSystem && strings.HasPrefix(msg.SenderID, "handoff:") {
|
||||
origChannel := msg.Metadata["origin_channel"]
|
||||
origPeerKind := msg.Metadata["origin_peer_kind"]
|
||||
origLocalKey := msg.Metadata["origin_local_key"]
|
||||
@@ -760,7 +779,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
|
||||
// --- Teammate message: bypass debounce, route to target agent session ---
|
||||
// Same pattern as delegate announce, using "delegate" lane.
|
||||
if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "teammate:") {
|
||||
if msg.Channel == tools.ChannelSystem && strings.HasPrefix(msg.SenderID, "teammate:") {
|
||||
origChannel := msg.Metadata["origin_channel"]
|
||||
origPeerKind := msg.Metadata["origin_peer_kind"]
|
||||
origLocalKey := msg.Metadata["origin_local_key"]
|
||||
@@ -774,7 +793,13 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
|
||||
if origChannel == "" || msg.ChatID == "" {
|
||||
slog.Warn("teammate message: missing origin", "sender", msg.SenderID)
|
||||
slog.Warn("teammate message: missing origin — DROPPED",
|
||||
"sender", msg.SenderID,
|
||||
"target", targetAgent,
|
||||
"origin_channel", origChannel,
|
||||
"chat_id", msg.ChatID,
|
||||
"user_id", msg.UserID,
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -807,8 +832,45 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
Stream: false,
|
||||
})
|
||||
|
||||
go func(origCh, chatID, senderID string, meta map[string]string) {
|
||||
go func(origCh, chatID, senderID string, meta, inMeta map[string]string) {
|
||||
outcome := <-outCh
|
||||
|
||||
// Auto-complete/fail the associated team task (v2 only).
|
||||
if taskIDStr := inMeta["team_task_id"]; taskIDStr != "" {
|
||||
teamTaskID, _ := uuid.Parse(taskIDStr)
|
||||
teamID, _ := uuid.Parse(inMeta["team_id"])
|
||||
if teamTaskID != uuid.Nil && teamStore != nil {
|
||||
// Only auto-complete/fail for v2 teams.
|
||||
team, _ := teamStore.GetTeam(ctx, teamID)
|
||||
if team != nil && isConsumerTeamV2(team) {
|
||||
if outcome.Err != nil {
|
||||
_ = teamStore.FailTask(ctx, teamTaskID, teamID, outcome.Err.Error())
|
||||
_ = teamStore.RecordTaskEvent(ctx, &store.TeamTaskEventData{
|
||||
TaskID: teamTaskID,
|
||||
EventType: "failed",
|
||||
ActorType: "agent",
|
||||
ActorID: inMeta["to_agent"],
|
||||
})
|
||||
} else {
|
||||
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]"
|
||||
}
|
||||
_ = teamStore.CompleteTask(ctx, teamTaskID, teamID, result)
|
||||
_ = teamStore.RecordTaskEvent(ctx, &store.TeamTaskEventData{
|
||||
TaskID: teamTaskID,
|
||||
EventType: "completed",
|
||||
ActorType: "agent",
|
||||
ActorID: inMeta["to_agent"],
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if outcome.Err != nil {
|
||||
slog.Error("teammate message: agent run failed", "error", outcome.Err)
|
||||
return
|
||||
@@ -827,7 +889,32 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
appendMediaToOutbound(&outMsg, outcome.Result.Media)
|
||||
msgBus.PublishOutbound(outMsg)
|
||||
}(origChannel, msg.ChatID, msg.SenderID, outMeta)
|
||||
}(origChannel, msg.ChatID, msg.SenderID, outMeta, msg.Metadata)
|
||||
continue
|
||||
}
|
||||
|
||||
// --- Command: /reset — clear session history ---
|
||||
if msg.Metadata["command"] == "reset" {
|
||||
agentID := msg.AgentID
|
||||
if agentID == "" {
|
||||
agentID = resolveAgentRoute(cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
peerKind := msg.PeerKind
|
||||
if peerKind == "" {
|
||||
peerKind = string(sessions.PeerDirect)
|
||||
}
|
||||
sessionKey := sessions.BuildScopedSessionKey(agentID, msg.Channel, sessions.PeerKind(peerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey)
|
||||
if msg.Metadata["is_forum"] == "true" && peerKind == string(sessions.PeerGroup) {
|
||||
var topicID int
|
||||
fmt.Sscanf(msg.Metadata["message_thread_id"], "%d", &topicID)
|
||||
if topicID > 0 {
|
||||
sessionKey = sessions.BuildGroupTopicSessionKey(agentID, msg.Channel, msg.ChatID, topicID)
|
||||
}
|
||||
}
|
||||
sessStore.Reset(sessionKey)
|
||||
sessStore.Save(sessionKey)
|
||||
providers.ResetCLISession("", sessionKey)
|
||||
slog.Info("inbound: /reset command", "session", sessionKey)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -903,6 +990,83 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
}
|
||||
|
||||
// autoSetFollowup sets followup reminders on in_progress tasks when the lead agent
|
||||
// replies on a real channel. Only sets followup if the task doesn't already have one
|
||||
// (respects LLM-initiated await_reply). Fire-and-forget, logs errors.
|
||||
func autoSetFollowup(ctx context.Context, teamStore store.TeamStore, agentStore store.AgentStore, agentKey, channel, chatID, content string) {
|
||||
if agentStore == nil {
|
||||
return
|
||||
}
|
||||
// agentKey may be a slug ("default") or a UUID string (from WS clients).
|
||||
var ag *store.AgentData
|
||||
var err error
|
||||
if id, parseErr := uuid.Parse(agentKey); parseErr == nil {
|
||||
ag, err = agentStore.GetByID(ctx, id)
|
||||
} else {
|
||||
ag, err = agentStore.GetByKey(ctx, agentKey)
|
||||
}
|
||||
if err != nil || ag == nil {
|
||||
return
|
||||
}
|
||||
team, err := teamStore.GetTeamForAgent(ctx, ag.ID)
|
||||
if err != nil || team == nil || team.LeadAgentID != ag.ID {
|
||||
return // only lead agent triggers auto-set
|
||||
}
|
||||
// Followup is a v2 feature.
|
||||
if !isConsumerTeamV2(team) {
|
||||
return
|
||||
}
|
||||
|
||||
interval, max := parseFollowupSettings(team)
|
||||
followupAt := time.Now().Add(interval)
|
||||
msg := truncateForReminder(content, 200)
|
||||
|
||||
n, err := teamStore.SetFollowupForActiveTasks(ctx, team.ID, channel, chatID, followupAt, max, msg)
|
||||
if err != nil {
|
||||
slog.Warn("auto-set followup failed", "channel", channel, "chat_id", chatID, "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info("auto-set followup: set", "channel", channel, "chat_id", chatID, "count", n, "followup_at", followupAt)
|
||||
}
|
||||
}
|
||||
|
||||
// isConsumerTeamV2 delegates to tools.IsTeamV2 for version checking.
|
||||
var isConsumerTeamV2 = tools.IsTeamV2
|
||||
|
||||
// parseFollowupSettings extracts followup interval and max reminders from team settings.
|
||||
func parseFollowupSettings(team *store.TeamData) (time.Duration, int) {
|
||||
const (
|
||||
defaultIntervalMins = 30
|
||||
defaultMax = 0 // unlimited
|
||||
)
|
||||
if team.Settings == nil {
|
||||
return time.Duration(defaultIntervalMins) * time.Minute, defaultMax
|
||||
}
|
||||
var settings map[string]any
|
||||
if json.Unmarshal(team.Settings, &settings) != nil {
|
||||
return time.Duration(defaultIntervalMins) * time.Minute, defaultMax
|
||||
}
|
||||
interval := defaultIntervalMins
|
||||
if v, ok := settings["followup_interval_minutes"].(float64); ok && v > 0 {
|
||||
interval = int(v)
|
||||
}
|
||||
max := defaultMax
|
||||
if v, ok := settings["followup_max_reminders"].(float64); ok && v >= 0 {
|
||||
max = int(v)
|
||||
}
|
||||
return time.Duration(interval) * time.Minute, max
|
||||
}
|
||||
|
||||
// truncateForReminder truncates content to maxLen chars, taking the last line as context.
|
||||
func truncateForReminder(content string, maxLen int) string {
|
||||
// Use last non-empty line as it's typically the most relevant.
|
||||
lines := strings.Split(strings.TrimSpace(content), "\n")
|
||||
msg := lines[len(lines)-1]
|
||||
if len(msg) > maxLen {
|
||||
msg = msg[:maxLen] + "..."
|
||||
}
|
||||
return msg
|
||||
}
|
||||
|
||||
// appendMediaToOutbound converts agent MediaResults to outbound MediaAttachments
|
||||
// on the given OutboundMessage. Handles voice annotation when applicable.
|
||||
func appendMediaToOutbound(msg *bus.OutboundMessage, media []agent.MediaResult) {
|
||||
|
||||
@@ -406,6 +406,8 @@ func wireExtras(
|
||||
TeamID: req.TeamID,
|
||||
TeamTaskID: req.TeamTaskID,
|
||||
ParentAgentID: req.ParentAgentID,
|
||||
WorkspaceChannel: req.WorkspaceChannel,
|
||||
WorkspaceChatID: req.WorkspaceChatID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -450,7 +452,7 @@ func wireExtras(
|
||||
toolsReg.Register(tools.NewEvaluateLoopTool(delegateMgr))
|
||||
|
||||
// Handoff tool (agent-to-agent conversation transfer)
|
||||
toolsReg.Register(tools.NewHandoffTool(delegateMgr, stores.Teams, stores.Sessions, msgBus))
|
||||
toolsReg.Register(tools.NewHandoffTool(delegateMgr, stores.Teams, stores.Sessions, msgBus, workspace))
|
||||
|
||||
// Inject delegation capability into existing SpawnTool
|
||||
if st, ok := toolsReg.Get("spawn"); ok {
|
||||
@@ -484,7 +486,7 @@ func wireExtras(
|
||||
slog.Info("delegate + delegate_search tools registered")
|
||||
}
|
||||
|
||||
// Register team tools (team_tasks + team_message) if team store is available.
|
||||
// Register team tools (team_tasks + team_message + workspace) if team store is available.
|
||||
if stores.Teams != nil && stores.Agents != nil {
|
||||
teamMgr := tools.NewTeamToolManager(stores.Teams, stores.Agents, msgBus)
|
||||
if delegateMgr != nil {
|
||||
@@ -492,6 +494,9 @@ func wireExtras(
|
||||
}
|
||||
toolsReg.Register(tools.NewTeamTasksTool(teamMgr))
|
||||
toolsReg.Register(tools.NewTeamMessageTool(teamMgr))
|
||||
toolsReg.Register(tools.NewWorkspaceWriteTool(teamMgr, workspace))
|
||||
toolsReg.Register(tools.NewWorkspaceReadTool(teamMgr, workspace))
|
||||
slog.Info("team + workspace tools registered", "workspace", workspace)
|
||||
|
||||
// Team cache invalidation via pub/sub
|
||||
msgBus.Subscribe(bus.TopicCacheTeam, func(event bus.Event) {
|
||||
|
||||
Reference in new issue
Block a user