mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-15 16:14:43 +00:00
fix(teams): inject PendingTeamDispatch in WS chat.send path
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
This commit is contained in:
1 parent
32ced98f6a
commit
be770252e6
4 files changed
+44
-4
No files matched your search
+6
-1
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in new issue
Block a user