feat(team-work): one-call LLM routing classifier with native DAG orchestration

cmd/gateway.go

@@ -50,6 +50,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/skills"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
 	"github.com/nextlevelbuilder/goclaw/internal/systemmessages"
+	"github.com/nextlevelbuilder/goclaw/internal/teamworkconfig"
 	"github.com/nextlevelbuilder/goclaw/internal/tools"
 	usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps"
 	usagepricing "github.com/nextlevelbuilder/goclaw/internal/usage/pricing"
@@ -301,7 +302,6 @@ func runGateway() {
 	if browserMgr != nil && pgStores != nil && pgStores.BrowserCookies != nil && cfg.Tools.Browser.CookieSyncEnabled {
 		browserMgr.SetCookieProvider(newStoreBrowserCookieProvider(pgStores.BrowserCookies))
 	}
-
 	if ttsTool != nil && pgStores.SystemConfigs != nil {
 		ttsTool.SetSystemConfigStore(pgStores.SystemConfigs)
 	}
@@ -369,6 +369,12 @@ func runGateway() {
 			store.WithTenantID(context.Background(), store.MasterTenantID),
 		); err == nil && len(sysConfigs) > 0 {
 			cfg.ApplySystemConfigs(sysConfigs)
+			// Phase 7 Decision 8: process-wide, startup-only keys (e.g.
+			// gateway.task_recovery_interval_sec) are applied here from the
+			// master-tenant seed ONLY — never from the per-tenant
+			// system-config-changed subscriber, which would let any tenant's edit
+			// mutate a shared runtime value the consuming singleton never re-reads.
+			cfg.ApplyStartupSystemConfigs(sysConfigs)
 			slog.Info("system_configs applied to in-memory config", "keys", len(sysConfigs))
 		}
 	}
@@ -418,7 +424,10 @@ func runGateway() {
 		slog.Debug("skipping MCP database init: pgStores.MCP is nil")
 	}
 
-	teamWorkEmbedder := setupMemoryEmbeddings(pgStores, providerRegistry)
+	// Shared embedding wiring for memory, KG, vault/skill search and semantic task
+	// search. The Team Work classifier is a one-call LLM route and deliberately
+	// consumes no embedder, so the resolved provider is not propagated to it.
+	setupMemoryEmbeddings(pgStores, providerRegistry)
 	usageCapSvc := usagecaps.NewService(pgStores.UsageCaps, pgStores.Providers)
 
 	// Resolve background provider for consolidation + vault enrichment.
