goclaw/internal/channels/activity_indicator.go

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