diff --git a/cmd/gateway.go b/cmd/gateway.go index 2eee6892..3b5745c0 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -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 diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index f83105e4..4e3a51b8 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -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) diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 4ddf7452..eb23839e 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -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). diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index 8ea08021..4aadfb92 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -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)