From 89e457012670e7b9055406500fae60137071fdd2 Mon Sep 17 00:00:00 2001 From: lxowalle Date: Mon, 13 Apr 2026 15:05:02 +0800 Subject: [PATCH] feat(agent): handle /btw as a side question outside session history --- pkg/agent/loop.go | 138 +++++++++++++++++++++++++++++++++++ pkg/agent/loop_test.go | 51 +++++++++++++ pkg/agent/steering_test.go | 108 +++++++++++++++++++++++++++ pkg/commands/builtin.go | 1 + pkg/commands/builtin_test.go | 52 +++++++++++++ pkg/commands/cmd_btw.go | 37 ++++++++++ pkg/commands/runtime.go | 7 +- 7 files changed, 393 insertions(+), 1 deletion(-) create mode 100644 pkg/commands/cmd_btw.go diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index ac230aa86..97e5dd7a7 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -641,6 +641,13 @@ 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) + if handled, response := al.tryHandlePriorityCommand(ctx, msg); handled { + if response != "" { + al.PublishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, response) + } + continue + } + logger.InfoCF("agent", "Redirecting inbound message to steering queue", map[string]any{ "channel": msg.Channel, @@ -1339,6 +1346,95 @@ func (al *AgentLoop) ProcessHeartbeat( }) } +func (al *AgentLoop) askSideQuestion( + ctx context.Context, + agent *AgentInstance, + opts *processOptions, + question string, +) (string, error) { + if agent == nil { + return "", fmt.Errorf("no agent available for /btw") + } + + question = strings.TrimSpace(question) + if question == "" { + return "", fmt.Errorf("Usage: /btw ") + } + + var channel, chatID, senderID, senderDisplayName string + var activeSkills []string + if opts != nil { + channel = opts.Channel + chatID = opts.ChatID + senderID = opts.SenderID + senderDisplayName = opts.SenderDisplayName + activeSkills = activeSkillNames(agent, *opts) + } + + messages := agent.ContextBuilder.BuildMessages( + nil, + "", + question, + nil, + channel, + chatID, + senderID, + senderDisplayName, + activeSkills..., + ) + + activeCandidates, activeModel, usedLight := al.selectCandidates(agent, question, messages) + activeProvider := agent.Provider + if usedLight && agent.LightProvider != nil { + activeProvider = agent.LightProvider + } + + llmOpts := map[string]any{ + "max_tokens": agent.MaxTokens, + "temperature": agent.Temperature, + "prompt_cache_key": agent.ID + ":btw", + } + if agent.ThinkingLevel != ThinkingOff { + if tc, ok := activeProvider.(providers.ThinkingCapable); ok && tc.SupportsThinking() { + llmOpts["thinking_level"] = string(agent.ThinkingLevel) + } + } + + callProvider := func(ctx context.Context, provider providers.LLMProvider, model string) (*providers.LLMResponse, error) { + return provider.Chat(ctx, messages, nil, model, llmOpts) + } + + if len(activeCandidates) > 1 && al.fallback != nil { + fbResult, err := al.fallback.Execute( + ctx, + activeCandidates, + func(ctx context.Context, providerName, model string) (*providers.LLMResponse, error) { + candidateProvider := activeProvider + if cp, ok := agent.CandidateProviders[providers.ModelKey(providerName, model)]; ok { + candidateProvider = cp + } + return callProvider(ctx, candidateProvider, model) + }, + ) + if err != nil { + return "", err + } + if fbResult.Response == nil { + return "", nil + } + return fbResult.Response.Content, nil + } + + resp, err := callProvider(ctx, activeProvider, activeModel) + if err != nil { + return "", err + } + if resp == nil { + return "", nil + } + return resp.Content, nil +} + func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { // Add message preview to log (show full content for error messages) var logContent string @@ -3490,6 +3586,14 @@ func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance, opts *processOpt if agent.ContextBuilder != nil { rt.ListSkillNames = agent.ContextBuilder.ListSkillNames } + rt.AskSideQuestion = func(ctx context.Context, question string) (string, error) { + question = strings.TrimSpace(question) + if question == "" { + return "", fmt.Errorf("Usage: /btw ") + } + + return al.askSideQuestion(ctx, agent, opts, question) + } rt.GetModelInfo = func() (string, string) { return agent.Model, resolvedCandidateProvider(agent.Candidates, cfg.Agents.Defaults.Provider) } @@ -3616,6 +3720,40 @@ 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 { + return true, fmt.Sprintf("Error processing message: %v", err) + } + return true, "Command unavailable in current context." + } + + opts := processOptions{ + SessionKey: resolveScopeKey(route, msg.SessionKey), + Channel: msg.Channel, + ChatID: msg.ChatID, + MessageID: msg.MessageID, + ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage), + SenderID: msg.SenderID, + SenderDisplayName: msg.Sender.DisplayName, + UserMessage: msg.Content, + Media: msg.Media, + DefaultResponse: defaultResponse, + } + + response, handled := al.handleCommand(ctx, msg, agent, &opts) + if !handled { + return false, "" + } + return true, response +} + // extractPeer extracts the routing peer from the inbound message's structured Peer field. func extractPeer(msg bus.InboundMessage) *routing.RoutePeer { if msg.Peer.Kind == "" { diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index a67c8d040..7e6f89b05 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -228,6 +228,57 @@ func TestProcessMessage_UseCommandLoadsRequestedSkill(t *testing.T) { } } +func TestProcessMessage_BtwCommandRunsWithoutPersistingHistory(t *testing.T) { + tmpDir := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &recordingProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + + msg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "telegram:123", + ChatID: "chat-1", + Content: "/btw explain side effects", + } + + response, err := al.processMessage(context.Background(), msg) + if err != nil { + t.Fatalf("processMessage() error = %v", err) + } + if response != "Mock response" { + t.Fatalf("processMessage() response = %q, want %q", response, "Mock response") + } + if len(provider.lastMessages) == 0 { + t.Fatal("provider did not receive any messages") + } + + lastMessage := provider.lastMessages[len(provider.lastMessages)-1] + if lastMessage.Role != "user" || lastMessage.Content != "explain side effects" { + t.Fatalf("last provider message = %+v, want stripped /btw question", lastMessage) + } + + route, _, err := al.resolveMessageRoute(msg) + if err != nil { + t.Fatalf("resolveMessageRoute() error = %v", err) + } + sessionKey := resolveScopeKey(route, msg.SessionKey) + history := al.GetRegistry().GetDefaultAgent().Sessions.GetHistory(sessionKey) + if len(history) != 0 { + t.Fatalf("session history len = %d, want 0 for /btw", len(history)) + } +} + func TestHandleCommand_UseCommandRejectsUnknownSkill(t *testing.T) { tmpDir := t.TempDir() cfg := &config.Config{ diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 75ba9861d..fd4364efc 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -1013,6 +1013,114 @@ 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{ + Channel: "test", + SenderID: "user1", + ChatID: "chat1", + Content: "execute sleep 60, then send OK", + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + btw := bus.InboundMessage{ + Channel: "test", + SenderID: "user1", + ChatID: "chat1", + Content: "/btw 你说283928", + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + + 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") + } + + 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 := routing.BuildAgentMainSessionKey(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(): + if outbound.Content != "long turn finished" { + t.Fatalf("expected original turn response after release, got %q", outbound.Content) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for original turn response") + } + + 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_Continue_PreservesSteeringMedia(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil { diff --git a/pkg/commands/builtin.go b/pkg/commands/builtin.go index 39e76f752..5cf9425cb 100644 --- a/pkg/commands/builtin.go +++ b/pkg/commands/builtin.go @@ -11,6 +11,7 @@ func BuiltinDefinitions() []Definition { showCommand(), listCommand(), useCommand(), + btwCommand(), switchCommand(), checkCommand(), clearCommand(), diff --git a/pkg/commands/builtin_test.go b/pkg/commands/builtin_test.go index 5fd8dd9bc..74fa7246e 100644 --- a/pkg/commands/builtin_test.go +++ b/pkg/commands/builtin_test.go @@ -188,3 +188,55 @@ func TestBuiltinUseCommand_PassthroughsToAgentLogic(t *testing.T) { t.Fatalf("/use command=%q, want=%q", res.Command, "use") } } + +func TestBuiltinBtwCommand_UsesSideQuestionRuntime(t *testing.T) { + rt := &Runtime{ + AskSideQuestion: func(ctx context.Context, question string) (string, error) { + if question != "what is 2+2?" { + t.Fatalf("question=%q, want %q", question, "what is 2+2?") + } + return "4", nil + }, + } + defs := BuiltinDefinitions() + ex := NewExecutor(NewRegistry(defs), rt) + + var reply string + res := ex.Execute(context.Background(), Request{ + Text: "/btw what is 2+2?", + Reply: func(text string) error { + reply = text + return nil + }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("/btw outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if reply != "4" { + t.Fatalf("/btw reply=%q, want=%q", reply, "4") + } +} + +func TestBuiltinBtwCommand_MissingQuestion(t *testing.T) { + defs := BuiltinDefinitions() + ex := NewExecutor(NewRegistry(defs), &Runtime{ + AskSideQuestion: func(context.Context, string) (string, error) { + return "", nil + }, + }) + + var reply string + res := ex.Execute(context.Background(), Request{ + Text: "/btw", + Reply: func(text string) error { + reply = text + return nil + }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("/btw outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if reply != "Usage: /btw " { + t.Fatalf("/btw reply=%q, want usage message", reply) + } +} diff --git a/pkg/commands/cmd_btw.go b/pkg/commands/cmd_btw.go new file mode 100644 index 000000000..cb2c7b222 --- /dev/null +++ b/pkg/commands/cmd_btw.go @@ -0,0 +1,37 @@ +package commands + +import ( + "context" + "strings" +) + +func btwCommand() Definition { + return Definition{ + Name: "btw", + Description: "Ask a side question without changing session history", + Usage: "/btw ", + Handler: func(ctx context.Context, req Request, rt *Runtime) error { + const emptyAnswerMsg = "The model returned an empty response. This may indicate a provider error or token limit." + + if rt == nil || rt.AskSideQuestion == nil { + return req.Reply(unavailableMsg) + } + + question := strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(req.Text), nthToken(req.Text, 0))) + if question == "" { + return req.Reply("Usage: /btw ") + } + + answer, err := rt.AskSideQuestion(ctx, question) + if err != nil { + return req.Reply(err.Error()) + } + if strings.TrimSpace(answer) == "" { + return req.Reply(emptyAnswerMsg) + } + + return req.Reply(answer) + + }, + } +} diff --git a/pkg/commands/runtime.go b/pkg/commands/runtime.go index 5ba6a1bd2..69373f561 100644 --- a/pkg/commands/runtime.go +++ b/pkg/commands/runtime.go @@ -1,6 +1,10 @@ package commands -import "github.com/sipeed/picoclaw/pkg/config" +import ( + "context" + + "github.com/sipeed/picoclaw/pkg/config" +) // Runtime provides runtime dependencies to command handlers. It is constructed // per-request by the agent loop so that per-request state (like session scope) @@ -8,6 +12,7 @@ import "github.com/sipeed/picoclaw/pkg/config" type Runtime struct { Config *config.Config GetModelInfo func() (name, provider string) + AskSideQuestion func(ctx context.Context, question string) (string, error) ListAgentIDs func() []string ListDefinitions func() []Definition ListSkillNames func() []string