opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #68 from dj-oyu/upstream-merge/loop-core-unified

Browse source 
feat: upstream core loop migration
This commit is contained in:
dj-oyu 2026-03-20 20:08:42 +09:00 • committed by GitHub
commit 1938c2e591
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 317 additions and 208 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

130
pkg/agent/loop.go
View file

@ -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,

220
pkg/agent/loop_commands.go
View file

@ -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 <name>", 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
}

10
pkg/agent/loop_ext.go
View file

@ -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

113
pkg/agent/loop_ext_test.go
View file

@ -405,7 +405,12 @@ 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 +493,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 +533,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 +553,12 @@ 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 <task> NOT to be handled (should fall through to LLM)")
@ -588,7 +598,12 @@ 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 +621,12 @@ 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 +644,12 @@ 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 +667,12 @@ 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 +698,12 @@ 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 +725,12 @@ 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 +758,21 @@ 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 +818,12 @@ 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 +837,12 @@ 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 +882,12 @@ 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 +939,12 @@ 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 +990,12 @@ 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 +2624,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 +2690,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)

32
pkg/agent/loop_run.go
View file

@ -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)
}

4
pkg/agent/loop_session.go
View file

@ -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 {

4
pkg/agent/loop_test.go
View file

@ -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)
}

12
pkg/channels/manager.go
View file

@ -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()