fix: wire DelegateManager into /stopall to cancel async delegations

DelegateAsync creates detached contexts (context.Background()) and runs
outside the scheduler, so /stopall's CancelSession() couldn't reach them.
Now /stopall also calls DelegateManager.CancelForOrigin() to cancel
active delegate tasks matching the origin channel+chatID.

Also silences context.Canceled errors in subagent/delegate announce
handlers to avoid sending spurious error messages after cancellation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
viettranxandClaude Opus 4.6 committed 2026-03-04 22:17:22 +07:00
1 parent b8afe6a899
commit 1841a7097c
4 files changed
+47 -6

No files matched your search

+3 -2
View File
@@ -614,6 +614,7 @@ func runGateway() {
// Declared here so it can be passed to registerAllMethods → AgentsMethods
// for immediate cache invalidation on agents.files.set.
var contextFileInterceptor *tools.ContextFileInterceptor
var delegateMgr *tools.DelegateManager
// Managed mode: set agent store for tools_invoke context injection + wire extras
if managedStores != nil && managedStores.Agents != nil {
@@ -629,7 +630,7 @@ func runGateway() {
}
}
contextFileInterceptor = wireManagedExtras(managedStores, agentRouter, providerRegistry, msgBus, sessStore, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, dynamicLoader)
contextFileInterceptor, delegateMgr = wireManagedExtras(managedStores, agentRouter, providerRegistry, msgBus, sessStore, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, dynamicLoader)
agentsH, skillsH, tracesH, mcpH, customToolsH, channelInstancesH, providersH, delegationsH, builtinToolsH := wireManagedHTTP(managedStores, cfg.Gateway.Token, msgBus, toolsReg, providerRegistry, permPE.IsOwner)
if agentsH != nil {
server.SetAgentsHandler(agentsH)
@@ -959,7 +960,7 @@ func runGateway() {
webFetchTool.UpdatePolicy(updatedCfg.Tools.WebFetch.Policy, updatedCfg.Tools.WebFetch.AllowedDomains, updatedCfg.Tools.WebFetch.BlockedDomains)
})
go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker)
go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, delegateMgr)
go func() {
sig := <-sigCh
+17 -1
View File
@@ -17,6 +17,7 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
"github.com/nextlevelbuilder/goclaw/internal/sessions"
"github.com/nextlevelbuilder/goclaw/internal/store"
"github.com/nextlevelbuilder/goclaw/internal/tools"
)
// makeSchedulerRunFunc creates the RunFunc for the scheduler.
@@ -41,7 +42,7 @@ func makeSchedulerRunFunc(agents *agent.Router, cfg *config.Config) scheduler.Ru
// 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) {
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, delegateMgr *tools.DelegateManager) {
slog.Info("inbound message consumer started")
// Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts).
@@ -412,6 +413,10 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
go func(origCh, chatID, senderID, label string, meta map[string]string) {
outcome := <-outCh
if outcome.Err != nil {
if errors.Is(outcome.Err, context.Canceled) {
slog.Info("subagent announce: run cancelled", "subagent", senderID)
return
}
slog.Error("subagent announce: agent run failed", "error", outcome.Err)
msgBus.PublishOutbound(bus.OutboundMessage{
Channel: origCh,
@@ -519,6 +524,10 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
go func(origCh, chatID, senderID string, meta map[string]string) {
outcome := <-outCh
if outcome.Err != nil {
if errors.Is(outcome.Err, context.Canceled) {
slog.Info("delegate announce: run cancelled", "delegation", senderID)
return
}
slog.Error("delegate announce: agent run failed", "error", outcome.Err)
msgBus.PublishOutbound(bus.OutboundMessage{
Channel: origCh,
@@ -719,6 +728,13 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
var cancelled bool
if cmd == "stopall" {
cancelled = sched.CancelSession(sessionKey)
// Also cancel async delegations for this chat (they bypass the scheduler)
if delegateMgr != nil {
dc := delegateMgr.CancelForOrigin(msg.Channel, msg.ChatID)
if dc > 0 {
cancelled = true
}
}
slog.Info("inbound: /stopall command", "session", sessionKey, "cancelled", cancelled)
} else {
cancelled = sched.CancelOneSession(sessionKey)
+4 -3
View File
@@ -43,9 +43,10 @@ func wireManagedExtras(
appCfg *config.Config,
sandboxMgr sandbox.Manager,
dynamicLoader *tools.DynamicToolLoader,
) *tools.ContextFileInterceptor {
) (*tools.ContextFileInterceptor, *tools.DelegateManager) {
// 1. Context file interceptor (created before resolver so callbacks can reference it)
var contextFileInterceptor *tools.ContextFileInterceptor
var delegateMgr *tools.DelegateManager
if stores.Agents != nil {
contextFileInterceptor = tools.NewContextFileInterceptor(stores.Agents, workspace)
}
@@ -323,7 +324,7 @@ func wireManagedExtras(
}
return dr, nil
}
delegateMgr := tools.NewDelegateManager(runAgentFn, stores.AgentLinks, stores.Agents, msgBus)
delegateMgr = tools.NewDelegateManager(runAgentFn, stores.AgentLinks, stores.Agents, msgBus)
if stores.Teams != nil {
delegateMgr.SetTeamStore(stores.Teams)
}
@@ -435,7 +436,7 @@ func wireManagedExtras(
}
slog.Info("managed mode: resolver + interceptors + cache subscribers wired")
return contextFileInterceptor
return contextFileInterceptor, delegateMgr
}
// wireManagedHTTP creates managed-mode HTTP handlers (agents + skills + traces + MCP + custom tools + channel instances + providers + delegations + builtin tools).
+23
View File
@@ -12,6 +12,29 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// CancelForOrigin cancels all active async delegations originating from a given channel + chatID.
// Used by /stopall to stop delegate tasks that bypass the scheduler.
func (dm *DelegateManager) CancelForOrigin(channel, chatID string) int {
count := 0
dm.active.Range(func(key, val any) bool {
t := val.(*DelegationTask)
if t.Status == "running" && t.OriginChannel == channel && t.OriginChatID == chatID {
if t.cancelFunc != nil {
t.cancelFunc()
}
t.Status = "cancelled"
now := time.Now()
t.CompletedAt = &now
dm.active.Delete(key)
dm.emitEvent("delegation.cancelled", t)
slog.Info("delegation cancelled by /stopall", "id", t.ID, "target", t.TargetAgentKey)
count++
}
return true
})
return count
}
// Cancel cancels a running delegation by ID.
func (dm *DelegateManager) Cancel(delegationID string) bool {
val, ok := dm.active.Load(delegationID)