From be770252e6206e3a46f3edaf962d8c56292e89b0 Mon Sep 17 00:00:00 2001 From: viettranx Date: Fri, 20 Mar 2026 23:03:59 +0700 Subject: [PATCH] fix(teams): inject PendingTeamDispatch in WS chat.send path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WS handleSend called loop.Run() directly without injecting PendingTeamDispatch into context, bypassing the gate that requires search/list before team task creation (ptd was nil → gate skipped). - Add InjectTeamDispatch helper: creates ptd, returns drain func that dispatches pending tasks + releases team lock (panic-safe) - Wire PostTurnProcessor to ChatMethods via setter - defer drainTeamDispatch() in handleSend goroutine --- cmd/gateway.go | 7 ++++++- cmd/gateway_methods.go | 7 ++++--- internal/gateway/methods/chat.go | 12 ++++++++++++ internal/tools/context_keys.go | 22 ++++++++++++++++++++++ 4 files changed, 44 insertions(+), 4 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index 406094cc..e8e6c43f 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -421,7 +421,12 @@ func runGateway() { // Register all RPC methods server.SetLogTee(logTee) - pairingMethods, heartbeatMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions) + pairingMethods, heartbeatMethods, chatMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions) + + // Wire post-turn processor for WS chat.send team task dispatch. + if postTurn != nil { + chatMethods.SetPostTurnProcessor(postTurn) + } // Wire pairing event broadcasts to all WS clients. pairingMethods.SetBroadcaster(server.BroadcastEvent) diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go index 7d40d754..9749a114 100644 --- a/cmd/gateway_methods.go +++ b/cmd/gateway_methods.go @@ -12,11 +12,12 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/tools" ) -func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, 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) (*methods.PairingMethods, *methods.HeartbeatMethods) { +func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, 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) (*methods.PairingMethods, *methods.HeartbeatMethods, *methods.ChatMethods) { router := server.Router() // Phase 1: Core methods - methods.NewChatMethods(agents, sessStore, server.RateLimiter(), msgBus).Register(router) + chatMethods := methods.NewChatMethods(agents, sessStore, server.RateLimiter(), msgBus) + chatMethods.Register(router) methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, contextFileInterceptor, msgBus).Register(router) methods.NewSessionsMethods(sessStore, msgBus).Register(router) methods.NewConfigMethods(cfg, cfgPath, configSecretsStore, msgBus).Register(router) @@ -61,5 +62,5 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore "phase2", []string{"skills", "cron", "heartbeat", "pairing", "usage", "exec_approval", "send"}, ) - return pairingMethods, heartbeatMethods + return pairingMethods, heartbeatMethods, chatMethods } diff --git a/internal/gateway/methods/chat.go b/internal/gateway/methods/chat.go index 8bb3013b..d670c6a3 100644 --- a/internal/gateway/methods/chat.go +++ b/internal/gateway/methods/chat.go @@ -16,6 +16,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/sessions" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/tools" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) @@ -25,12 +26,18 @@ type ChatMethods struct { sessions store.SessionStore rateLimiter *gateway.RateLimiter eventBus bus.EventPublisher + postTurn tools.PostTurnProcessor } func NewChatMethods(agents *agent.Router, sess store.SessionStore, rl *gateway.RateLimiter, eventBus bus.EventPublisher) *ChatMethods { return &ChatMethods{agents: agents, sessions: sess, rateLimiter: rl, eventBus: eventBus} } +// SetPostTurnProcessor sets the post-turn processor for team task dispatch. +func (m *ChatMethods) SetPostTurnProcessor(pt tools.PostTurnProcessor) { + m.postTurn = pt +} + // Register adds chat methods to the router. func (m *ChatMethods) Register(router *gateway.MethodRouter) { router.Register(protocol.MethodChatSend, m.handleSend) @@ -150,6 +157,10 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re // Fallback: injection failed (channel full), proceed with new run } + // Inject team dispatch tracker: gates team_tasks create (must search/list first) + // and defers task dispatch to post-turn. + runCtxBase, drainTeamDispatch := tools.InjectTeamDispatch(runCtxBase, m.postTurn) + // Create cancellable context for abort support (matching TS AbortController pattern). runCtx, cancel := context.WithCancel(runCtxBase) injectCh := m.agents.RegisterRun(runID, sessionKey, params.AgentID, cancel) @@ -158,6 +169,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re go func() { defer m.agents.UnregisterRun(runID) defer cancel() + defer drainTeamDispatch() // dispatch pending team tasks + release lock (even on panic) // Parse media items (supports both legacy string paths and new {path,filename} objects). items := params.parseMedia() diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index 89db61fb..e38240c6 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -2,6 +2,7 @@ package tools import ( "context" + "log/slog" "sync" "github.com/google/uuid" @@ -383,6 +384,27 @@ func PendingTeamDispatchFromCtx(ctx context.Context) *PendingTeamDispatch { return v } +// InjectTeamDispatch creates a fresh PendingTeamDispatch context for a direct +// loop.Run() call (WS chat.send, HTTP API, etc.) and returns a drain function +// that must be called after the run completes. The drain function dispatches +// any pending team tasks via the provided PostTurnProcessor. It is safe to +// call even if no tasks were created. Pass nil postTurn if not available. +func InjectTeamDispatch(ctx context.Context, postTurn PostTurnProcessor) (context.Context, func()) { + ptd := NewPendingTeamDispatch() + ctx = WithPendingTeamDispatch(ctx, ptd) + drain := func() { + ptd.ReleaseTeamLock() + if postTurn != nil { + for teamID, taskIDs := range ptd.Drain() { + if err := postTurn.ProcessPendingTasks(context.Background(), teamID, taskIDs); err != nil { + slog.Warn("post_turn: dispatch failed", "team_id", teamID, "error", err) + } + } + } + } + return ctx, drain +} + // --- Run media file paths (for team workspace auto-collect) --- const ctxRunMediaPaths toolContextKey = "tool_run_media_paths"