feat(bitrix24): agent activity indicator via InputAction.notify [B24:2794]

Show an ephemeral "agent is working" indicator (thinking/searching/
generating/analyzing…) in Bitrix24 chat while the agent processes, so
users on this non-streaming channel aren't left staring at silence
until the final reply. Send behavior is unchanged; no LLM call, no DB.

Generic layer (internal/channels):
- New optional ActivityIndicatorChannel interface.
- HandleAgentEvent routes run.started->THINKING, tool.call->mapped
  status, tool.result->ANALYZING; static tool->status mapping.
- Conditional heartbeat ticker fills LLM-inference gaps (re-sends only
  when idle), 5s per-run throttle caps call rate; ticker stopped on
  terminal events AND in UnregisterRun (safety net for missed terminals).

Bitrix24 (internal/channels/bitrix24):
- OnActivityEvent calls imbot.v2.Chat.InputAction.notify best-effort,
  drop-on-limit (raw Client.Call, no retry) so cosmetic notifies never
  steal leaky-bucket capacity from real message sends.
- Per-channel activity_indicator toggle (default on).

Web UI + docs: dashboard toggle, i18n en/vi/zh, channels doc section.
Tests: 26 unit tests incl. UnregisterRun-stops-ticker regression.
This commit is contained in:
DangTinh311 committed 2026-07-14 15:28:14 +07:00
1 parent 6167a53bbe
commit af43c952ed
14 files changed
+1099

No files matched your search

+23
View File
@@ -168,12 +168,35 @@ Every channel must implement the base interface:
| `StreamingChannel` | Real-time streaming updates | Telegram, Slack |
| `WebhookChannel` | Webhook HTTP handler mounting | Facebook, Feishu/Lark, Pancake |
| `ReactionChannel` | Status reactions on messages | Telegram, Slack, Feishu |
| `ActivityIndicatorChannel` | Ephemeral "agent is working" indicator | Bitrix24 |
| `BlockReplyChannel` | Override gateway block_reply setting | Discord, Feishu/Lark, Pancake, Slack, Zalo OA, Zalo Personal |
| `ChatBehaviorChannel` | Override gateway chat_behavior setting | Bitrix24, Discord, Feishu/Lark, Pancake, Slack, Telegram, WhatsApp, Zalo OA, Zalo Personal |
| `ReasoningDeliveryChannel` | Override channel-visible reasoning delivery | Telegram |
`BaseChannel` provides a shared implementation that all channels embed: allowlist matching, `HandleMessage()`, `CheckPolicy()`, and user ID extraction.
### Activity Indicator (`ActivityIndicatorChannel`)
Shows a native, ephemeral "agent is working" indicator while the agent thinks or runs tools,
so users on non-streaming channels aren't left staring at silence until the final reply. It is
**not** a chat message — nothing is persisted, no extra LLM call is made.
Driven by the existing agent event stream in `Manager.HandleAgentEvent`:
- `run.started` → `THINKING`, and a conditional heartbeat ticker starts.
- `tool.call` → status mapped from the tool name (`SEARCHING`, `READING_DOCS`, `GENERATING`,
`CONNECTING`, `PROCESSING`) via `resolveToolActivityStatus`.
- `tool.result` → `ANALYZING`.
- terminal events → ticker stops.
Because non-streaming turns emit no events during LLM inference, a **conditional heartbeat
ticker** re-sends the current status only when the run has been idle beyond a threshold — filling
the gap without spamming. Calls are **best-effort and dropped on rate limit** (they never retry
into the portal's leaky bucket, so real message sends are never starved).
**Bitrix24** implements it via `imbot.v2.Chat.InputAction.notify` (status codes
`IMBOT_AGENT_ACTION_*`). Toggle per channel with `activity_indicator` (default on).
### Webhook Mount
Channels implementing `WebhookChannel` expose an HTTP handler that can be mounted on the gateway's main HTTP mux. This enables single-port operation — no separate webhook server needed.
+152
View File
@@ -0,0 +1,152 @@
package channels
import (
"context"
"log/slog"
"strings"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// Activity status codes are Bitrix24 imbot InputAction.notify codes. They are kept
// generic in the channel layer; a channel that cannot use them may translate or ignore.
// See imbot.v2.Chat.InputAction.notify (statusMessageCode).
const (
ActivityStatusThinking = "IMBOT_AGENT_ACTION_THINKING"
ActivityStatusSearching = "IMBOT_AGENT_ACTION_SEARCHING"
ActivityStatusGenerating = "IMBOT_AGENT_ACTION_GENERATING"
ActivityStatusAnalyzing = "IMBOT_AGENT_ACTION_ANALYZING"
ActivityStatusProcessing = "IMBOT_AGENT_ACTION_PROCESSING"
ActivityStatusReadingDoc = "IMBOT_AGENT_ACTION_READING_DOCS"
ActivityStatusConnecting = "IMBOT_AGENT_ACTION_CONNECTING"
)
// Activity indicator timing. The indicator auto-expires after activityDuration on the
// platform side; the heartbeat re-sends the current status while the run is idle (no
// events) so it never disappears mid-run. The throttle caps the call rate per run so
// bursts of fast tool calls cannot spam the platform REST API.
const (
activityThrottle = 5 * time.Second // min interval between notifies per run
activityTickerInterval = 12 * time.Second // heartbeat check cadence
activityHeartbeatIdle = 20 * time.Second // re-send only if idle longer than this
)
// resolveToolActivityStatus maps a tool name to an activity status code.
// Static mapping (no DB); case-insensitive substring match. Mirrors the pattern of
// resolveToolReactionStatus but targets the richer Bitrix activity vocabulary.
func resolveToolActivityStatus(toolName string) string {
n := strings.ToLower(toolName)
switch {
case containsAnySubstr(n, "web", "search", "browser", "fetch"):
return ActivityStatusSearching
case containsAnySubstr(n, "read_file", "list_files", "vault", "memory", "docs", "skill"):
return ActivityStatusReadingDoc
case containsAnySubstr(n, "image", "tts", "speech", "video", "music", "generate"):
return ActivityStatusGenerating
case containsAnySubstr(n, "mcp", "bitrix", "crm"):
return ActivityStatusConnecting
case containsAnySubstr(n, "delegate", "subagent", "team"):
return ActivityStatusProcessing
default:
return ActivityStatusProcessing
}
}
func containsAnySubstr(s string, subs ...string) bool {
for _, sub := range subs {
if strings.Contains(s, sub) {
return true
}
}
return false
}
// fireActivity sends an activity notify best-effort, subject to the per-run throttle.
// If status is non-empty it becomes the current status; an empty status re-sends the
// current one (used by the heartbeat). The actual REST call runs in a detached goroutine
// so a slow/failed indicator never blocks agent event routing.
func (m *Manager) fireActivity(rc *RunContext, ch ActivityIndicatorChannel, status string) {
rc.mu.Lock()
// Update the current status BEFORE the throttle gate on purpose: latest status
// wins even when this particular call is throttled, so the next heartbeat/notify
// reflects the most recent phase.
if status != "" {
rc.activityStatus = status
}
if time.Since(rc.lastActivityAt) < activityThrottle {
rc.mu.Unlock()
return
}
rc.lastActivityAt = time.Now()
cur := rc.activityStatus
chatID := rc.ChatID
tenant := rc.TenantID
channelName := rc.ChannelName
rc.mu.Unlock()
go func() {
// Fresh ctx (the event ctx may be cancelled by the time this runs). Tenant scope
// is attached for parity with other channel event handlers / channels that need
// it; Bitrix24 derives portal auth from its client instance, not from ctx.
ctx := context.Background()
if tenant != uuid.Nil {
ctx = store.WithTenantID(ctx, tenant)
}
if err := ch.OnActivityEvent(ctx, chatID, cur); err != nil {
// Best-effort: drop-on-limit. Never surface indicator failures.
slog.Debug("activity indicator failed", "channel", channelName, "status", cur, "error", err)
}
}()
}
// startActivityTicker starts the per-run heartbeat that keeps the indicator alive across
// LLM-inference windows (where no agent events fire). Start-once per run. The ticker only
// re-sends when the run has been idle longer than activityHeartbeatIdle, so tool-active
// runs generate no extra calls. Must be paired with stopActivityTicker on terminal events.
func (m *Manager) startActivityTicker(rc *RunContext, ch ActivityIndicatorChannel) {
rc.mu.Lock()
if rc.activityStarted {
rc.mu.Unlock()
return
}
rc.activityStarted = true
stop := make(chan struct{})
ticker := time.NewTicker(activityTickerInterval)
rc.activityStop = stop
rc.activityTicker = ticker
rc.mu.Unlock()
go func() {
for {
select {
case <-stop:
return
case <-ticker.C:
rc.mu.Lock()
idle := time.Since(rc.lastActivityAt)
rc.mu.Unlock()
if idle >= activityHeartbeatIdle {
m.fireActivity(rc, ch, "") // re-send current status
}
}
}
}()
}
// stopActivityTicker stops the heartbeat goroutine and ticker. Idempotent — safe to call
// on any terminal event even if the ticker was never started.
func (m *Manager) stopActivityTicker(rc *RunContext) {
rc.mu.Lock()
if rc.activityStop != nil {
close(rc.activityStop)
rc.activityStop = nil
}
if rc.activityTicker != nil {
rc.activityTicker.Stop()
rc.activityTicker = nil
}
rc.mu.Unlock()
}
@@ -0,0 +1,541 @@
package channels
import (
"context"
"errors"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
)
// fakeActivityIndicatorChannel is a minimal test implementation of ActivityIndicatorChannel.
// It records all OnActivityEvent calls for assertion in tests.
type fakeActivityIndicatorChannel struct {
*BaseChannel
mu sync.Mutex
calls []ActivityEventCall // recorded calls with timestamps
callTimes []time.Time // exact timing of each call
shouldBeginFail bool // if true, future calls return error
}
type ActivityEventCall struct {
ChatID string
StatusCode string
Timestamp time.Time
}
func newFakeActivityIndicatorChannel(name string) *fakeActivityIndicatorChannel {
return &fakeActivityIndicatorChannel{
BaseChannel: NewBaseChannel(name, bus.New(), nil),
calls: []ActivityEventCall{},
callTimes: []time.Time{},
}
}
func (c *fakeActivityIndicatorChannel) Start(context.Context) error {
c.SetRunning(true)
return nil
}
func (c *fakeActivityIndicatorChannel) Stop(context.Context) error {
c.SetRunning(false)
return nil
}
func (c *fakeActivityIndicatorChannel) Send(context.Context, bus.OutboundMessage) error {
return nil
}
func (c *fakeActivityIndicatorChannel) OnActivityEvent(ctx context.Context, chatID, statusCode string) error {
c.mu.Lock()
defer c.mu.Unlock()
now := time.Now()
c.calls = append(c.calls, ActivityEventCall{
ChatID: chatID,
StatusCode: statusCode,
Timestamp: now,
})
c.callTimes = append(c.callTimes, now)
if c.shouldBeginFail {
return errors.New("activity indicator test error")
}
return nil
}
func (c *fakeActivityIndicatorChannel) GetCallCount() int {
c.mu.Lock()
defer c.mu.Unlock()
return len(c.calls)
}
func (c *fakeActivityIndicatorChannel) GetCalls() []ActivityEventCall {
c.mu.Lock()
defer c.mu.Unlock()
// Return a copy to avoid race on inspection
calls := make([]ActivityEventCall, len(c.calls))
copy(calls, c.calls)
return calls
}
// TestResolveToolActivityStatus_SearchTools tests substring mapping for search/web tools.
func TestResolveToolActivityStatus_SearchTools(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"web_search lowercase", "web_search", ActivityStatusSearching},
{"Web_Search uppercase", "Web_Search", ActivityStatusSearching},
{"browser tool", "browser_automation", ActivityStatusSearching},
{"fetch_url", "fetch_url", ActivityStatusSearching},
{"search_documents", "search_documents", ActivityStatusSearching},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestResolveToolActivityStatus_ReadDocTools tests file/memory/docs tools.
// Note: "vault_search" and similar tools containing "search" substring will match the
// search pattern first (order-dependent), so they return SEARCHING not READING_DOCS.
// This is expected behavior due to the switch case order in resolveToolActivityStatus.
func TestResolveToolActivityStatus_ReadDocTools(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"read_file", "read_file", ActivityStatusReadingDoc},
{"list_files", "list_files", ActivityStatusReadingDoc},
{"vault_only", "vault", ActivityStatusReadingDoc},
{"memory_store", "memory_store", ActivityStatusReadingDoc},
{"docs_retrieve", "docs_retrieve", ActivityStatusReadingDoc},
{"skill_invoke", "skill_invoke", ActivityStatusReadingDoc},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestResolveToolActivityStatus_GeneratingTools tests image/tts/media generation tools.
func TestResolveToolActivityStatus_GeneratingTools(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"create_image", "create_image", ActivityStatusGenerating},
{"image_edit", "image_edit", ActivityStatusGenerating},
{"tts_speech", "tts_speech", ActivityStatusGenerating},
{"speech_to_text", "speech_to_text", ActivityStatusGenerating},
{"video_generator", "video_generator", ActivityStatusGenerating},
{"music_compose", "music_compose", ActivityStatusGenerating},
{"generate_diagram", "generate_diagram", ActivityStatusGenerating},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestResolveToolActivityStatus_ConnectingTools tests MCP/Bitrix/CRM tools.
// Note: tools with "search" in the name (like "crm_contact_search") will match the
// search pattern first due to order-dependent switch logic, returning SEARCHING instead.
// Only tools with mcp/bitrix/crm (without search) return CONNECTING.
func TestResolveToolActivityStatus_ConnectingTools(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"mcp_bx24_crm_deal_list", "mcp_bx24_crm_deal_list", ActivityStatusConnecting},
{"bitrix_api", "bitrix_api", ActivityStatusConnecting},
{"crm_direct_call", "crm_direct_call", ActivityStatusConnecting},
{"mcp_custom_tool", "mcp_custom_tool", ActivityStatusConnecting},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestResolveToolActivityStatus_ProcessingTools tests delegation/subagent/team tools.
func TestResolveToolActivityStatus_ProcessingTools(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"delegate", "delegate", ActivityStatusProcessing},
{"subagent_invoke", "subagent_invoke", ActivityStatusProcessing},
{"team_create_task", "team_create_task", ActivityStatusProcessing},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestResolveToolActivityStatus_UnknownTools tests default fallback for unmapped tools.
func TestResolveToolActivityStatus_UnknownTools(t *testing.T) {
tests := []struct {
name string
toolName string
}{
{"unknown_tool", "unknown_tool"},
{"foo_bar", "foo_bar"},
{"empty_string", ""},
{"single_char", "x"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != ActivityStatusProcessing {
t.Errorf("resolveToolActivityStatus(%q) = %q, want default %q", tt.toolName, got, ActivityStatusProcessing)
}
})
}
}
// TestResolveToolActivityStatus_CaseInsensitive tests case-insensitive matching.
func TestResolveToolActivityStatus_CaseInsensitive(t *testing.T) {
tests := []struct {
name string
toolName string
want string
}{
{"MCP_BX24_CRM", "MCP_BX24_CRM", ActivityStatusConnecting},
{"WEB_SEARCH", "WEB_SEARCH", ActivityStatusSearching},
{"Read_File", "Read_File", ActivityStatusReadingDoc},
{"CREATE_IMAGE", "CREATE_IMAGE", ActivityStatusGenerating},
{"DELEGATE", "DELEGATE", ActivityStatusProcessing},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveToolActivityStatus(tt.toolName)
if got != tt.want {
t.Errorf("resolveToolActivityStatus(%q) = %q, want %q", tt.toolName, got, tt.want)
}
})
}
}
// TestManagerFireActivity_Throttle tests that fireActivity respects the 5-second throttle.
func TestManagerFireActivity_Throttle(t *testing.T) {
mgr := NewManager(bus.New())
fakeChannel := newFakeActivityIndicatorChannel("test")
// Create RunContext with all required fields
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
ToolStatusEnabled: true,
}
// First fireActivity call should succeed
mgr.fireActivity(rc, fakeChannel, ActivityStatusThinking)
// Wait a short time to ensure timestamp difference is meaningful
time.Sleep(100 * time.Millisecond)
// Second fireActivity call within 5s throttle window → should be dropped
mgr.fireActivity(rc, fakeChannel, ActivityStatusSearching)
// Give goroutines time to fire
time.Sleep(100 * time.Millisecond)
// Should only have 1 OnActivityEvent call due to throttle
if fakeChannel.GetCallCount() != 1 {
t.Errorf("expected 1 OnActivityEvent call after throttle, got %d", fakeChannel.GetCallCount())
}
// Manually advance lastActivityAt to bypass throttle
rc.mu.Lock()
rc.lastActivityAt = time.Now().Add(-time.Duration(activityThrottle) - 1*time.Second)
rc.mu.Unlock()
// Now fireActivity should succeed again
mgr.fireActivity(rc, fakeChannel, ActivityStatusAnalyzing)
// Give goroutines time to fire
time.Sleep(100 * time.Millisecond)
// Should now have 2 calls
if fakeChannel.GetCallCount() != 2 {
t.Errorf("expected 2 OnActivityEvent calls after throttle bypass, got %d", fakeChannel.GetCallCount())
}
// Verify second call has the new status
calls := fakeChannel.GetCalls()
if len(calls) > 1 && calls[1].StatusCode != ActivityStatusAnalyzing {
t.Errorf("second call status = %q, want %q", calls[1].StatusCode, ActivityStatusAnalyzing)
}
}
// TestManagerFireActivity_EmptyStatusResends tests that empty status re-sends current status.
func TestManagerFireActivity_EmptyStatusResend(t *testing.T) {
mgr := NewManager(bus.New())
fakeChannel := newFakeActivityIndicatorChannel("test")
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
}
// First call sets status
mgr.fireActivity(rc, fakeChannel, ActivityStatusThinking)
time.Sleep(100 * time.Millisecond)
// Bypass throttle
rc.mu.Lock()
rc.lastActivityAt = time.Now().Add(-time.Duration(activityThrottle) - 1*time.Second)
rc.mu.Unlock()
// Second call with empty status should re-send current status
mgr.fireActivity(rc, fakeChannel, "")
time.Sleep(100 * time.Millisecond)
calls := fakeChannel.GetCalls()
if len(calls) < 2 {
t.Fatalf("expected at least 2 calls, got %d", len(calls))
}
// Both calls should have the same status (first one)
if calls[0].StatusCode != ActivityStatusThinking {
t.Errorf("first call status = %q, want %q", calls[0].StatusCode, ActivityStatusThinking)
}
if calls[1].StatusCode != ActivityStatusThinking {
t.Errorf("second call status = %q, want %q (re-sent)", calls[1].StatusCode, ActivityStatusThinking)
}
}
// TestManagerStartActivityTicker_StartOnce tests that ticker is started only once.
func TestManagerStartActivityTicker_StartOnce(t *testing.T) {
mgr := NewManager(bus.New())
fakeChannel := newFakeActivityIndicatorChannel("test")
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
}
// First start should succeed
mgr.startActivityTicker(rc, fakeChannel)
rc.mu.Lock()
firstTicker := rc.activityTicker
firstStarted := rc.activityStarted
rc.mu.Unlock()
if !firstStarted {
t.Errorf("first startActivityTicker should set activityStarted=true")
}
if firstTicker == nil {
t.Errorf("first startActivityTicker should create a ticker")
}
// Second start should be idempotent (no-op)
mgr.startActivityTicker(rc, fakeChannel)
rc.mu.Lock()
secondTicker := rc.activityTicker
secondStarted := rc.activityStarted
rc.mu.Unlock()
if !secondStarted {
t.Errorf("second startActivityTicker should keep activityStarted=true")
}
if secondTicker != firstTicker {
t.Errorf("second startActivityTicker should reuse the same ticker, not create a new one")
}
}
// TestManagerStopActivityTicker_Idempotent tests that stopActivityTicker can be called safely multiple times.
func TestManagerStopActivityTicker_Idempotent(t *testing.T) {
mgr := NewManager(bus.New())
fakeChannel := newFakeActivityIndicatorChannel("test")
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
}
// Start the ticker
mgr.startActivityTicker(rc, fakeChannel)
// First stop should succeed
mgr.stopActivityTicker(rc)
rc.mu.Lock()
afterFirstStop := rc.activityTicker
afterFirstStopChan := rc.activityStop
rc.mu.Unlock()
if afterFirstStop != nil {
t.Errorf("first stopActivityTicker should clear ticker")
}
if afterFirstStopChan != nil {
t.Errorf("first stopActivityTicker should clear stop channel")
}
// Second stop should also succeed (idempotent)
mgr.stopActivityTicker(rc)
// No panic = success
// Third stop should also succeed
mgr.stopActivityTicker(rc)
// No panic = success
}
// TestManagerStopActivityTicker_NeverStarted tests that stopActivityTicker handles never-started case.
func TestManagerStopActivityTicker_NeverStarted(t *testing.T) {
mgr := NewManager(bus.New())
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
}
// Never called startActivityTicker, just stop
mgr.stopActivityTicker(rc)
// Should not panic; no-op
if rc.activityTicker != nil {
t.Errorf("stopActivityTicker on never-started should be safe, but ticker is non-nil")
}
}
// TestContainsAnySubstr_Match tests that containsAnySubstr finds substrings.
func TestContainsAnySubstr_Match(t *testing.T) {
tests := []struct {
name string
s string
subs []string
want bool
}{
{"first match", "hello world", []string{"hello"}, true},
{"middle match", "hello world", []string{"lo wo"}, true},
{"last match", "hello world", []string{"world"}, true},
{"multi match first", "hello world", []string{"hello", "xor"}, true},
{"multi match second", "hello world", []string{"xor", "world"}, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := containsAnySubstr(tt.s, tt.subs...)
if got != tt.want {
t.Errorf("containsAnySubstr(%q, %v) = %v, want %v", tt.s, tt.subs, got, tt.want)
}
})
}
}
// TestContainsAnySubstr_NoMatch tests that containsAnySubstr returns false when no substring matches.
func TestContainsAnySubstr_NoMatch(t *testing.T) {
tests := []struct {
name string
s string
subs []string
}{
{"no match", "hello world", []string{"xor"}},
{"empty list", "hello world", []string{}},
{"empty string", "", []string{"hello"}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := containsAnySubstr(tt.s, tt.subs...)
if got {
t.Errorf("containsAnySubstr(%q, %v) = true, want false", tt.s, tt.subs)
}
})
}
}
// TestManagerUnregisterRun_StopsActivityTicker is the C1 regression test: UnregisterRun
// (the safety-net cleanup used when a terminal AgentEvent never reaches HandleAgentEvent)
// MUST stop the heartbeat, otherwise the goroutine leaks and keeps firing
// InputAction.notify forever. Before the fix, UnregisterRun cleaned quick-ack/bubbles but
// not the activity ticker.
func TestManagerUnregisterRun_StopsActivityTicker(t *testing.T) {
mgr := NewManager(bus.New())
fakeChannel := newFakeActivityIndicatorChannel("test")
rc := &RunContext{
ChannelName: "test",
ChatID: "chat123",
MessageID: "msg456",
TenantID: uuid.Nil,
}
const runID = "run-c1"
mgr.runs.Store(runID, rc)
mgr.startActivityTicker(rc, fakeChannel)
rc.mu.Lock()
running := rc.activityTicker != nil
rc.mu.Unlock()
if !running {
t.Fatalf("ticker should be running before UnregisterRun")
}
// Simulate the missed-terminal path: no run.completed event reaches
// HandleAgentEvent, only the consumer's safety-net UnregisterRun fires.
mgr.UnregisterRun(runID)
rc.mu.Lock()
ticker := rc.activityTicker
stop := rc.activityStop
rc.mu.Unlock()
if ticker != nil {
t.Errorf("UnregisterRun should stop the activity ticker to prevent goroutine/REST leak; got non-nil ticker")
}
if stop != nil {
t.Errorf("UnregisterRun should clear the stop channel; got non-nil")
}
}
+71
View File
@@ -0,0 +1,71 @@
package bitrix24
import (
"context"
"log/slog"
"strings"
"time"
"github.com/nextlevelbuilder/goclaw/internal/channels"
)
// Compile-time guard: *Channel must satisfy ActivityIndicatorChannel so
// HandleAgentEvent's type assertion routes activity events to Bitrix24.
var _ channels.ActivityIndicatorChannel = (*Channel)(nil)
// Activity indicator tuning. The platform auto-expires the indicator after
// activityDurationSeconds; the channel layer (Manager heartbeat) re-sends before it
// lapses. The call uses a short timeout so a slow REST round-trip cannot pile up.
const (
activityDurationSeconds = 30
activityCallTimeout = 4 * time.Second
)
// activityIndicatorEnabled reports whether the ephemeral activity indicator is on.
// Default is on (nil config value).
func (c *Channel) activityIndicatorEnabled() bool {
return c.cfg.ActivityIndicator == nil || *c.cfg.ActivityIndicator
}
// OnActivityEvent implements channels.ActivityIndicatorChannel. It shows a native
// Bitrix24 "agent is working" indicator via imbot.v2.Chat.InputAction.notify.
//
// This is COSMETIC and BEST-EFFORT: it deliberately does NOT use
// callWithRateLimitRetry. On QUERY_LIMIT_EXCEEDED (or any error) the notify is dropped
// silently so it never retries into — and never steals leaky-bucket capacity from — the
// real message sends. Rate limits are per-portal/shared-per-IP, so yielding here keeps
// imbot.v2.Chat.Message.send unaffected.
//
// statusCode is a Bitrix action code (IMBOT_AGENT_ACTION_*). Empty → default typing.
//
// MCP-DERIVED: imbot.v2.Chat.InputAction.notify param casing (botId/dialogId/
// statusMessageCode/duration) mirrors imbot.v2.Chat.Message.send; verify live vs the
// portal before release (Hard Rule #13).
func (c *Channel) OnActivityEvent(ctx context.Context, chatID, statusCode string) error {
if !c.activityIndicatorEnabled() {
return nil
}
botID := c.BotID()
// Liveness guard — mirror Send (send.go): skip when not fully started.
if c.Client() == nil || botID <= 0 || strings.TrimSpace(chatID) == "" {
return nil
}
params := map[string]any{
"botId": botID,
"dialogId": chatID,
"duration": activityDurationSeconds,
}
if statusCode != "" {
params["statusMessageCode"] = statusCode
}
cctx, cancel := context.WithTimeout(ctx, activityCallTimeout)
defer cancel()
if _, err := c.Client().Call(cctx, "imbot.v2.Chat.InputAction.notify", params); err != nil {
// Drop-on-limit: cosmetic indicator must never disturb real traffic.
slog.Debug("bitrix24.activity.notify dropped",
"chat_id", chatID, "code", statusCode, "err", err)
}
return nil
}
+250
View File
@@ -0,0 +1,250 @@
package bitrix24
import (
"context"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
)
// TestActivityIndicatorEnabled_Nil tests that nil config means enabled (default on).
func TestActivityIndicatorEnabled_Nil(t *testing.T) {
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // default
},
}
if !ch.activityIndicatorEnabled() {
t.Errorf("activityIndicatorEnabled() with nil config should return true (default on)")
}
}
// TestActivityIndicatorEnabled_ExplicitTrue tests that *true enables the indicator.
func TestActivityIndicatorEnabled_ExplicitTrue(t *testing.T) {
trueVal := true
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: &trueVal,
},
}
if !ch.activityIndicatorEnabled() {
t.Errorf("activityIndicatorEnabled() with *true should return true")
}
}
// TestActivityIndicatorEnabled_ExplicitFalse tests that *false disables the indicator.
func TestActivityIndicatorEnabled_ExplicitFalse(t *testing.T) {
falseVal := false
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: &falseVal,
},
}
if ch.activityIndicatorEnabled() {
t.Errorf("activityIndicatorEnabled() with *false should return false")
}
}
// TestOnActivityEvent_DisabledReturnsNilNoCall tests that disabled indicator returns nil and makes no call.
func TestOnActivityEvent_DisabledReturnsNilNoCall(t *testing.T) {
falseVal := false
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: &falseVal,
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
err := ch.OnActivityEvent(context.Background(), "chat123", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with disabled indicator should return nil, got %v", err)
}
if len(rt.reqs) != 0 {
t.Errorf("disabled indicator should make no REST calls, made %d", len(rt.reqs))
}
}
// TestOnActivityEvent_EnabledMakesCall tests that enabled indicator makes a REST call.
func TestOnActivityEvent_EnabledMakesCall(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // default: enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
ch.botID = 123
err := ch.OnActivityEvent(context.Background(), "chat123", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with enabled indicator should return nil, got %v", err)
}
if len(rt.reqs) != 1 {
t.Errorf("enabled indicator should make 1 REST call, made %d", len(rt.reqs))
}
}
// TestOnActivityEvent_NilClientReturnsNil tests that nil client returns nil without calling.
func TestOnActivityEvent_NilClientReturnsNil(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = nil // nil client
err := ch.OnActivityEvent(context.Background(), "chat123", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with nil client should return nil, got %v", err)
}
if len(rt.reqs) != 0 {
t.Errorf("nil client should make no REST calls, made %d", len(rt.reqs))
}
}
// TestOnActivityEvent_InvalidBotIDReturnsNil tests that botID <= 0 returns nil without calling.
func TestOnActivityEvent_InvalidBotIDReturnsNil(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
ch.botID = 0 // invalid: <= 0
err := ch.OnActivityEvent(context.Background(), "chat123", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with invalid botID should return nil, got %v", err)
}
if len(rt.reqs) != 0 {
t.Errorf("invalid botID should make no REST calls, made %d", len(rt.reqs))
}
}
// TestOnActivityEvent_EmptyChatIDReturnsNil tests that empty chatID returns nil without calling.
func TestOnActivityEvent_EmptyChatIDReturnsNil(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
ch.botID = 123
// Empty chatID should fail guard
err := ch.OnActivityEvent(context.Background(), "", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with empty chatID should return nil, got %v", err)
}
if len(rt.reqs) != 0 {
t.Errorf("empty chatID should make no REST calls, made %d", len(rt.reqs))
}
// Whitespace-only chatID should also fail guard
err = ch.OnActivityEvent(context.Background(), " \t ", "IMBOT_AGENT_ACTION_THINKING")
if err != nil {
t.Errorf("OnActivityEvent with whitespace chatID should return nil, got %v", err)
}
if len(rt.reqs) != 0 {
t.Errorf("whitespace chatID should make no REST calls, made %d", len(rt.reqs))
}
}
// TestOnActivityEvent_ValidCallIncludesParams tests that enabled/valid call includes all required params.
func TestOnActivityEvent_ValidCallIncludesParams(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
ch.botID = 456
err := ch.OnActivityEvent(context.Background(), "chat789", "IMBOT_AGENT_ACTION_SEARCHING")
if err != nil {
t.Errorf("OnActivityEvent should return nil, got %v", err)
}
if len(rt.reqs) != 1 {
t.Fatalf("expected 1 call, got %d", len(rt.reqs))
}
// Verify params
params := rt.reqs[0]
if params.Get("botId") != "456" {
t.Errorf("botId param = %q, want 456", params.Get("botId"))
}
if params.Get("dialogId") != "chat789" {
t.Errorf("dialogId param = %q, want chat789", params.Get("dialogId"))
}
if params.Get("statusMessageCode") != "IMBOT_AGENT_ACTION_SEARCHING" {
t.Errorf("statusMessageCode param = %q, want IMBOT_AGENT_ACTION_SEARCHING", params.Get("statusMessageCode"))
}
if params.Get("duration") != "30" {
t.Errorf("duration param = %q, want 30", params.Get("duration"))
}
}
// TestOnActivityEvent_EmptyStatusCodeOmitsParam tests that empty status code omits the statusMessageCode param.
func TestOnActivityEvent_EmptyStatusCodeOmitsParam(t *testing.T) {
rt := &captureRT{}
ch := &Channel{
cfg: bitrixInstanceConfig{
ActivityIndicator: nil, // enabled
},
BaseChannel: channels.NewBaseChannel("test", bus.New(), nil),
}
ch.client = newStubClient("test.bitrix24.com", rt)
ch.botID = 789
err := ch.OnActivityEvent(context.Background(), "chat999", "")
if err != nil {
t.Errorf("OnActivityEvent with empty status should return nil, got %v", err)
}
if len(rt.reqs) != 1 {
t.Fatalf("expected 1 call, got %d", len(rt.reqs))
}
// Verify statusMessageCode is absent when empty
params := rt.reqs[0]
if params.Get("statusMessageCode") != "" {
t.Errorf("empty status should omit statusMessageCode, but got %q", params.Get("statusMessageCode"))
}
// Other params should still be present
if params.Get("botId") != "789" {
t.Errorf("botId should be present, got %q", params.Get("botId"))
}
}
// TestCompileTimeGuard verifies that *Channel satisfies ActivityIndicatorChannel.
func TestCompileTimeGuard(t *testing.T) {
// This is a compile-time check, but we verify the interface at runtime here.
var _ channels.ActivityIndicatorChannel = (*Channel)(nil)
t.Log("*Channel implements ActivityIndicatorChannel interface")
}
+5
View File
@@ -72,6 +72,11 @@ type bitrixInstanceConfig struct {
Streaming *bool `json:"streaming,omitempty"`
ReactionLevel string `json:"reaction_level,omitempty"` // off|minimal|full
// ActivityIndicator toggles the ephemeral "agent is working" indicator
// (imbot.v2.Chat.InputAction.notify) shown while the agent thinks / runs
// tools. nil = enabled (default on). Best-effort, dropped on rate limit.
ActivityIndicator *bool `json:"activity_indicator,omitempty"`
// Misc
HistoryLimit int `json:"history_limit,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"`
+13
View File
@@ -199,6 +199,19 @@ type ReactionChannel interface {
ClearReaction(ctx context.Context, chatID string, messageID string) error
}
// ActivityIndicatorChannel is optionally implemented by channels that can show an
// ephemeral "agent is working" indicator (e.g. Bitrix24 imbot InputAction.notify)
// while the agent thinks or runs tools — without persisting a chat message.
//
// statusCode is a platform-native activity code (see activity_indicator.go); an empty
// string means "keep the current / default working indicator". Implementations MUST be
// best-effort and non-blocking on the caller's side: the indicator is cosmetic, so a
// failed or rate-limited call must be dropped silently and never affect the real reply.
type ActivityIndicatorChannel interface {
Channel
OnActivityEvent(ctx context.Context, chatID string, statusCode string) error
}
// BaseChannel provides shared functionality for all channel implementations.
// Channel implementations should embed this struct.
type BaseChannel struct {
+17
View File
@@ -428,9 +428,26 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
}
}
// Forward to ActivityIndicatorChannel — ephemeral "agent is working" indicator
// (e.g. Bitrix24 imbot InputAction.notify). Events set the status content; a
// conditional heartbeat ticker (started here) fills LLM-inference gaps. Best-effort:
// fireActivity never blocks and drops on error/rate-limit.
if actCh, ok := ch.(ActivityIndicatorChannel); ok {
switch eventType {
case protocol.AgentEventRunStarted:
m.fireActivity(rc, actCh, ActivityStatusThinking)
m.startActivityTicker(rc, actCh)
case protocol.AgentEventToolCall:
m.fireActivity(rc, actCh, resolveToolActivityStatus(extractPayloadString(payload, "name")))
case protocol.AgentEventToolResult:
m.fireActivity(rc, actCh, ActivityStatusAnalyzing)
}
}
// Clean up on terminal events
if eventType == protocol.AgentEventRunCompleted || eventType == protocol.AgentEventRunFailed || eventType == protocol.AgentEventRunCancelled {
m.cancelQuickAck(rc)
m.stopActivityTicker(rc)
rc.mu.Lock()
stopReasoningBubbleTimerLocked(rc)
rc.mu.Unlock()
+9
View File
@@ -58,6 +58,15 @@ type RunContext struct {
tagParsePending string // raw trailing text withheld because it may be a split <think> tag
reasoningBubbles *reasoningBubbleBuffer
reasoningBubbleTimer *time.Timer
// Activity indicator state (for ActivityIndicatorChannel, e.g. Bitrix24).
// Ephemeral "agent is working" indicator driven by agent events + a conditional
// heartbeat ticker. All fields guarded by mu.
activityStatus string // current platform-native status code (e.g. THINKING)
lastActivityAt time.Time // last time a notify was sent (throttle + heartbeat gate)
activityStarted bool // true once the heartbeat ticker is running (start-once guard)
activityTicker *time.Ticker // heartbeat ticker; nil when not running
activityStop chan struct{} // closed to stop the heartbeat goroutine
}
// Manager manages all registered channels, handling their lifecycle
+5
View File
@@ -45,6 +45,11 @@ func (m *Manager) UnregisterRun(runID string) {
if val, ok := m.runs.LoadAndDelete(runID); ok {
if rc, ok := val.(*RunContext); ok {
m.cancelQuickAck(rc)
// Also stop the activity heartbeat here (not only in the terminal-event
// branch) — terminal AgentEvents are not guaranteed to reach
// HandleAgentEvent, and without this the heartbeat goroutine leaks and
// keeps firing InputAction.notify forever. stopActivityTicker is idempotent.
m.stopActivityTicker(rc)
m.flushReasoningBubblesForContext(rc)
}
}
+4
View File
@@ -395,6 +395,10 @@
"reaction_level": {
"label": "Reaction Level"
},
"activity_indicator": {
"label": "Activity Indicator",
"help": "Show a native \"agent is working\" indicator (thinking/searching/generating…) while the agent processes. Ephemeral, not a chat message."
},
"media_max_mb": {
"label": "Max Media Size (MB)"
},
+4
View File
@@ -394,6 +394,10 @@
"reaction_level": {
"label": "Mức độ reaction"
},
"activity_indicator": {
"label": "Chỉ báo hoạt động",
"help": "Hiển thị chỉ báo \"agent đang làm việc\" (đang nghĩ/tìm/tạo…) trong lúc agent xử lý. Tạm thời, không phải tin nhắn."
},
"media_max_mb": {
"label": "Kích thước media tối đa (MB)"
},
+4
View File
@@ -394,6 +394,10 @@
"reaction_level": {
"label": "表情回应级别"
},
"activity_indicator": {
"label": "活动指示器",
"help": "在智能体处理时显示原生的“智能体工作中”指示器(思考/搜索/生成…)。临时显示,不是聊天消息。"
},
"media_max_mb": {
"label": "最大媒体大小 (MB)"
},
@@ -293,6 +293,7 @@ export const configSchema: Record<string, FieldDef[]> = {
{ key: "history_limit", label: "Group History Limit", type: "number", defaultValue: 0, help: "Max pending group messages for context (0 = disabled)" },
{ key: "streaming", label: "Streaming", type: "boolean", defaultValue: true, help: "Stream response progressively." },
{ key: "reaction_level", label: "Reaction Level", type: "select", options: [{ value: "off", label: "Off" }, { value: "minimal", label: "Minimal" }, { value: "full", label: "Full" }], defaultValue: "minimal", help: "Typing/status reactions while the agent is processing." },
{ key: "activity_indicator", label: "Activity Indicator", type: "boolean", defaultValue: true, help: "Show a native \"agent is working\" indicator (thinking/searching/generating…) while the agent processes. Ephemeral, not a chat message." },
{ key: "text_chunk_limit", label: "Text Chunk Limit", type: "number", defaultValue: 4000, help: "Max characters per outbound message." },
{ key: "media_max_mb", label: "Max Media Size (MB)", type: "number", defaultValue: 20, help: "Max inbound media download size." },
{ key: "allow_from", label: "Allowed Users (DM)", type: "tags", help: "Bitrix24 user IDs allowed to DM the bot. Empty = no allowlist filter." },