diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 96a43ab08..00a38761c 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -653,6 +653,18 @@ func (al *AgentLoop) drainBusToSteering(ctx context.Context, activeScope, active // Transcribe audio if needed before steering, so the agent sees text. msg, _ = al.transcribeAudioInMessage(ctx, msg) + // Handle priority commands (e.g. /btw) immediately instead of queueing them. + if handled, response := al.tryHandlePriorityCommand(ctx, msg); handled { + if response != "" { + al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + Channel: msg.Channel, + ChatID: msg.ChatID, + Content: response, + }) + } + continue + } + logger.InfoCF("agent", "Redirecting inbound message to steering queue", map[string]any{ "channel": msg.Channel, @@ -4047,6 +4059,39 @@ func mapCommandError(result commands.ExecuteResult) string { return fmt.Sprintf("Failed to execute /%s: %v", result.Command, result.Err) } +func (al *AgentLoop) tryHandlePriorityCommand(ctx context.Context, msg bus.InboundMessage) (bool, string) { + cmdName, ok := commands.CommandName(msg.Content) + if !ok || cmdName != "btw" { + return false, "" + } + + route, agent, err := al.resolveMessageRoute(msg) + if err != nil || agent == nil { + if err != nil { + logger.ErrorCF("agent", fmt.Sprintf("Error resolving route for /btw: %v", err), nil) + return true, fmt.Sprintf("Error processing message: %v", err) + } + logger.WarnCF("agent", "/btw command unavailable: no agent resolved", nil) + return true, "Command unavailable in current context." + } + + allocation := al.allocateRouteSession(route, msg) + opts := processOptions{ + SessionKey: resolveScopeKey(allocation.SessionKey, msg.SessionKey), + Channel: msg.Channel, + ChatID: msg.ChatID, + SenderID: msg.SenderID, + SenderDisplayName: msg.Sender.DisplayName, + Media: msg.Media, + } + + response, handled := al.handleCommand(ctx, msg, agent, &opts) + if !handled { + return false, "" + } + return true, response +} + // isNativeSearchProvider reports whether the given LLM provider implements // NativeSearchCapable and returns true for SupportsNativeSearch. func isNativeSearchProvider(p providers.LLMProvider) bool { diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 8e6063f08..a414a211b 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -1010,6 +1010,139 @@ func TestAgentLoop_Steering_DirectResponseContinuesWithQueuedMessage(t *testing. } } +func TestAgentLoop_Steering_BtwCommandBypassesQueuedTurn(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + provider := &blockingDirectProvider{ + firstStarted: make(chan struct{}), + releaseFirst: make(chan struct{}), + firstResp: "long turn finished", + finalResp: "btw immediate reply", + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, provider) + + runCtx, cancelRun := context.WithCancel(context.Background()) + defer cancelRun() + runErrCh := make(chan error, 1) + go func() { + runErrCh <- al.Run(runCtx) + }() + + first := bus.InboundMessage{ + Context: bus.InboundContext{ + Channel: "test", + ChatID: "chat1", + ChatType: "direct", + SenderID: "user1", + }, + Content: "execute sleep 60, then send OK", + } + btw := bus.InboundMessage{ + Context: bus.InboundContext{ + Channel: "test", + ChatID: "chat1", + ChatType: "direct", + SenderID: "user1", + }, + Content: "/btw what is the current progress?", + } + + pubCtx, pubCancel := context.WithTimeout(context.Background(), 2*time.Second) + defer pubCancel() + if err := msgBus.PublishInbound(pubCtx, first); err != nil { + t.Fatalf("publish first inbound: %v", err) + } + + select { + case <-provider.firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for first LLM call to start") + } + + messageTool, ok := al.GetRegistry().GetDefaultAgent().Tools.Get("message") + var mt *tools.MessageTool + if !ok { + mt = tools.NewMessageTool() + al.RegisterTool(mt) + } else { + var typeOK bool + mt, typeOK = messageTool.(*tools.MessageTool) + if !typeOK { + t.Fatal("expected message tool type") + } + } + mt.SetSendCallback(func(ctx context.Context, channel, chatID, content, replyToMessageID string) error { + return nil + }) + if result := mt.Execute(context.Background(), map[string]any{ + "channel": "test", + "chat_id": "chat1", + "content": "already sent from busy turn", + }); result == nil || result.IsError { + t.Fatalf("message tool setup result = %+v, want successful send", result) + } + + if err := msgBus.PublishInbound(pubCtx, btw); err != nil { + t.Fatalf("publish /btw inbound: %v", err) + } + + select { + case outbound := <-msgBus.OutboundChan(): + if outbound.Content != "btw immediate reply" { + t.Fatalf("expected /btw reply before long turn completion, got %q", outbound.Content) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for /btw outbound response") + } + + sessionKey := session.BuildMainSessionKey(routing.DefaultAgentID) + if msgs := al.dequeueSteeringMessagesForScope(sessionKey); len(msgs) != 0 { + t.Fatalf("expected /btw to bypass steering queue, got %v", msgs) + } + + close(provider.releaseFirst) + + select { + case outbound := <-msgBus.OutboundChan(): + t.Fatalf("expected busy turn final response to stay suppressed, got %q", outbound.Content) + case <-time.After(2 * time.Second): + } + + provider.mu.Lock() + callCount := provider.calls + provider.mu.Unlock() + if callCount != 2 { + t.Fatalf("provider call count = %d, want 2", callCount) + } + + cancelRun() + select { + case err := <-runErrCh: + if err != nil { + t.Fatalf("Run returned error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for Run to stop") + } +} + func TestAgentLoop_AgentForSession_UsesStoredScopeMetadata(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil {