mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-13 08:15:29 +00:00
feat(bitrix24): status-reaction support via imbot.v2 reactions [B24:2794] (#1446)
Bitrix24 was the only channel that never showed the agent's progress as an emoji reaction on the user's message (Telegram/Slack/Feishu/Discord/... all do). Two things were missing: the channel didn't implement channels.ReactionChannel, and its inbound metadata omitted the generic "message_id" key the gateway consumer reads to thread status/streaming events back to a message. - handle.go: set meta["message_id"] = MESSAGE_ID (every other channel does this; without it the run has no message id to react to). Kept the existing bitrix_message_id key for reply-link routing. - reactions.go (new): implement OnReactionEvent + ClearReaction. Map agent status → a Bitrix v2 reaction code (thinkingFace / fire / whiteHeavyCheckMark / crossMark / ...), debounce intermediate updates (terminal states bypass), and replace the previous reaction via imbot.v2.Chat.Message.Reaction.delete + .add (Bitrix has no in-place replace). REACTION_ALREADY_SET is treated as success. ReactionLevel off/minimal/full already existed in the channel config. - channel.go: add the per-message reactions state map + compile-time assertion. Tests: full flow (add → delete+add on status change), minimal skips intermediate, off/invalid/unknown are no-ops, ClearReaction removes current. Co-authored-by: DangTinh311 <dangtinh31193@gmail.com>
This commit is contained in:
1 parent
6cf48a6b08
commit
d07dca440a
4 files changed
+307
No files matched your search
@@ -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 "<chatID>:<messageID>";
|
||||
// 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).
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user