diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index dafecf3b3..6ffbc89f1 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -347,17 +347,17 @@ func (al *AgentLoop) Run(ctx context.Context) error { <-workerDone }() + inbound := al.bus.InboundChan() for al.running.Load() { + var msg bus.InboundMessage select { case <-ctx.Done(): return nil - default: - } - - msg, ok := al.bus.ConsumeInbound(ctx) - - if !ok { - continue + case m, ok := <-inbound: + if !ok { + return nil + } + msg = m } // Echo commands sent from the Mini App so the user can see what was sent. @@ -375,8 +375,8 @@ func (al *AgentLoop) Run(ctx context.Context) error { } // Fast path: handle slash commands immediately without blocking the LLM worker. - - if response, handled := al.handleCommand(ctx, msg); handled { + defaultAgent := al.registry.GetDefaultAgent() + if response, handled := al.handleCommand(ctx, msg, defaultAgent, msg.SessionKey); handled { if response != "" { _ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{ Channel: msg.Channel, @@ -496,6 +496,14 @@ func (al *AgentLoop) llmWorker(ctx context.Context, queue <-chan bus.InboundMess // llmWorkerNormal processes a single non-PDF message. func (al *AgentLoop) llmWorkerNormal(ctx context.Context, msg bus.InboundMessage) { + al.activeRequests.Add(1) + defer al.activeRequests.Done() + + // Ensure typing indicator is stopped when processing completes. + if al.channelManager != nil { + defer al.channelManager.InvokeTypingStop(msg.Channel, msg.ChatID) + } + // Reset per-round message-tool state so a previous round's // tool-sent flag does not suppress this round's response. if defaultAgent := al.registry.GetDefaultAgent(); defaultAgent != nil { @@ -1041,6 +1049,54 @@ func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, cha }) } +// resolveMessageRoute resolves the agent and routing info for an inbound message. +// It looks up the agent registry to determine which agent handles the message +// and resets the message tool context for the new round. +func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) { + registry := al.GetRegistry() + route := registry.ResolveRoute(routing.RouteInput{ + Channel: msg.Channel, + AccountID: msg.Metadata[metadataKeyAccountID], + Peer: extractPeer(msg), + ParentPeer: extractParentPeer(msg), + GuildID: msg.Metadata[metadataKeyGuildID], + TeamID: msg.Metadata[metadataKeyTeamID], + }) + + agent, ok := registry.GetAgent(route.AgentID) + if !ok { + agent = registry.GetDefaultAgent() + } + if agent == nil { + return route, nil, fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID) + } + + // Reset message-tool state for this round + if tool, ok := agent.Tools.Get("message"); ok { + if mt, ok := tool.(tools.ContextualTool); ok { + mt.SetContext(msg.Channel, msg.ChatID) + } + } + + logger.InfoCF("agent", "Routed message", + map[string]any{ + "agent_id": agent.ID, + "session_key": route.SessionKey, + "matched_by": route.MatchedBy, + }) + + return route, agent, nil +} + +// resolveScopeKey returns the session key to use: honors a pre-set key (from +// ProcessDirect/cron) over the route-resolved key. +func resolveScopeKey(route routing.ResolvedRoute, msgSessionKey string) string { + if msgSessionKey != "" { + return msgSessionKey + } + return route.SessionKey +} + 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 @@ -1091,62 +1147,18 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) // Expand fork-specific /skill and /plan commands expansionCompact := al.expandForkCommands(&msg) - // Check for commands - - if response, handled := al.handleCommand(ctx, msg); handled { + // Check for commands (using default agent, before routing) + if response, handled := al.handleCommand(ctx, msg, al.registry.GetDefaultAgent(), msg.SessionKey); handled { return response, nil } // Route to determine agent and session key - - registry := al.GetRegistry() - route := registry.ResolveRoute(routing.RouteInput{ - Channel: msg.Channel, - - AccountID: msg.Metadata["account_id"], - - Peer: extractPeer(msg), - - ParentPeer: extractParentPeer(msg), - - GuildID: msg.Metadata["guild_id"], - - TeamID: msg.Metadata["team_id"], - }) - - agent, ok := registry.GetAgent(route.AgentID) - if !ok { - agent = registry.GetDefaultAgent() - } - if agent == nil { - return "", fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID) + route, agent, err := al.resolveMessageRoute(msg) + if err != nil { + return "", err } - // Reset message-tool state for this round so we don't skip publishing due to a previous round. - - if tool, ok := agent.Tools.Get("message"); ok { - if mt, ok := tool.(tools.ContextualTool); ok { - mt.SetContext(msg.Channel, msg.ChatID) - } - } - - // Use routed session key, but honor ANY pre-set session key (for ProcessDirect/cron) - - sessionKey := route.SessionKey - - if msg.SessionKey != "" { - sessionKey = msg.SessionKey - } - - logger.InfoCF("agent", "Routed message", - - map[string]any{ - "agent_id": agent.ID, - - "session_key": sessionKey, - - "matched_by": route.MatchedBy, - }) + sessionKey := resolveScopeKey(route, msg.SessionKey) return al.runAgentLoop(ctx, agent, processOptions{ SessionKey: sessionKey, diff --git a/pkg/agent/loop_commands.go b/pkg/agent/loop_commands.go index 0b51f8b0b..a8dd3846f 100644 --- a/pkg/agent/loop_commands.go +++ b/pkg/agent/loop_commands.go @@ -8,164 +8,136 @@ import ( "time" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/commands" + "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/session" "github.com/sipeed/picoclaw/pkg/stats" ) -func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) (string, bool) { - content := strings.TrimSpace(msg.Content) +// buildCommandsRuntime constructs a commands.Runtime wired to the current +// agent and loop state. This is the upstream pattern for providing runtime +// dependencies to command handlers. +func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance, sessionKey string) *commands.Runtime { + return &commands.Runtime{ + Config: al.GetConfig(), + GetModelInfo: func() (string, string) { + if agent == nil { + return "unknown", "unknown" + } + prov, _ := providers.ExtractProtocol(agent.Model) + return agent.Model, prov + }, + ListAgentIDs: func() []string { + return al.GetRegistry().ListAgentIDs() + }, + ListDefinitions: func() []commands.Definition { + return al.cmdRegistry.Definitions() + }, + GetEnabledChannels: func() []string { + if al.channelManager == nil { + return nil + } + return al.channelManager.GetEnabledChannels() + }, + SwitchModel: func(value string) (string, error) { + if agent == nil { + return "", fmt.Errorf("no default agent configured") + } + old := agent.Model + agent.Model = value + return old, nil + }, + SwitchChannel: func(value string) error { + if al.channelManager == nil { + return fmt.Errorf("channel manager not initialized") + } + if _, exists := al.channelManager.GetChannel(value); !exists && value != "cli" { + return fmt.Errorf("channel '%s' not found or not enabled", value) + } + return nil + }, + ClearHistory: func() error { + if agent == nil || sessionKey == "" { + return fmt.Errorf("no active session") + } + agent.Sessions.SetHistory(sessionKey, nil) + agent.Sessions.SetSummary(sessionKey, "") + return agent.Sessions.Save(sessionKey) + }, + ReloadConfig: func() error { + if al.reloadFunc != nil { + return al.reloadFunc() + } + return fmt.Errorf("reload not available") + }, + } +} - if !strings.HasPrefix(content, "/") { +// handleCommand processes slash commands. It first tries the upstream +// commands.Executor (for /show, /list, /switch, /check, /clear, /reload, etc.), +// then falls back to fork-specific commands (/session, /skills, /plan, /heartbeat). +func (al *AgentLoop) handleCommand( + ctx context.Context, + msg bus.InboundMessage, + agent *AgentInstance, + sessionKey string, +) (string, bool) { + content := strings.TrimSpace(msg.Content) + if !commands.HasCommandPrefix(content) { return "", false } - parts := strings.Fields(content) + // Build a reply collector — the Executor calls req.Reply with the response. + var response string + replyFn := func(text string) error { + response = text + return nil + } + rt := al.buildCommandsRuntime(agent, sessionKey) + exec := commands.NewExecutor(al.cmdRegistry, rt) + result := exec.Execute(ctx, commands.Request{ + Channel: msg.Channel, + ChatID: msg.ChatID, + SenderID: msg.SenderID, + Text: content, + Reply: replyFn, + }) + + if result.Outcome == commands.OutcomeHandled { + if result.Err != nil { + return fmt.Sprintf("Command error: %v", result.Err), true + } + return response, true + } + + // Fallback: fork-specific commands not in the upstream registry + parts := strings.Fields(content) if len(parts) == 0 { return "", false } - cmd := parts[0] - args := parts[1:] switch cmd { - case "/show": - - if len(args) < 1 { - return "Usage: /show [model|channel|agents]", true - } - - switch args[0] { - case "model": - - defaultAgent := al.registry.GetDefaultAgent() - - if defaultAgent == nil { - return "No default agent configured", true - } - - return fmt.Sprintf("Current model: %s", defaultAgent.Model), true - - case "channel": - - return fmt.Sprintf("Current channel: %s", msg.Channel), true - - case "agents": - - agentIDs := al.registry.ListAgentIDs() - - return fmt.Sprintf("Registered agents: %s", strings.Join(agentIDs, ", ")), true - - default: - - return fmt.Sprintf("Unknown show target: %s", args[0]), true - } - - case "/list": - - if len(args) < 1 { - return "Usage: /list [models|channels|agents]", true - } - - switch args[0] { - case "models": - - return "Available models: configured in config.json per agent", true - - case "channels": - - if al.channelManager == nil { - return "Channel manager not initialized", true - } - - channels := al.channelManager.GetEnabledChannels() - - if len(channels) == 0 { - return "No channels enabled", true - } - - return fmt.Sprintf("Enabled channels: %s", strings.Join(channels, ", ")), true - - case "agents": - - agentIDs := al.registry.ListAgentIDs() - - return fmt.Sprintf("Registered agents: %s", strings.Join(agentIDs, ", ")), true - - default: - - return fmt.Sprintf("Unknown list target: %s", args[0]), true - } - - case "/switch": - - if len(args) < 3 || args[1] != "to" { - return "Usage: /switch [model|channel] to ", true - } - - target := args[0] - - value := args[2] - - switch target { - case "model": - - defaultAgent := al.registry.GetDefaultAgent() - - if defaultAgent == nil { - return "No default agent configured", true - } - - oldModel := defaultAgent.Model - - defaultAgent.Model = value - - return fmt.Sprintf("Switched model from %s to %s", oldModel, value), true - - case "channel": - - if al.channelManager == nil { - return "Channel manager not initialized", true - } - - if _, exists := al.channelManager.GetChannel(value); !exists && value != "cli" { - return fmt.Sprintf("Channel '%s' not found or not enabled", value), true - } - - return fmt.Sprintf("Switched target channel to %s", value), true - - default: - - return fmt.Sprintf("Unknown switch target: %s", target), true - } - case "/session": - return al.handleSessionCommand(args, msg.SessionKey), true case "/skills": - return al.handleSkillsCommand(), true case "/plan": - resp, handled := al.handlePlanCommand(args, msg.SessionKey) - if handled { al.notifyStateChange() } - return resp, handled case "/heartbeat": - resp, handled := al.handleHeartbeatCommand(args, msg) - if handled { al.notifyStateChange() } - return resp, handled } diff --git a/pkg/agent/loop_ext.go b/pkg/agent/loop_ext.go index 632084a70..fcd29a5fd 100644 --- a/pkg/agent/loop_ext.go +++ b/pkg/agent/loop_ext.go @@ -39,8 +39,12 @@ type loopExt struct { activeTasks sync.Map // sessionKey → *activeTask + activeRequests sync.WaitGroup // tracks in-flight LLM worker requests + done chan struct{} // closed by Close() to stop background goroutines + reloadFunc func() error // upstream compat: called by buildCommandsRuntime + saveConfig func(*config.Config) error onHeartbeatThreadUpdate func(int) @@ -131,6 +135,12 @@ func (al *AgentLoop) pruneMediaCache() { } } +// SetReloadFunc registers a callback to reload config from disk. +// Used by buildCommandsRuntime to wire the /reload command. +func (al *AgentLoop) SetReloadFunc(fn func() error) { + al.reloadFunc = fn +} + // SetConfigSaver registers a callback to persist config changes. func (al *AgentLoop) SetConfigSaver(fn func(*config.Config) error) { al.saveConfig = fn diff --git a/pkg/agent/loop_ext_test.go b/pkg/agent/loop_ext_test.go index 79e12b80f..f9c94ed3b 100644 --- a/pkg/agent/loop_ext_test.go +++ b/pkg/agent/loop_ext_test.go @@ -405,7 +405,7 @@ func TestPlanCommand_ShowNoPlan(t *testing.T) { defer cleanup() - response, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan"}) + response, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan"}, al.registry.GetDefaultAgent(), "") if !handled { t.Fatal("expected /plan to be handled") @@ -488,7 +488,7 @@ func TestHeartbeatCommandThreadHerePersistsConfig(t *testing.T) { ChatID: "-100500/42", } - resp, handled := al.handleCommand(context.Background(), msg) + resp, handled := al.handleCommand(context.Background(), msg, al.registry.GetDefaultAgent(), "") if !handled { t.Fatal("expected /heartbeat command to be handled") @@ -528,7 +528,7 @@ func TestHeartbeatCommandThreadOff(t *testing.T) { Channel: "telegram", ChatID: "-100500/42", - }) + }, al.registry.GetDefaultAgent(), "") if !handled { t.Fatal("expected /heartbeat command to be handled") @@ -548,7 +548,7 @@ func TestPlanCommand_StartNewPlan(t *testing.T) { defer cleanup() - _, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan Set up monitoring"}) + _, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan Set up monitoring"}, al.registry.GetDefaultAgent(), "") if handled { t.Fatal("expected /plan NOT to be handled (should fall through to LLM)") @@ -588,7 +588,7 @@ func TestPlanCommand_StartBlockedByExisting(t *testing.T) { al.expandPlanCommand(bus.InboundMessage{Content: "/plan First task"}) - response, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan Second task"}) + response, handled := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan Second task"}, al.registry.GetDefaultAgent(), "") if !handled { t.Fatal("expected second /plan to be handled (blocked)") @@ -606,7 +606,7 @@ func TestPlanCommand_Clear(t *testing.T) { al.expandPlanCommand(bus.InboundMessage{Content: "/plan Test task"}) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan clear"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan clear"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "Plan cleared") { t.Errorf("expected 'Plan cleared', got %q", response) @@ -624,7 +624,7 @@ func TestPlanCommand_ClearNoPlan(t *testing.T) { defer cleanup() - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan clear"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan clear"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "No active plan") { t.Errorf("expected 'No active plan', got %q", response) @@ -642,7 +642,7 @@ func TestPlanCommand_Start(t *testing.T) { _ = agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "approved") { t.Errorf("expected 'approved', got %q", response) @@ -668,7 +668,7 @@ func TestPlanCommand_StartFromReview(t *testing.T) { _ = agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "approved") { t.Errorf("expected 'approved', got %q", response) @@ -690,7 +690,7 @@ func TestPlanCommand_StartNoPhases(t *testing.T) { al.expandPlanCommand(bus.InboundMessage{Content: "/plan Test task"}) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "no phases") { t.Errorf("expected 'no phases' error, got %q", response) @@ -718,11 +718,11 @@ func TestPlanCommand_StartAlreadyExecuting(t *testing.T) { _ = agent.ContextBuilder.WriteMemory(plan) - al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}) + al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}, al.registry.GetDefaultAgent(), "") al.planStartPending = false - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan start"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "already executing") { t.Errorf("expected 'already executing', got %q", response) @@ -768,7 +768,7 @@ Test context agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan done 1"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan done 1"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "Marked step 1") { t.Errorf("expected confirmation, got %q", response) @@ -782,7 +782,7 @@ func TestPlanCommand_DoneInvalidStep(t *testing.T) { al.expandPlanCommand(bus.InboundMessage{Content: "/plan Test task"}) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan done abc"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan done abc"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "positive integer") { t.Errorf("expected step validation error, got %q", response) @@ -822,7 +822,7 @@ Test context agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan add New step here"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan add New step here"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "Added step") { t.Errorf("expected 'Added step', got %q", response) @@ -874,7 +874,7 @@ Test agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan next"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan next"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "phase 2") { t.Errorf("expected 'phase 2', got %q", response) @@ -920,7 +920,7 @@ Production server agent.ContextBuilder.WriteMemory(plan) - response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan"}) + response, _ := al.handleCommand(context.Background(), bus.InboundMessage{Content: "/plan"}, al.registry.GetDefaultAgent(), "") if !strings.Contains(response, "Deploy app") { t.Errorf("expected task name in display, got %q", response) @@ -2549,7 +2549,7 @@ func TestPlanCommand_StartClear(t *testing.T) { Content: "/plan start clear", SessionKey: "test-session", - }) + }, al.registry.GetDefaultAgent(), "") if !handled { t.Fatal("expected /plan start clear to be handled") @@ -2615,7 +2615,7 @@ func TestPlanCommand_StartWithoutClear_PreservesHistory(t *testing.T) { Content: "/plan start", SessionKey: "test-session", - }) + }, al.registry.GetDefaultAgent(), "") if strings.Contains(response, "clean history") { t.Errorf("did not expect 'clean history' in response, got %q", response) diff --git a/pkg/agent/loop_run.go b/pkg/agent/loop_run.go index 66d4bed82..d4d25e022 100644 --- a/pkg/agent/loop_run.go +++ b/pkg/agent/loop_run.go @@ -768,6 +768,26 @@ func (al *AgentLoop) redirectMessageToolForTask(agent *AgentInstance, task *acti }) } +// selectCandidates resolves the model candidates for a turn. Called once per +// turn (sticky) so the model doesn't change mid-iteration. Plan model hooks +// may further override per-iteration via hooks.SelectModel(). +// +// Uses the agent's pre-resolved Candidates/LightCandidates. When a Router is +// configured, it scores the message and may select the light model tier. +func (al *AgentLoop) selectCandidates( + agent *AgentInstance, + userMsg string, + history []providers.Message, +) []providers.FallbackCandidate { + if agent.Router != nil && len(agent.LightCandidates) > 0 { + _, isLight, _ := agent.Router.SelectModel(userMsg, history, agent.Model) + if isLight { + return agent.LightCandidates + } + } + return agent.Candidates +} + // runLLMIteration executes the LLM call loop with tool handling using hooks. func (al *AgentLoop) runLLMIteration( ctx context.Context, @@ -779,6 +799,10 @@ func (al *AgentLoop) runLLMIteration( ) (string, int, error) { hooks := al.buildHooks(agent, opts, task, planSnapshot) + // Select candidates once per turn (sticky). Plan model hooks may + // override per-iteration via hooks.SelectModel(). + turnCandidates := al.selectCandidates(agent, opts.UserMessage, messages) + iteration := 0 var finalContent string @@ -799,8 +823,9 @@ func (al *AgentLoop) runLLMIteration( // Build tool definitions providerToolDefs := hooks.FilterTools(agent.Tools.ToProviderDefs()) - // Resolve model and candidates for this call - candidates := agent.Candidates + // Resolve model and candidates for this call. + // Default to turn-level candidates; hooks may override per-iteration. + candidates := turnCandidates activeModel := agent.Model if m, c := hooks.SelectModel(); m != "" { activeModel = m @@ -939,6 +964,9 @@ func (al *AgentLoop) runLLMIteration( // Execute tool calls and collect results lastBlocker := al.executeToolCalls(ctx, agent, normalizedToolCalls, &messages, opts, hooks, iteration) + // Tick TTL-based tool expiry after execution + agent.Tools.TickTTL() + hooks.InjectReminders(iteration, &messages, lastBlocker) hooks.RefreshSystemPrompt(messages) } diff --git a/pkg/agent/loop_session.go b/pkg/agent/loop_session.go index 2d38e8af3..d3ee6eded 100644 --- a/pkg/agent/loop_session.go +++ b/pkg/agent/loop_session.go @@ -97,9 +97,9 @@ func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, c tokenEstimate := al.estimateTokens(newHistory) - threshold := agent.ContextWindow * 75 / 100 + threshold := agent.ContextWindow * agent.SummarizeTokenPercent / 100 - if len(newHistory) > 20 || tokenEstimate > threshold { + if len(newHistory) > agent.SummarizeMessageThreshold || tokenEstimate > threshold { summarizeKey := agent.ID + ":" + sessionKey if _, loading := al.summarizing.LoadOrStore(summarizeKey, true); !loading { diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 179cfa860..919a5260a 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -557,7 +557,7 @@ func TestProcessMessage_CommandOutcomes(t *testing.T) { Content: "/show channel", Peer: baseMsg.Peer, }) - if showResp != "Current channel: whatsapp" { + if showResp != "Current Channel: whatsapp" { t.Fatalf("unexpected /show reply: %q", showResp) } if provider.calls != 0 { @@ -641,7 +641,7 @@ func TestProcessMessage_SwitchModelShowModelConsistency(t *testing.T) { ID: "user1", }, }) - if !strings.Contains(showResp, "Current model: after-switch") { + if !strings.Contains(showResp, "Current Model: after-switch") { t.Fatalf("unexpected /show model reply after switch: %q", showResp) } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 9ab0a53e5..738be58b3 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -805,6 +805,18 @@ func (m *Manager) runTTLJanitor(ctx context.Context) { } } +// InvokeTypingStop stops any active typing indicator for the given channel/chatID. +// Safe to call even if no typing indicator is active. Intended for use in defer +// after the LLM worker finishes processing a message. +func (m *Manager) InvokeTypingStop(channel, chatID string) { + key := channel + ":" + chatID + if v, loaded := m.typingStops.LoadAndDelete(key); loaded { + if entry, ok := v.(typingEntry); ok && entry.stop != nil { + entry.stop() + } + } +} + func (m *Manager) GetChannel(name string) (Channel, bool) { m.mu.RLock() defer m.mu.RUnlock()