feat: add team work classification routing (#1379)

Approved by github-maintain bot. Clean team work classification feature with comprehensive tests and i18n.

cmd/gateway.go

@@ -372,7 +372,7 @@ func runGateway() {
 		slog.Debug("skipping MCP database init: pgStores.MCP is nil")
 	}
 
-	setupMemoryEmbeddings(pgStores, providerRegistry)
+	teamWorkEmbedder := setupMemoryEmbeddings(pgStores, providerRegistry)
 	usageCapSvc := usagecaps.NewService(pgStores.UsageCaps, pgStores.Providers)
 
 	// Resolve background provider for consolidation + vault enrichment.
@@ -532,6 +532,7 @@ func runGateway() {
 		domainBus:        domainBus,
 		usageCapSvc:      usageCapSvc,
 		audioMgr:         audioMgr,
+		teamWorkEmbedder: teamWorkEmbedder,
 	}
 
 	gatewayAddr := loopbackAddr(cfg.Gateway.Host, cfg.Gateway.Port)
@@ -639,7 +640,7 @@ func runGateway() {
 	// Register all RPC methods
 	server.SetLogTee(logTee)
 	server.SetRuntimeLogsHandler(httpapi.NewRuntimeLogsHandler(logTee))
-	pairingMethods, heartbeatMethods, chatMethods, cfgPermsMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Tracing, pgStores.RunTimeline, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc, providerRegistry)
+	pairingMethods, heartbeatMethods, chatMethods, cfgPermsMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Tracing, pgStores.RunTimeline, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, pgStores.AgentLinks, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc, providerRegistry, teamWorkEmbedder)
 
 	// Phase 3: Agent hooks RPC methods (hooks.list/create/update/delete/toggle/test/history).
 	if hs, ok := pgStores.Hooks.(hooks.HookStore); ok && hs != nil {

cmd/gateway_consumer.go

@@ -15,6 +15,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/bus"
 	"github.com/nextlevelbuilder/goclaw/internal/channels"
 	"github.com/nextlevelbuilder/goclaw/internal/config"
+	"github.com/nextlevelbuilder/goclaw/internal/memory"
 	"github.com/nextlevelbuilder/goclaw/internal/providers"
 	"github.com/nextlevelbuilder/goclaw/internal/scheduler"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -27,7 +28,7 @@ import (
 // and routes them through the scheduler/agent loop, then publishes the response back.
 // Also handles subagent announcements: routes them through the parent agent's session
 // (matching TS subagent-announce.ts pattern) so the agent can reformulate for the user.
-func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor, subagentMgr *tools.SubagentManager, usageCapSvc *usagecaps.Service, providerReg *providers.Registry) {
+func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, agentLinkStore store.AgentLinkStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor, subagentMgr *tools.SubagentManager, usageCapSvc *usagecaps.Service, providerReg *providers.Registry, teamWorkEmbedder memory.EmbeddingProvider) {
 	slog.Info("inbound message consumer started")
 
 	// Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts).
@@ -53,6 +54,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 		ChannelMgr:       channelMgr,
 		MsgBus:           msgBus,
 		TeamStore:        teamStore,
+		AgentLinkStore:   agentLinkStore,
 		AgentStore:       agentStore,
 		SessStore:        sessStore,
 		PostTurn:         postTurn,
@@ -61,6 +63,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 		SubagentMgr:      subagentMgr,
 		UsageCaps:        usageCapSvc,
 		ProviderReg:      providerReg,
+		TeamWorkEmbedder: teamWorkEmbedder,
 		GetAnnounceMu:    getAnnounceMu,
 	}
 

cmd/gateway_consumer_deps.go

@@ -7,6 +7,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/bus"
 	"github.com/nextlevelbuilder/goclaw/internal/channels"
 	"github.com/nextlevelbuilder/goclaw/internal/config"
+	"github.com/nextlevelbuilder/goclaw/internal/memory"
 	"github.com/nextlevelbuilder/goclaw/internal/providers"
 	"github.com/nextlevelbuilder/goclaw/internal/scheduler"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -23,6 +24,7 @@ type ConsumerDeps struct {
 	ChannelMgr       *channels.Manager
 	MsgBus           *bus.MessageBus
 	TeamStore        store.TeamStore
+	AgentLinkStore   store.AgentLinkStore
 	AgentStore       store.AgentStore
 	SessStore        store.SessionStore
 	PostTurn         tools.PostTurnProcessor
@@ -32,6 +34,7 @@ type ConsumerDeps struct {
 	SubagentMgr      *tools.SubagentManager
 	UsageCaps        *usagecaps.Service
 	ProviderReg      *providers.Registry
+	TeamWorkEmbedder memory.EmbeddingProvider
 	BgWg             sync.WaitGroup
 	GetAnnounceMu    func(string) *sync.Mutex
 }

cmd/gateway_consumer_normal.go

@@ -407,11 +407,16 @@ func processNormalMessage(
 		}
 	}
 
+	inboundMessage := msg.Content
+
 	// Inject tenant context from channel instance so all store queries are tenant-scoped.
 	if msg.TenantID != uuid.Nil {
 		ctx = store.WithTenantID(ctx, msg.TenantID)
 	}
 
+	gate := applyTeamWorkGateForInbound(ctx, deps, msg, sessionKey, agentID, peerKind, agentLoop.UUID(), skillFilter, agentLoop.Provider(), agentLoop.Model())
+	inboundMessage = gate.Message
+
 	// Inject post-turn dispatch tracker so team task creates are deferred.
 	ptd := tools.NewPendingTeamDispatch()
 	schedCtx := tools.WithPendingTeamDispatch(ctx, ptd)
@@ -447,7 +452,7 @@ func processNormalMessage(
 	// Schedule through main lane (per-session concurrency controlled by maxConcurrent)
 	outCh := deps.Sched.ScheduleWithOpts(schedCtx, "main", agent.RunRequest{
 		SessionKey:   sessionKey,
-		Message:      msg.Content,
+		Message:      inboundMessage,
 		Media:        reqMedia,
 		ForwardMedia: fwdMedia,
 		Channel:      msg.Channel,
@@ -471,6 +476,7 @@ func processNormalMessage(
 		ToolAllow:                  msg.ToolAllow,
 		TelegramManagerPermissions: msg.TelegramManagerPermissions,
 		ExtraSystemPrompt:          extraPrompt,
+		TeamWorkDirective:          gate.Directive,
 		SkillFilter:                skillFilter,
 	}, scheduler.ScheduleOpts{
 		MaxConcurrent: maxConcurrent,
@@ -615,7 +621,7 @@ func processNormalMessage(
 		if deps.TeamStore != nil && channel != tools.ChannelSystem && channel != tools.ChannelTeammate && channel != tools.ChannelDashboard {
 			go autoSetFollowup(ctx, deps.TeamStore, deps.AgentStore, agentKey, channel, chatID, replyContent)
 		}
-	}(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, msg.Content, outMeta, blockReply, chatBehavior, channelStream, ptd, msg.TenantID, agentLoop.UUID(), agentLoop.OtherConfig())
+	}(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, inboundMessage, outMeta, blockReply, chatBehavior, channelStream, ptd, msg.TenantID, agentLoop.UUID(), agentLoop.OtherConfig())
 }
 
 func buildDeliveryRuntime(ctx context.Context, deps *ConsumerDeps, agentLoop agent.Agent, behavior channels.ResolvedChatBehavior, msg bus.InboundMessage, userID, peerKind, channelType, agentKey string) channels.DeliveryRuntime {

cmd/gateway_deps.go

@@ -10,6 +10,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/eventbus"
 	"github.com/nextlevelbuilder/goclaw/internal/gateway"
 	httpapi "github.com/nextlevelbuilder/goclaw/internal/http"
+	"github.com/nextlevelbuilder/goclaw/internal/memory"
 	"github.com/nextlevelbuilder/goclaw/internal/providers"
 	"github.com/nextlevelbuilder/goclaw/internal/skills"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -39,4 +40,5 @@ type gatewayDeps struct {
 	usageCapSvc      *usagecaps.Service
 	audioMgr         *audio.Manager      // nil if TTS not configured; used by TTSHandler
 	ttsHandler       *httpapi.TTSHandler // nil if TTS not configured; for hot-reload
+	teamWorkEmbedder memory.EmbeddingProvider
 }

cmd/gateway_lifecycle.go

@@ -147,7 +147,7 @@ func (d *gatewayDeps) runLifecycle(
 		d.channelMgr.SetContactCollector(contactCollector)
 	}
 
-	go consumeInboundMessages(ctx, d.msgBus, d.agentRouter, d.cfg, deps.sched, d.channelMgr, deps.consumerTeamStore, deps.quotaChecker, d.pgStores.Sessions, d.pgStores.Agents, contactCollector, deps.postTurn, deps.subagentMgr, d.usageCapSvc, d.providerRegistry)
+	go consumeInboundMessages(ctx, d.msgBus, d.agentRouter, d.cfg, deps.sched, d.channelMgr, deps.consumerTeamStore, d.pgStores.AgentLinks, deps.quotaChecker, d.pgStores.Sessions, d.pgStores.Agents, contactCollector, deps.postTurn, deps.subagentMgr, d.usageCapSvc, d.providerRegistry, d.teamWorkEmbedder)
 
 	// Webhook callback worker — delivers async webhook_calls rows to receiver callback_url.
 	// Runs in both editions: Standard (PG, concurrency=4) and Lite (SQLite, concurrency=1).

cmd/gateway_methods.go

@@ -10,19 +10,21 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/config"
 	"github.com/nextlevelbuilder/goclaw/internal/gateway"
 	"github.com/nextlevelbuilder/goclaw/internal/gateway/methods"
+	"github.com/nextlevelbuilder/goclaw/internal/memory"
 	"github.com/nextlevelbuilder/goclaw/internal/providers"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
 	"github.com/nextlevelbuilder/goclaw/internal/tools"
 	usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps"
 )
 
-func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, tracingStore store.TracingStore, runTimeline store.RunTimelineStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee, heartbeatStore store.HeartbeatStore, configPermStore store.ConfigPermissionStore, sysConfigStore store.SystemConfigStore, tenantStore store.TenantStore, skillTenantCfgStore store.SkillTenantConfigStore, audioMgr *audio.Manager, usageCapSvc *usagecaps.Service, providerReg *providers.Registry) (*methods.PairingMethods, *methods.HeartbeatMethods, *methods.ChatMethods, *methods.ConfigPermissionsMethods) {
+func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, tracingStore store.TracingStore, runTimeline store.RunTimelineStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, agentLinkStore store.AgentLinkStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee, heartbeatStore store.HeartbeatStore, configPermStore store.ConfigPermissionStore, sysConfigStore store.SystemConfigStore, tenantStore store.TenantStore, skillTenantCfgStore store.SkillTenantConfigStore, audioMgr *audio.Manager, usageCapSvc *usagecaps.Service, providerReg *providers.Registry, teamWorkEmbedder memory.EmbeddingProvider) (*methods.PairingMethods, *methods.HeartbeatMethods, *methods.ChatMethods, *methods.ConfigPermissionsMethods) {
 	router := server.Router()
 
 	// Phase 1: Core methods
 	chatMethods := methods.NewChatMethods(agents, sessStore, cfg, server.RateLimiter(), msgBus)
 	chatMethods.SetAudioManager(audioMgr) // Wire TTS auto-apply for WS responses
 	chatMethods.SetUsageCapService(usageCapSvc)
+	chatMethods.SetTeamWorkClassification(agentStore, teamStore, agentLinkStore, teamWorkEmbedder)
 	chatMethods.Register(router)
 	methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, contextFileInterceptor, msgBus).Register(router)
 	methods.NewSessionsMethods(sessStore, msgBus, cfg).Register(router)

cmd/gateway_setup.go

@@ -17,6 +17,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/config"
 	"github.com/nextlevelbuilder/goclaw/internal/edition"
 	mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp"
+	"github.com/nextlevelbuilder/goclaw/internal/memory"
 	"github.com/nextlevelbuilder/goclaw/internal/permissions"
 	"github.com/nextlevelbuilder/goclaw/internal/providers"
 	"github.com/nextlevelbuilder/goclaw/internal/sandbox"
@@ -370,9 +371,11 @@ func wireTracingAndCron(
 func setupMemoryEmbeddings(
 	pgStores *store.Stores,
 	providerRegistry *providers.Registry,
-) {
+) memory.EmbeddingProvider {
+	var resolved memory.EmbeddingProvider
 	if pgStores.Memory != nil {
 		if embProvider := resolveEmbeddingProvider(pgStores.Providers, providerRegistry, pgStores.SystemConfigs); embProvider != nil {
+			resolved = embProvider
 			pgStores.Memory.SetEmbeddingProvider(embProvider)
 			slog.Info("memory embeddings enabled", "provider", embProvider.Name(), "model", embProvider.Model())
 
@@ -431,6 +434,7 @@ func setupMemoryEmbeddings(
 			slog.Warn("memory embeddings disabled (no API key), chunks stored without vectors")
 		}
 	}
+	return resolved
 }
 
 // seedSystemConfigs ensures system_configs has all expected keys for all tenants.

cmd/gateway_system_config_sync.go

@@ -108,6 +108,7 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
 	setIntAllowZero("gateway.inbound_debounce_ms", cfg.Gateway.InboundDebounceMs)
 	setBool("gateway.block_reply", cfg.Gateway.BlockReply)
 	setBool("gateway.tool_status", cfg.Gateway.ToolStatus)
+	setBool("gateway.team_work_classify", cfg.Gateway.TeamWorkClassify)
 	setInt("gateway.task_recovery_interval_sec", cfg.Gateway.TaskRecoveryIntervalSec)
 
 	// Background workers

cmd/gateway_team_work_classify.go

@@ -0,0 +1,66 @@
+package cmd
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/agent"
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
+	"github.com/nextlevelbuilder/goclaw/internal/providers"
+	"github.com/nextlevelbuilder/goclaw/internal/teamworkclassify"
+)
+
+type teamWorkGateOutcome struct {
+	Message   string
+	Directive *agent.TeamWorkDirective
+}
+
+func applyTeamWorkGateForInbound(ctx context.Context, deps *ConsumerDeps, msg bus.InboundMessage, sessionKey, agentKey, peerKind string, agentUUID uuid.UUID, skillFilter []string, provider providers.Provider, model string) teamWorkGateOutcome {
+	out := teamWorkGateOutcome{Message: msg.Content}
+	if deps == nil || deps.Cfg == nil || deps.Cfg.Gateway.TeamWorkClassify == nil || !*deps.Cfg.Gateway.TeamWorkClassify {
+		return out
+	}
+	if deps.TeamWorkEmbedder == nil {
+		slog.Info("team_work_classify: skipped; embedding unavailable", "agent", agentKey, "session", sessionKey)
+		return out
+	}
+	if agentUUID == uuid.Nil {
+		return out
+	}
+	mode := agent.ResolveOrchestrationMode(ctx, agentUUID, deps.TeamStore, deps.AgentLinkStore)
+	if mode == agent.ModeSpawn {
+		slog.Info("team_work_classify: skipped; no team/delegate capability", "agent", agentKey, "session", sessionKey)
+		return out
+	}
+	if msg.Metadata["run_kind"] != "" || msg.Metadata["delegation_id"] != "" || msg.Metadata["subagent_id"] != "" || bus.IsInternalSender(msg.SenderID) {
+		return out
+	}
+
+	input := teamworkclassify.BuildInputFromStores(ctx, teamworkclassify.ProfileStores{
+		Agents:     deps.AgentStore,
+		Teams:      deps.TeamStore,
+		AgentLinks: deps.AgentLinkStore,
+	}, teamworkclassify.BuildInputOptions{
+		Mode:        teamworkclassify.Mode(mode),
+		Message:     msg.Content,
+		AgentID:     agentUUID,
+		ToolAllow:   msg.ToolAllow,
+		SkillFilter: skillFilter,
+		Embedder:    deps.TeamWorkEmbedder,
+	})
+	result := teamworkclassify.ClassifyWithLLM(ctx, input, provider, model, deps.UsageCaps)
+	slog.Info("team_work_classify: decision", "agent", agentKey, "session", sessionKey, "mode", mode, "decision", result.Decision, "self_score", result.SelfScore, "collaboration_score", result.CollaborationScore, "reason", result.Reason)
+	if result.Decision == teamworkclassify.DecisionTeam {
+		out.Directive = &agent.TeamWorkDirective{
+			Mode:            string(result.Mode),
+			Source:          "llm",
+			Reason:          result.Reason,
+			OriginalMessage: msg.Content,
+			RequiredTool:    result.RequiredTool,
+			WorkflowHint:    result.WorkflowHint,
+		}
+	}
+	return out
+}

cmd/gateway_team_work_classify_test.go

@@ -0,0 +1,65 @@
+package cmd
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
+	"github.com/nextlevelbuilder/goclaw/internal/config"
+	"github.com/nextlevelbuilder/goclaw/internal/providers"
+)
+
+type teamWorkGateTestEmbedder struct {
+	called bool
+}
+
+func (e *teamWorkGateTestEmbedder) Name() string  { return "test-embedder" }
+func (e *teamWorkGateTestEmbedder) Model() string { return "test-embedding" }
+func (e *teamWorkGateTestEmbedder) Embed(context.Context, []string) ([][]float32, error) {
+	e.called = true
+	return [][]float32{{1, 0}}, nil
+}
+
+type teamWorkGateTestProvider struct {
+	called bool
+}
+
+func (p *teamWorkGateTestProvider) Name() string         { return "test-provider" }
+func (p *teamWorkGateTestProvider) DefaultModel() string { return "test-model" }
+func (p *teamWorkGateTestProvider) Chat(context.Context, providers.ChatRequest) (*providers.ChatResponse, error) {
+	p.called = true
+	return &providers.ChatResponse{Content: `{"decision":"team","mode":"team","required_tool":"team_tasks"}`}, nil
+}
+func (p *teamWorkGateTestProvider) ChatStream(context.Context, providers.ChatRequest, func(providers.StreamChunk)) (*providers.ChatResponse, error) {
+	p.called = true
+	return nil, nil
+}
+
+func TestApplyTeamWorkGateForInboundSkipsAgentWithoutTeamOrDelegateLink(t *testing.T) {
+	enabled := true
+	embedder := &teamWorkGateTestEmbedder{}
+	provider := &teamWorkGateTestProvider{}
+
+	out := applyTeamWorkGateForInbound(context.Background(), &ConsumerDeps{
+		Cfg:              &config.Config{Gateway: config.GatewayConfig{TeamWorkClassify: &enabled}},
+		TeamWorkEmbedder: embedder,
+	}, bus.InboundMessage{
+		Content:  "lập kế hoạch content và chiến lược cho chiến dịch mới",
+		Metadata: map[string]string{},
+	}, "session:test", "bao-an", "direct", uuid.New(), nil, provider, "test-model")
+
+	if out.Message != "lập kế hoạch content và chiến lược cho chiến dịch mới" {
+		t.Fatalf("Message = %q, want original message", out.Message)
+	}
+	if out.Directive != nil {
+		t.Fatalf("Directive = %+v, want nil", out.Directive)
+	}
+	if embedder.called {
+		t.Fatal("embedder was called even though agent has no team/delegate capability")
+	}
+	if provider.called {
+		t.Fatal("provider was called even though agent has no team/delegate capability")
+	}
+}