fix(bitrix24): OpenLine greeting and MCP provisioning (#1284)

* fix(bitrix24): skip join greeting in open channels [B24:2794]

* wip: rework bitrix24 mcp provisioning + skills prompt section (tam thoi, se lam lai) [B24:2794]

---------

Co-authored-by: DangTinh311 <dangtinh31193@gmail.com>
This commit is contained in:
SYNITYandDangTinh311 authored and GitHub committed 2026-06-29 08:54:00 +07:00
1 parent 0184678dc3
commit 23a4b7407c
7 files changed
+191 -71

No files matched your search

+5 -20
View File
@@ -711,29 +711,14 @@ func buildSkillsSection(skillsSummary string, hasSkillSearch, hasSkillManage boo
if skillsSummary != "" {
// Inline mode: skills XML is in the prompt (like TS).
// Agent scans <available_skills> descriptions directly.
lines = append(lines,
"## Skills (mandatory)",
"",
"Before replying, scan `<available_skills>` below.",
"If a skill clearly applies, read its SKILL.md at the `<location>` path with `read_file`, then follow it.",
"If multiple could apply, choose the most specific one. Never read more than one skill up front.",
"If none apply, proceed normally.",
"",
skillsSummary,
"",
)
lines = append(lines, "## Skills (mandatory)", "")
lines = append(lines, skillLoadingProtocolLines()...)
lines = append(lines, skillsSummary, "")
} else if hasSkillSearch {
// Search mode: too many skills to inline, agent uses skill_search tool.
lines = append(lines, "## Skills (mandatory)", "")
lines = append(lines, skillLoadingProtocolLines()...)
lines = append(lines,
"## Skills (mandatory)",
"",
"Before replying, check if a skill applies:",
"1. Run `skill_search` with **English keywords** describing the domain (e.g. \"weather\", \"translate\", \"github\").",
" Even if the user writes in another language, always search in English.",
"2. If a match is found, read its SKILL.md at the returned `location` with `read_file`, then follow it.",
"3. If multiple skills match, choose the most specific one. Never read more than one skill up front.",
"4. If no match, proceed normally.",
"",
"Constraints:",
"- Prefer `skill_search` over `browser` or `web_search` when the domain might have a skill.",
"- If skill_search returns no results, fall back to other tools freely.",
+19 -5
View File
@@ -271,11 +271,7 @@ func buildSkillsHybridSection(pinnedSummary string, hasSearch, hasManage bool) [
)
}
if hasSearch {
lines = append(lines,
"For other skills, run `skill_search` with **English keywords** describing the domain.",
"If a match is found, read its SKILL.md at the returned location, then follow it.",
"",
)
lines = append(lines, skillLoadingProtocolLines()...)
}
if hasManage {
lines = append(lines, "### Skill Creation", "",
@@ -286,6 +282,24 @@ func buildSkillsHybridSection(pinnedSummary string, hasSearch, hasManage bool) [
return lines
}
// skillLoadingProtocolLines is the shared, mechanical protocol for activating
// and loading skills. Centralized so every skills section (hybrid, inline,
// search) gives the model identical path-safe steps: read the EXACT location
// verbatim, never guess a SKILL.md path.
func skillLoadingProtocolLines() []string {
return []string{
"Skill loading protocol — follow exactly:",
"1. Check the `<available_skills>` list in this prompt.",
"2. If a clearly matching skill is listed there: call `use_skill(\"<name>\")`; then, if `use_skill` did not return the skill's full instructions, call `read_file` with the EXACT `<location>` from `<available_skills>` (copy it verbatim, including the leading `/`); then follow the SKILL.md.",
"3. If no clear match is listed: call `skill_search` with **English keywords**, pick the most specific result, call `use_skill(\"<name>\")`, then `read_file` the EXACT `location` returned by `skill_search`; then follow the SKILL.md.",
"4. Never guess, construct, or modify SKILL.md paths — only use a `location` value copied verbatim.",
"5. If `read_file` fails, do not invent another path: call `skill_search` again by skill name and read the `location` it returns.",
"6. `use_skill` only activates/traces the skill; it does not load the instructions unless it returns the full content.",
"7. Read at most one skill up front; if none apply, proceed normally.",
"",
}
}
// buildSandboxSection creates the "## Sandbox" section matching TS system-prompt.ts lines 476-519.
func buildSandboxSection(cfg SystemPromptConfig) []string {
lines := []string{
+15 -3
View File
@@ -118,8 +118,7 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) {
// require_mention is true, reply to everything when false) or silently
// merge into the "direct" path. Recognise it up-front so the gate below
// can apply the right policy.
isOpenChannel := strings.EqualFold(evt.Params.MessageType, "L") ||
strings.EqualFold(evt.Params.ChatEntityType, "LINES")
isOpenChannel := isOpenChannelParams(&evt.Params)
// Force group routing for Open Channel so the existing mention-strip /
// readable-mention pipeline below runs, and so the session key includes
// the chat id instead of dumping every participant into the "direct"
@@ -374,7 +373,7 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) {
// the MCP server's tools, which is strictly better UX than the channel
// denying the message. The typed errors let tests assert behavior
// without string matching.
if err := c.provisionIfMissing(ctx, senderID, evt.Auth); err != nil {
if err := c.provisionIfMissing(ctx, senderID, evt.Params.FromIsConnector, evt.Auth); err != nil {
switch {
case errors.Is(err, ErrProvisionDisabled),
errors.Is(err, ErrProvisionSkippedOpenChannel),
@@ -432,6 +431,14 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) {
// chat. Failure is non-fatal — the agent will still respond to the user's
// first real message.
func (c *Channel) handleJoin(ctx context.Context, evt *Event) {
// Open Channel sessions are customer-facing. Keep join events silent so
// adding the bot does not push an unsolicited greeting to the customer.
if isOpenChannelParams(&evt.Params) {
slog.Debug("bitrix24: welcome message skipped for open channel",
"dialog_id", evt.Params.DialogID)
return
}
client := c.Client()
botID := c.BotID()
if client == nil || botID <= 0 {
@@ -451,6 +458,11 @@ func (c *Channel) handleJoin(ctx context.Context, evt *Event) {
}
}
func isOpenChannelParams(p *EventParams) bool {
return p != nil && (strings.EqualFold(p.MessageType, "L") ||
strings.EqualFold(p.ChatEntityType, "LINES"))
}
// isMentionedParams checks all three sources Bitrix24 may use to convey a
// bot mention, in authority order:
//
+21
View File
@@ -653,6 +653,27 @@ func TestHandleMessage_Blocked_DoesNotCollectContact(t *testing.T) {
// --- Open Channel (MESSAGE_TYPE="L") gating ---------------------------------
func TestIsOpenChannelParams(t *testing.T) {
tests := []struct {
name string
params *EventParams
want bool
}{
{name: "nil", params: nil, want: false},
{name: "message type L", params: &EventParams{MessageType: "L"}, want: true},
{name: "lines entity", params: &EventParams{ChatEntityType: "LINES"}, want: true},
{name: "case insensitive", params: &EventParams{ChatEntityType: "lines"}, want: true},
{name: "regular group", params: &EventParams{MessageType: "C"}, want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := isOpenChannelParams(tt.params); got != tt.want {
t.Errorf("isOpenChannelParams() = %v; want %v", got, tt.want)
}
})
}
}
// TestHandleMessage_OpenChannel_ConnectorWithoutMentionDropped covers the
// customer side of the Open Channel gate on a GROUP-style connector
// (Zalo personal) without an explicit mention: customers chatting amongst
+12
View File
@@ -433,6 +433,18 @@ func (p *Portal) Exchange(ctx context.Context, code string) error {
return p.persistState(context.WithoutCancel(ctx))
}
// RefreshUserToken refreshes an arbitrary per-user Bitrix OAuth token using
// this portal's client_id/secret (grant_type=refresh_token). Used by the MCP
// provisioner to renew a stored per-user refresh_token when an inbound event
// doesn't carry OAuth tokens. Keeps client_id/secret encapsulated in Portal.
//
// NOTE: this does NOT touch the portal's own token state — it operates purely
// on the caller-supplied refresh_token. Bitrix rotates the refresh_token on
// each call, so the caller MUST persist the returned pair.
func (p *Portal) RefreshUserToken(ctx context.Context, refreshToken string) (*TokenResponse, error) {
return p.client.RefreshToken(ctx, p.creds.ClientID, p.creds.ClientSecret, refreshToken)
}
func (p *Portal) validateTokenResponseIdentity(flow string, tr *TokenResponse) error {
if tr == nil {
return fmt.Errorf("bitrix24 %s: nil token response", flow)
+100 -30
View File
@@ -139,11 +139,14 @@ func (c *Channel) initMCPProvisioner(ctx context.Context) error {
// HandleMessage regardless, so user messages always get processed.
//
// Called from handleMessage after EnsureContact, before HandleMessage.
func (c *Channel) provisionIfMissing(ctx context.Context, userID string, auth EventAuth) error {
// Skip #1: Open Channel bot. No per-user credentials for transient
// customers — see type docstring.
if c.IsOpenChannelBot() {
slog.Debug("bitrix24 mcp: provision skip open channel", "channel", c.Name(), "user_id", userID)
func (c *Channel) provisionIfMissing(ctx context.Context, userID string, fromConnector bool, auth EventAuth) error {
// Skip #1: Open Channel message from an external connector customer.
// Transient customers reach the bot via a connector (Zalo/FB/etc.) and
// report IS_CONNECTOR=Y — they are not Bitrix users and have no per-user
// OAuth, so skip them. Internal staff who join an Open Channel session
// report IS_CONNECTOR=N and DO have OAuth, so they must be provisioned.
if c.IsOpenChannelBot() && fromConnector {
slog.Debug("bitrix24 mcp: provision skip open channel connector", "channel", c.Name(), "user_id", userID)
return ErrProvisionSkippedOpenChannel
}
@@ -210,11 +213,15 @@ func (c *Channel) provisionIfMissing(ctx context.Context, userID string, auth Ev
// so a failed auto-onboard can trigger 3–5 attempts per second
// without this guard. TTL = 60s covers the retry burst window and
// the typical "MCP server blip" recovery time.
if c.isMCPProvisionDebounced(c.mcpServerID, userID) {
// Atomic check-and-set: an aggressive Bitrix retry burst can deliver
// several events for the same user near-simultaneously. A separate
// check-then-mark leaves a TOCTOU gap where two of them both pass the gate
// and both run a self-refresh, double-rotating the refresh_token (the
// second then fails on the now-consumed token). Acquire under one lock.
if !c.tryAcquireMCPProvision(c.mcpServerID, userID) {
slog.Warn("bitrix24 mcp: provision debounced", "channel", c.Name(), "user_id", userID)
return ErrProvisionDebounced
}
c.markMCPProvisionDebounced(c.mcpServerID, userID)
// OAuth tokens are plumbed through the webhook event's auth block —
// MCP server uses them to call Bitrix REST on behalf of this user.
@@ -222,7 +229,15 @@ func (c *Channel) provisionIfMissing(ctx context.Context, userID string, auth Ev
// but surface them here with a clearer error so operators don't have
// to trace to mcp_client.go.
if auth.Domain == "" || auth.AccessToken == "" || auth.RefreshToken == "" {
return fmt.Errorf("bitrix24 mcp: incomplete auth block (domain/access_token/refresh_token required)")
// Direct/group chatbot events don't carry OAuth tokens in the top-level
// auth[] block — Bitrix only ships them on Open Channel events. If this
// user already has stored credentials, refresh the USER-context token
// from the stored refresh_token instead of failing, so per-user MCP
// stays alive without depending on the event carrying tokens.
if err := c.selfRefreshUserCreds(ctx, userID, existing); err != nil {
return err
}
return nil
}
resp, err := c.mcpClient.autoOnboard(ctx, autoOnboardRequest{
@@ -263,36 +278,91 @@ func (c *Channel) provisionIfMissing(ctx context.Context, userID string, auth Ev
return nil
}
// isMCPProvisionDebounced reports whether a provisioning attempt for
// (serverID, userID) ran within the last mcpProvisionDebounceTTL. Also
// opportunistically prunes expired entries so the map doesn't grow
// unbounded across long-lived channels.
func (c *Channel) isMCPProvisionDebounced(serverID uuid.UUID, userID string) bool {
c.mcpProvMu.Lock()
defer c.mcpProvMu.Unlock()
key := mcpDebounceKey{serverID: serverID, userID: userID}
if ts, ok := c.mcpDebounce[key]; ok {
if time.Since(ts) < mcpProvisionDebounceTTL {
return true
}
// Expired — delete so the map stays lean. Cheap to do here since
// we're already holding the lock for the check.
delete(c.mcpDebounce, key)
// selfRefreshUserCreds renews a user's per-user Bitrix MCP credentials WITHOUT
// relying on an inbound event carrying OAuth tokens. It refreshes the stored
// refresh_token via OAuth (grant_type=refresh_token, using the portal's
// client_id/secret), re-onboards the MCP server with the rotated tokens, and
// persists them. Bitrix rotates the refresh_token on every call, so persisting
// the new pair is mandatory — otherwise the next refresh would use a dead token.
//
// Returns an error (surfaced to the caller as a degradation notice) when there
// are no stored credentials to refresh, the stored refresh_token is dead
// (invalid_grant → user must re-authorize), or onboarding/persisting fails.
func (c *Channel) selfRefreshUserCreds(ctx context.Context, userID string, existing *store.MCPUserCredentials) error {
if existing == nil || existing.Env == nil {
return fmt.Errorf("bitrix24 mcp: incomplete auth block and no stored credentials to refresh")
}
return false
refreshToken := strings.TrimSpace(existing.Env["BITRIX_REFRESH_TOKEN"])
domain := strings.TrimSpace(existing.Env["BITRIX_DOMAIN"])
if refreshToken == "" || domain == "" {
return fmt.Errorf("bitrix24 mcp: incomplete auth block and stored credentials missing domain/refresh_token")
}
if c.portal == nil {
return fmt.Errorf("bitrix24 mcp: portal unavailable for self-refresh")
}
tr, err := c.portal.RefreshUserToken(ctx, refreshToken)
if err != nil {
slog.Warn("bitrix24 mcp: self-refresh oauth failed", "channel", c.Name(), "user_id", userID, "err", err)
return fmt.Errorf("bitrix24 mcp: self-refresh oauth failed: %w", err)
}
resp, err := c.mcpClient.autoOnboard(ctx, autoOnboardRequest{
Domain: domain,
BitrixUserID: userID,
AccessToken: tr.AccessToken,
RefreshToken: tr.RefreshToken,
ExpiresIn: int(tr.ExpiresIn),
})
if err != nil {
slog.Warn("bitrix24 mcp: self-refresh re-onboard failed", "channel", c.Name(), "user_id", userID, "err", err)
return fmt.Errorf("bitrix24 mcp: self-refresh re-onboard failed: %w", err)
}
expiresAt := time.Now().Add(time.Duration(tr.ExpiresIn) * time.Second).UTC().Format(time.RFC3339)
creds := store.MCPUserCredentials{
APIKey: resp.APIKey,
Env: map[string]string{
"BITRIX_DOMAIN": domain,
"BITRIX_ACCESS_TOKEN": tr.AccessToken,
"BITRIX_REFRESH_TOKEN": tr.RefreshToken, // rotated — persist the new one
"BITRIX_EXPIRES_AT": expiresAt,
},
}
// Decouple persist from the caller's context: once Bitrix rotated the
// refresh_token, a canceled request ctx must NOT drop the new token to the
// store — the old token is already dead and would strand the user until
// re-authorization. Mirrors Portal.refreshLocked's WithoutCancel rationale.
if err := c.mcpStore.SetUserCredentials(context.WithoutCancel(ctx), c.mcpServerID, userID, creds); err != nil {
return fmt.Errorf("bitrix24 mcp: persist self-refreshed credentials: %w", err)
}
slog.Info("bitrix24 mcp: self-refreshed user credentials",
"channel", c.Name(), "user_id", userID, "mcp_server_id", c.mcpServerID, "created", resp.Created)
return nil
}
func (c *Channel) markMCPProvisionDebounced(serverID uuid.UUID, userID string) {
// tryAcquireMCPProvision atomically checks the debounce window for
// (serverID, userID) and, if clear, records a fresh attempt — all under a
// single lock so concurrent webhook-retry events can't both pass the gate
// (which would double-rotate the refresh_token in selfRefreshUserCreds).
// Returns true when the caller acquired the slot (NOT debounced), false
// otherwise. Prunes the entry on the way through to keep the map lean.
func (c *Channel) tryAcquireMCPProvision(serverID uuid.UUID, userID string) bool {
c.mcpProvMu.Lock()
defer c.mcpProvMu.Unlock()
if c.mcpDebounce == nil {
// Defensive: initMCPProvisioner allocates this, but if some code
// path bypassed init (e.g. test that constructs Channel directly
// and then calls provisionIfMissing with provisioning enabled),
// a nil map write would panic. Allocate on demand instead.
// Defensive: initMCPProvisioner allocates this, but if some code path
// bypassed init (e.g. a test that constructs Channel directly), a nil
// map write would panic. Allocate on demand instead.
c.mcpDebounce = make(map[mcpDebounceKey]time.Time)
}
c.mcpDebounce[mcpDebounceKey{serverID: serverID, userID: userID}] = time.Now()
key := mcpDebounceKey{serverID: serverID, userID: userID}
if ts, ok := c.mcpDebounce[key]; ok && time.Since(ts) < mcpProvisionDebounceTTL {
return false // still within the debounce window
}
c.mcpDebounce[key] = time.Now()
return true
}
// mcpUserNotifyDebounceTTL controls how often a single user can receive
+19 -13
View File
@@ -203,12 +203,18 @@ func TestProvisionIfMissing_OpenChannelBot_Skipped(t *testing.T) {
}
bc := ch.(*Channel)
err = bc.provisionIfMissing(context.Background(), "42", validAuth())
// IS_CONNECTOR=Y (external connector customer) → skipped: not a Bitrix user.
err = bc.provisionIfMissing(context.Background(), "42", true, validAuth())
if !errors.Is(err, ErrProvisionSkippedOpenChannel) {
t.Fatalf("err = %v; want ErrProvisionSkippedOpenChannel", err)
t.Fatalf("connector message: err = %v; want ErrProvisionSkippedOpenChannel", err)
}
if mcpStore.getUserCallCount != 0 {
t.Errorf("Open Channel bot must not hit MCP store; got %d GetUserCredentials calls", mcpStore.getUserCallCount)
t.Errorf("Open Channel connector message must not hit MCP store; got %d GetUserCredentials calls", mcpStore.getUserCallCount)
}
// IS_CONNECTOR=N (internal staff in an Open Channel) → must NOT connector-skip.
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); errors.Is(err, ErrProvisionSkippedOpenChannel) {
t.Fatalf("internal staff (IS_CONNECTOR=N) must not be connector-skipped; got %v", err)
}
}
@@ -230,7 +236,7 @@ func TestProvisionIfMissing_Disabled(t *testing.T) {
}
bc := ch.(*Channel)
err = bc.provisionIfMissing(context.Background(), "42", validAuth())
err = bc.provisionIfMissing(context.Background(), "42", false, validAuth())
if !errors.Is(err, ErrProvisionDisabled) {
t.Fatalf("err = %v; want ErrProvisionDisabled", err)
}
@@ -264,7 +270,7 @@ func TestProvisionIfMissing_ExistingCreds_NoHTTP(t *testing.T) {
},
}
if err := bc.provisionIfMissing(context.Background(), "42", validAuth()); err != nil {
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); err != nil {
t.Fatalf("err = %v; want nil", err)
}
if httpCalls != 0 {
@@ -302,7 +308,7 @@ func TestProvisionIfMissing_NearExpiry_RefreshHTTP(t *testing.T) {
},
}
if err := bc.provisionIfMissing(context.Background(), "42", validAuth()); err != nil {
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); err != nil {
t.Fatalf("err = %v; want nil", err)
}
if httpCalls != 1 {
@@ -335,7 +341,7 @@ func TestProvisionIfMissing_LegacyNoExpiry_RefreshHTTP(t *testing.T) {
APIKey: "legacy-key",
}
if err := bc.provisionIfMissing(context.Background(), "42", validAuth()); err != nil {
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); err != nil {
t.Fatalf("err = %v; want nil", err)
}
if httpCalls != 1 {
@@ -369,7 +375,7 @@ func TestProvisionIfMissing_WarmExpiry_NoHTTP(t *testing.T) {
},
}
if err := bc.provisionIfMissing(context.Background(), "42", validAuth()); err != nil {
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); err != nil {
t.Fatalf("err = %v; want nil", err)
}
if httpCalls != 0 {
@@ -393,7 +399,7 @@ func TestProvisionIfMissing_MintAndPersist(t *testing.T) {
bc := newProvisionerTestChannel(t, mcpStore, srv.URL, "B")
before := time.Now()
err := bc.provisionIfMissing(context.Background(), "42", validAuth())
err := bc.provisionIfMissing(context.Background(), "42", false, validAuth())
if err != nil {
t.Fatalf("provisionIfMissing: %v", err)
}
@@ -451,7 +457,7 @@ func TestProvisionIfMissing_Debounce(t *testing.T) {
bc := newProvisionerTestChannel(t, mcpStore, srv.URL, "B")
// First attempt succeeds and marks the debounce.
if err := bc.provisionIfMissing(context.Background(), "42", validAuth()); err != nil {
if err := bc.provisionIfMissing(context.Background(), "42", false, validAuth()); err != nil {
t.Fatalf("first attempt: %v", err)
}
if httpCalls != 1 {
@@ -466,7 +472,7 @@ func TestProvisionIfMissing_Debounce(t *testing.T) {
delete(mcpStore.userCreds, credKey(bc.mcpServerID, "42"))
mcpStore.mu.Unlock()
err := bc.provisionIfMissing(context.Background(), "42", validAuth())
err := bc.provisionIfMissing(context.Background(), "42", false, validAuth())
if !errors.Is(err, ErrProvisionDebounced) {
t.Fatalf("second attempt: %v; want ErrProvisionDebounced", err)
}
@@ -489,7 +495,7 @@ func TestProvisionIfMissing_HTTPFailure_Surfaces(t *testing.T) {
mcpStore := newFakeMCPStore()
bc := newProvisionerTestChannel(t, mcpStore, srv.URL, "B")
err := bc.provisionIfMissing(context.Background(), "42", validAuth())
err := bc.provisionIfMissing(context.Background(), "42", false, validAuth())
if err == nil {
t.Fatal("401 from MCP must produce an error")
}
@@ -534,7 +540,7 @@ func TestProvisionIfMissing_MissingAuthBlock(t *testing.T) {
before := httpCalls
// Use a fresh userID per subcase so the debounce from a prior
// case doesn't mask a regression.
err := bc.provisionIfMissing(context.Background(), tc.name, tc.auth)
err := bc.provisionIfMissing(context.Background(), tc.name, false, tc.auth)
if err == nil {
t.Fatalf("missing %s should fail", tc.name)
}