@@ -539,6 +548,18 @@ func runGateway() {
 	server.SetPairingService(pgStores.Pairing)
 	server.SetMessageBus(msgBus)
 	server.SetExecApprovalManager(execApprovalMgr)
+	if teamAccessStore, ok := pgStores.Teams.(store.UserTeamIDLister); ok {
+		server.Router().SetTeamAccessStore(teamAccessStore)
+	}
+	msgBus.Subscribe("team-access-snapshot", func(e bus.Event) {
+		p, ok := e.Payload.(bus.CacheInvalidatePayload)
+		if !ok || p.Kind != bus.CacheKindTeamAccess {
+			return
+		}
+		// Refresh asynchronously: MessageBus broadcasts synchronously, so a team
+		// mutation must never wait on its own snapshot reload.
+		go server.Router().HandleTeamAccessInvalidation(context.Background(), p.TenantID)
+	})
 	server.SetOAuthHandler(httpapi.NewOAuthHandler(pgStores.Providers, pgStores.ConfigSecrets, providerRegistry, msgBus))
 
 	// contextFileInterceptor is created inside wireExtras.
@@ -617,10 +638,20 @@ func runGateway() {
 		mcpOAuthRefresher = mcpoauth.NewRefresher(pgStores.MCPOAuthTokens, security.NewSafeClient(15*time.Second))
 	}
 
+	// One shared per-tenant Team Work config resolver is needed by both inbound
+	// classification and coordinator recovery replans. It is created before
+	// wireExtras so the injected replan service observes identical overrides.
+	teamWorkCfg := teamworkconfig.NewResolver(teamworkconfig.Defaults{
+		ClassifyEnabled:    cfg.Gateway.TeamWorkClassify,
+		ClassifierProvider: cfg.Gateway.TeamWorkClassifyProvider,
+		ClassifierModel:    cfg.Gateway.TeamWorkClassifyModel,
+		ClassifyTimeoutSec: cfg.Gateway.TeamWorkClassifyTimeoutSec,
+	}, pgStores.SystemConfigs)
+
 	var mcpPool *mcpbridge.Pool
 	var mediaStore *media.Store
 	var postTurn tools.PostTurnProcessor
-	contextFileInterceptor, mcpPool, mediaStore, postTurn = wireExtras(pgStores, agentRouter, providerRegistry, modelReg, msgBus, pgStores.Sessions, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, redisClient, domainBus, usageCapSvc, mcpOAuthRefresher, childRunAdmission)
+	contextFileInterceptor, mcpPool, mediaStore, postTurn = wireExtras(pgStores, agentRouter, providerRegistry, modelReg, msgBus, pgStores.Sessions, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, redisClient, domainBus, usageCapSvc, teamWorkCfg, mcpOAuthRefresher, childRunAdmission)
 	if mcpPool != nil {
 		defer mcpPool.Stop()
 	}
@@ -634,6 +665,7 @@ func runGateway() {
 		providerRegistry: providerRegistry,
 		agentRouter:      agentRouter,
 		toolsReg:         toolsReg,
+		toolPE:           toolPE,
 		skillsLoader:     skillsLoader,
 		enrichProgress:   enrichProgress,
 		enrichWorker:     enrichWorker,
@@ -643,7 +675,7 @@ func runGateway() {
 		domainBus:        domainBus,
 		usageCapSvc:      usageCapSvc,
 		audioMgr:         audioMgr,
-		teamWorkEmbedder: teamWorkEmbedder,
+		teamWorkCfg:      teamWorkCfg,
 	}
 
 	gatewayAddr := loopbackAddr(cfg.Gateway.Host, cfg.Gateway.Port)
@@ -751,7 +783,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, pgStores.AgentLinks, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc, providerRegistry, teamWorkEmbedder)
+	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, skillsLoader, pgStores.ConfigSecrets, pgStores.Teams, pgStores.AgentLinks, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc, providerRegistry, teamWorkMCPBatchStore(pgStores.MCP), pgStores.BuiltinTools, pgStores.BuiltinToolTenantCfgs, toolPE, toolsReg)
 
 	// Phase 3: Agent hooks RPC methods (hooks.list/create/update/delete/toggle/test/history).
 	if hs, ok := pgStores.Hooks.(hooks.HookStore); ok && hs != nil {
@@ -965,7 +997,7 @@ func runGateway() {
 	registerConfigChannels(cfg, channelMgr, msgBus, pgStores, instanceLoader, audioMgr)
 
 	// Register channels/instances/links/teams RPC methods
-	chInstancesM := wireChannelRPCMethods(server, pgStores, channelMgr, instanceLoader, agentRouter, msgBus, cfg, workspace)
+	chInstancesM := wireChannelRPCMethods(server, pgStores, channelMgr, instanceLoader, agentRouter, msgBus, cfg, workspace, postTurn)
 
 	// Bitrix24 orphan-bot cleaner. Fires from channel_instances delete handler
 	// when the channel is no longer loaded in the Manager (typical scenario:
@@ -1042,6 +1074,17 @@ func runGateway() {
 	// Inbound message consumer setup
 	consumerTeamStore := pgStores.Teams
 
+	// The shared per-tenant Team Work resolver was created before wireExtras so
+	// coordinator recovery and inbound classification read the same tenant cache.
+	if chatMethods != nil {
+		chatMethods.SetTeamWorkConfigResolver(deps.teamWorkCfg)
+	}
+	// Process-local cache invalidation (Phase 7 review 7B-H3): drop a tenant's
+	// cached Team Work settings when its system_configs change. Registered on its
+	// own bus subscriber ID so it coexists with the shared-config refresh
+	// subscriber rather than overwriting it.
+	registerTeamWorkConfigInvalidator(msgBus, deps.teamWorkCfg)
+
 	// Quota checker: enforces per-user/group request limits.
 	config.MergeChannelGroupQuotas(cfg)
 	var quotaChecker *channels.QuotaChecker
@@ -1098,6 +1141,7 @@ func runGateway() {
 		auditCh:           auditCh,
 		sigCh:             sigCh,
 		terminateProcess:  os.Exit,
+		chatMethods:       chatMethods,
 	})
 }
 

cmd/gateway_announce_routing_test.go

@@ -1,8 +1,13 @@
 package cmd
 
 import (
+	"context"
 	"testing"
 
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/config"
+	"github.com/nextlevelbuilder/goclaw/internal/store"
 	"github.com/nextlevelbuilder/goclaw/internal/tools"
 )
 
@@ -18,7 +23,7 @@ import (
 // skips WithSenderID and the Lead's resume has empty sender attribution.
 func TestAnnounceRouting_PropagatesSenderAndRole(t *testing.T) {
 	const (
-		realSender = "5218954741"   // Telegram numeric user id
+		realSender = "5218954741" // Telegram numeric user id
 		realRole   = "admin"
 		realUserID = "group:telegram:-1003812294018"
 	)
@@ -72,3 +77,29 @@ func TestAnnounceRouting_EmptyMetaPropagatesEmpty(t *testing.T) {
 		t.Errorf("OriginRole = %q, want empty", r.OriginRole)
 	}
 }
+
+type announceRoutingAgentStore struct {
+	store.AgentStore
+	agent *store.AgentData
+}
+
+func (s *announceRoutingAgentStore) GetByID(context.Context, uuid.UUID) (*store.AgentData, error) {
+	return s.agent, nil
+}
+
+func TestResolveTeammateLeadAgentUsesLeaderAgentIDBeforeFromAgent(t *testing.T) {
+	leadID := uuid.MustParse("20000000-0000-0000-0000-000000000001")
+	got := resolveTeammateLeadAgent(context.Background(), nil, map[string]string{
+		tools.MetaLeaderAgentID: leadID.String(),
+		tools.MetaFromAgent:     "khanh-developer",
+	}, &ConsumerDeps{
+		Cfg: &config.Config{},
+		AgentStore: &announceRoutingAgentStore{agent: &store.AgentData{
+			BaseModel: store.BaseModel{ID: leadID},
+			AgentKey:  "bao-an",
+		}},
+	})
+	if got != "bao-an" {
+		t.Fatalf("expected lead agent bao-an from leader_agent_id, got %q", got)
+	}
+}

cmd/gateway_channels_setup.go

@@ -25,6 +25,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/gateway/methods"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
 	"github.com/nextlevelbuilder/goclaw/internal/systemmessages"
+	"github.com/nextlevelbuilder/goclaw/internal/tools"
 	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
 )
 
@@ -149,7 +150,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
 // Returns the channel-instances methods handler so the caller can register
 // per-channel-type orphan cleaners (e.g. Bitrix24 imbot.unregister) after
 // per-channel dependencies (portal store, encryption key) are in scope.
-func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, channelMgr *channels.Manager, instanceLoader *channels.InstanceLoader, agentRouter *agent.Router, msgBus *bus.MessageBus, cfg *config.Config, dataDir string) *methods.ChannelInstancesMethods {
+func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, channelMgr *channels.Manager, instanceLoader *channels.InstanceLoader, agentRouter *agent.Router, msgBus *bus.MessageBus, cfg *config.Config, dataDir string, postTurn tools.PostTurnProcessor) *methods.ChannelInstancesMethods {
 	// Register channels RPC methods (after channelMgr is initialized with all channels)
 	methods.NewChannelsMethods(channelMgr).Register(server.Router())
 	methods.NewChatBehaviorMethods(cfg, channelMgr).Register(server.Router())
@@ -171,7 +172,12 @@ func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, chann
 
 	// Register agent teams WS RPC methods
 	if pgStores.Teams != nil {
-		methods.NewTeamsMethods(pgStores.Teams, pgStores.Agents, pgStores.AgentLinks, agentRouter, msgBus, msgBus, dataDir).Register(server.Router())
+		teamsMethods := methods.NewTeamsMethods(pgStores.Teams, pgStores.Agents, pgStores.AgentLinks, agentRouter, msgBus, msgBus, dataDir)
+		teamsMethods.SetPostTurnProcessor(postTurn)
+		if provider, ok := postTurn.(tools.WorkflowActionServiceProvider); ok {
+			teamsMethods.SetWorkflowActionService(provider.WorkflowActionService())
+		}
+		teamsMethods.Register(server.Router())
 	}
 
 	return chInstancesM

cmd/gateway_consumer.go

@@ -15,10 +15,11 @@ 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/skills"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
+	"github.com/nextlevelbuilder/goclaw/internal/teamworkconfig"
 	"github.com/nextlevelbuilder/goclaw/internal/tools"
 	usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps"
 	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
@@ -28,7 +29,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, 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) {
+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, skillsLoader *skills.Loader, mcpStore store.MCPAgentGrantBatchStore, builtinToolStore store.BuiltinToolStore, tenantToolStore store.BuiltinToolTenantConfigStore, toolPolicy *tools.PolicyEngine, toolRegistry *tools.Registry, teamWorkCfg *teamworkconfig.Resolver) {
 	slog.Info("inbound message consumer started")
 
 	// Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts).
@@ -63,8 +64,14 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 		SubagentMgr:      subagentMgr,
 		UsageCaps:        usageCapSvc,
 		ProviderReg:      providerReg,
-		TeamWorkEmbedder: teamWorkEmbedder,
+		SkillsLoader:     skillsLoader,
+		MCPStore:         mcpStore,
+		BuiltinToolStore: builtinToolStore,
+		TenantToolStore:  tenantToolStore,
+		ToolPolicy:       toolPolicy,
+		ToolRegistry:     toolRegistry,
 		GetAnnounceMu:    getAnnounceMu,
+		TeamWorkCfg:      teamWorkCfg,
 	}
 
 	// Track running teammate tasks so they can be cancelled when the task is

