diff --git a/internal/channels/bitrix24/channel.go b/internal/channels/bitrix24/channel.go index 69cbd089..252fdee2 100644 --- a/internal/channels/bitrix24/channel.go +++ b/internal/channels/bitrix24/channel.go @@ -56,6 +56,12 @@ type Channel struct { mentionMu sync.Mutex mentionRe *mentionMatcher + // reactions tracks the current status-reaction emoji per user message so + // OnReactionEvent can replace the previous one as the agent's status + // progresses (thinking → tool → done). Keyed by ":"; + // values are *reactionState. See reactions.go. + reactions sync.Map + // MCP lazy provisioner (Phase C). All fields nil / zero when // provisioning is disabled — channel then works exactly as before // (messages flow through without trying to mint MCP credentials). diff --git a/internal/channels/bitrix24/handle.go b/internal/channels/bitrix24/handle.go index c9d1c591..2c8ee702 100644 --- a/internal/channels/bitrix24/handle.go +++ b/internal/channels/bitrix24/handle.go @@ -303,6 +303,11 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) { // Bitrix replyId. MetaKeyMessageID: evt.Params.MessageID, MetaKeyVisibility: visibility, + // Generic run-tracking key the gateway consumer reads to thread + // streaming / status-reaction events back to this message (all other + // channels set it). Distinct from bitrix_message_id above (same value + // here, but that key is Bitrix-specific reply-link routing). + "message_id": evt.Params.MessageID, } // Echo the connector sender tag back on the reply so the Open Channel // connector routes the answer to the right external message: diff --git a/internal/channels/bitrix24/reactions.go b/internal/channels/bitrix24/reactions.go new file mode 100644 index 00000000..85865997 --- /dev/null +++ b/internal/channels/bitrix24/reactions.go @@ -0,0 +1,167 @@ +package bitrix24 + +import ( + "context" + "errors" + "log/slog" + "strconv" + "strings" + "sync" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/channels" +) + +// Compile-time guarantee the Bitrix24 channel provides status reactions. +var _ channels.ReactionChannel = (*Channel)(nil) + +// reactionDebounceInterval throttles intermediate status-reaction updates so a +// fast think→tool→tool sequence doesn't hammer the REST API. Terminal states +// (done/error) bypass the debounce so the final outcome always lands. +const reactionDebounceInterval = 700 * time.Millisecond + +// statusReaction maps a GoClaw agent status to a Bitrix24 v2 reaction code. +// Codes come from imbot.v2.Chat.Message.Reaction.add's documented set. +var statusReaction = map[string]string{ + "queued": "eyes", // 👀 waiting to process + "thinking": "thinkingFace", // 🤔 processing + "tool": "fire", // 🔥 running a tool + "coding": "fire", // 🔥 executing code + "web": "fire", // 🔥 browsing / API call + "done": "whiteHeavyCheckMark", // ✅ completed + "error": "crossMark", // ❌ failed + "stall": "sleepingSymbol", // 😴 no activity + "stallSoft": "sleepingSymbol", // 😴 + "stallHard": "flushedFace", // 😳 +} + +// reactionState tracks the current reaction code set on one user message so +// OnReactionEvent can replace it as the agent's status progresses. +type reactionState struct { + mu sync.Mutex + current string + lastUpdate time.Time +} + +// OnReactionEvent sets a status-reaction on the user's message to show agent +// progress (thinking → tool → done/error). Implements channels.ReactionChannel. +// messageID is the Bitrix MESSAGE_ID (numeric) of the message that started the run. +func (c *Channel) OnReactionEvent(ctx context.Context, chatID, messageID, status string) error { + level := c.cfg.ReactionLevel + if level == "" || level == "off" { + return nil + } + code, ok := statusReaction[status] + if !ok { + return nil + } + terminal := status == "done" || status == "error" + // "minimal" shows only the final outcome (done/error), not intermediate steps. + if level == "minimal" && !terminal { + return nil + } + + msgID, err := strconv.Atoi(strings.TrimSpace(messageID)) + if err != nil || msgID <= 0 { + return nil // not a Bitrix message id (e.g. connector synthetic id) + } + botID := c.BotID() + if botID <= 0 || c.Client() == nil { + return nil + } + + key := chatID + ":" + messageID + stVal, _ := c.reactions.LoadOrStore(key, &reactionState{}) + st := stVal.(*reactionState) + + st.mu.Lock() + defer st.mu.Unlock() + + if !terminal && time.Since(st.lastUpdate) < reactionDebounceInterval { + return nil + } + if st.current == code { + return nil + } + + // Bitrix has no "replace" — delete the previous reaction, then add the new. + if st.current != "" { + c.removeReaction(ctx, botID, msgID, st.current) + } + if err := c.addReaction(ctx, botID, msgID, code); err != nil { + slog.Debug("bitrix24 reaction: add failed", + "code", code, "message_id", msgID, "err", err) + return nil + } + st.current = code + st.lastUpdate = time.Now() + + // The run is over on terminal states — stop tracking this message. + if terminal { + c.reactions.Delete(key) + } + return nil +} + +// ClearReaction removes the current status-reaction from a message. +// Implements channels.ReactionChannel. +func (c *Channel) ClearReaction(ctx context.Context, chatID, messageID string) error { + key := chatID + ":" + messageID + stVal, ok := c.reactions.LoadAndDelete(key) + if !ok { + return nil + } + st := stVal.(*reactionState) + st.mu.Lock() + defer st.mu.Unlock() + if st.current == "" { + return nil + } + msgID, err := strconv.Atoi(strings.TrimSpace(messageID)) + if err != nil || msgID <= 0 { + return nil + } + if botID := c.BotID(); botID > 0 && c.Client() != nil { + c.removeReaction(ctx, botID, msgID, st.current) + } + st.current = "" + return nil +} + +// addReaction adds a bot reaction via imbot.v2.Chat.Message.Reaction.add. +// REACTION_ALREADY_SET is treated as success (the reaction is already present). +func (c *Channel) addReaction(ctx context.Context, botID, messageID int, code string) error { + _, err := c.Client().Call(ctx, "imbot.v2.Chat.Message.Reaction.add", map[string]any{ + "botId": botID, + "messageId": messageID, + "reaction": code, + }) + if err != nil && isReactionAlreadySet(err) { + return nil + } + return err +} + +// removeReaction removes a bot reaction via imbot.v2.Chat.Message.Reaction.delete. +// Best-effort: a failed delete is logged and swallowed (worst case a stale +// reaction lingers, which is cosmetic). +func (c *Channel) removeReaction(ctx context.Context, botID, messageID int, code string) { + if _, err := c.Client().Call(ctx, "imbot.v2.Chat.Message.Reaction.delete", map[string]any{ + "botId": botID, + "messageId": messageID, + "reaction": code, + }); err != nil { + slog.Debug("bitrix24 reaction: delete failed", + "code", code, "message_id", messageID, "err", err) + } +} + +// isReactionAlreadySet reports whether err is the benign REACTION_ALREADY_SET +// application error from Bitrix (the bot already reacted with this code). +func isReactionAlreadySet(err error) bool { + var apiErr *APIError + if errors.As(err, &apiErr) { + return apiErr.Code == "REACTION_ALREADY_SET" + } + return false +} diff --git a/internal/channels/bitrix24/reactions_test.go b/internal/channels/bitrix24/reactions_test.go new file mode 100644 index 00000000..3eef5585 --- /dev/null +++ b/internal/channels/bitrix24/reactions_test.go @@ -0,0 +1,129 @@ +package bitrix24 + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +type reactionCall struct { + op string // "add" | "delete" + reaction string +} + +// reactionRecorder serves the v2 reaction endpoints and records each call. +func reactionRecorder(t *testing.T) (*httptest.Server, func() []reactionCall) { + t.Helper() + var mu sync.Mutex + var calls []reactionCall + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = r.ParseForm() + op := "" + switch { + case strings.HasSuffix(r.URL.Path, "/rest/imbot.v2.Chat.Message.Reaction.add.json"): + op = "add" + case strings.HasSuffix(r.URL.Path, "/rest/imbot.v2.Chat.Message.Reaction.delete.json"): + op = "delete" + default: + t.Errorf("unexpected reaction path: %q", r.URL.Path) + } + mu.Lock() + calls = append(calls, reactionCall{op: op, reaction: r.Form.Get("reaction")}) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"result": true}) + })) + return srv, func() []reactionCall { + mu.Lock() + defer mu.Unlock() + out := make([]reactionCall, len(calls)) + copy(out, calls) + return out + } +} + +func TestOnReactionEvent_FullFlow_AddThenReplaceViaV2(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "full" + ctx := context.Background() + + // thinking → add thinkingFace + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") + // done (terminal, bypasses debounce) → delete thinkingFace + add whiteHeavyCheckMark + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "done") + + calls := snapshot() + if len(calls) != 3 { + t.Fatalf("expected 3 reaction calls (add, delete, add), got %d: %+v", len(calls), calls) + } + if calls[0].op != "add" || calls[0].reaction != "thinkingFace" { + t.Errorf("call[0] = %+v; want add thinkingFace", calls[0]) + } + if calls[1].op != "delete" || calls[1].reaction != "thinkingFace" { + t.Errorf("call[1] = %+v; want delete thinkingFace", calls[1]) + } + if calls[2].op != "add" || calls[2].reaction != "whiteHeavyCheckMark" { + t.Errorf("call[2] = %+v; want add whiteHeavyCheckMark", calls[2]) + } +} + +func TestOnReactionEvent_MinimalSkipsIntermediate(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "minimal" + ctx := context.Background() + + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") // skipped in minimal + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "error") // terminal → shown + + calls := snapshot() + if len(calls) != 1 || calls[0].op != "add" || calls[0].reaction != "crossMark" { + t.Fatalf("minimal should send only the terminal reaction; got %+v", calls) + } +} + +func TestOnReactionEvent_OffAndInvalidAreNoOp(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + + // off level + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "off" + _ = bc.OnReactionEvent(context.Background(), "chat100", "5001", "done") + + // full level but non-numeric message id (connector synthetic id) + bc2 := newChannelWithBoundPortal(t, srv) + bc2.cfg.ReactionLevel = "full" + _ = bc2.OnReactionEvent(context.Background(), "chat100", "not-a-number", "done") + + // full level, unknown status + bc2.cfg.ReactionLevel = "full" + _ = bc2.OnReactionEvent(context.Background(), "chat100", "5001", "bogus-status") + + if calls := snapshot(); len(calls) != 0 { + t.Fatalf("expected no reaction calls for off/invalid/unknown, got %+v", calls) + } +} + +func TestClearReaction_RemovesCurrent(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "full" + ctx := context.Background() + + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") // add thinkingFace + _ = bc.ClearReaction(ctx, "chat100", "5001") // delete thinkingFace + + calls := snapshot() + if len(calls) != 2 || calls[1].op != "delete" || calls[1].reaction != "thinkingFace" { + t.Fatalf("expected add then delete thinkingFace; got %+v", calls) + } +}