cmd/gateway_consumer_deps.go

@@ -7,10 +7,11 @@ 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/skills"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
+	"github.com/nextlevelbuilder/goclaw/internal/teamworkconfig"
 	"github.com/nextlevelbuilder/goclaw/internal/tools"
 	usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps"
 )
@@ -34,7 +35,17 @@ type ConsumerDeps struct {
 	SubagentMgr      *tools.SubagentManager
 	UsageCaps        *usagecaps.Service
 	ProviderReg      *providers.Registry
-	TeamWorkEmbedder memory.EmbeddingProvider
+	SkillsLoader     *skills.Loader
+	MCPStore         store.MCPAgentGrantBatchStore
+	BuiltinToolStore store.BuiltinToolStore
+	TenantToolStore  store.BuiltinToolTenantConfigStore
+	ToolPolicy       *tools.PolicyEngine
+	ToolRegistry     *tools.Registry
 	BgWg             sync.WaitGroup
 	GetAnnounceMu    func(string) *sync.Mutex
+	// TeamWorkCfg resolves per-tenant Team Work classifier settings for the
+	// inbound ingress. Shared with the WS surface (same *teamworkconfig.Resolver
+	// instance) so cache invalidation is coherent. When nil, the inbound gate
+	// falls back to Cfg's file-config values (pre-isolation behavior).
+	TeamWorkCfg *teamworkconfig.Resolver
 }