wangjunxiao/nanoclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(channels): shrink bridge shouldEngage to flood gate + subscribe signal

Browse Source 
Before this change the bridge and the router both owned engage_mode
policy. Bridge's shouldEngage had a full switch over mention /
mention-sticky / pattern + source-based rules + engage_pattern regex
test + ignored_message_policy accumulate fallback. Router's
evaluateEngage had the same switch against the same fields. Two
parallel logic paths with subtle vocabulary differences (bridge: "which
SDK handler fired"; router: "what isMention says"). Every time we
touched one we had to reason about the other — the Telegram
hasMention bug and the "pattern mode silently drops in group chats"
bug were both drift between the two.

Refactor to one place. Router keeps all per-wiring policy — engage
mode, pattern regex, sender scope, ignored-message policy — unchanged.
Bridge drops to a coarse flood gate + subscribe signal:

  - forward: does this channel have ANY wiring? Forward if yes.
    Unknown channels still forward for subscribed/mention/dm (they may
    be newly auto-created, or will trigger the coming
    channel-registration flow). Unknown channels DROP for new-message
    so we don't flood from every unsubscribed thread the bot happens
    to sit in.

  - stickySubscribe: any mention-sticky wiring on the channel AND the
    source is mention or dm. Coarse union — subscribe is idempotent
    and one call serves every sticky wiring.

The `text` param on shouldEngage is gone (pattern regex lives in the
router now). Four bridge handler sites simplify accordingly. messageToInbound
still carries the SDK-confirmed isMention flag through to the router
unchanged.

Behavioral delta: pure-mention-wired channels (no pattern, no
accumulate) will now see every plain group message reach the router
before being dropped there, where before the bridge dropped at the
transport boundary. Extra DB lookup per dropped message in this
specific case; acceptable for the cleaner seam and can be optimized
back at the bridge if it ever matters in practice.

Bridge tests prune the 10 engage_mode-specific cases that covered
logic now owned by evaluateEngage in the router (host-core.test.ts
covers it end-to-end through routeInbound). Bridge tests keep only
what's bridge-specific: the flood gate and the stickySubscribe
coarse union.

172 tests pass (was 182 — net -10 redundant bridge tests).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gavrielc
2026-04-20 13:32:08 +03:00
parent 68058cbc4a
commit b15972284b
2 changed files with 98 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
src/channels/chat-sdk-bridge.ts
View File

@@ -85,94 +85,46 @@ export interface ChatSdkBridgeConfig {
export type EngageSource = 'subscribed' | 'mention' | 'dm' | 'new-message';
/**
* Should a message from (channelId, source, text) be forwarded to the host,
* and if so, should the bridge subscribe the thread?
* Bridge-level forwarding decision — a coarse flood gate, not policy.
*
* The router owns per-wiring engage_mode / engage_pattern / sender_scope /
* ignored_message_policy (see `evaluateEngage` in src/router.ts). The bridge
* only answers two questions:
*
* 1. `forward` — is this message worth sending to the host at all?
* - Known channel (any wiring): yes. Router will decide what engages /
* accumulates / drops per wiring.
* - Unknown channel: yes for subscribed / mention / DM (triggers the
* router's auto-create or channel-registration flow); no for
* `new-message`. onNewMessage(/./, …) fires for every message in
* every unsubscribed thread the bot can see, including channels the
* bot merely joined but was never wired to — forwarding everything
* would flood the host.
*
* 2. `stickySubscribe` — should the bridge call `thread.subscribe()`?
* - Yes if ANY wiring on this channel is mention-sticky AND the
* source is an actual mention / DM. Coarse (no per-wiring picking)
* but harmless: subscription is idempotent and one call serves
* every mention-sticky wiring on the channel. Once subscribed,
* follow-ups route through onSubscribedMessage.
*
* Exported for testability — see `chat-sdk-bridge.test.ts`.
*
* We take the union across wired agents: if any wiring would engage OR any
* wiring has `ignoredMessagePolicy='accumulate'`, the message is forwarded.
* The host router then does the per-wiring decision in `deliverToAgent` —
* engaging agents get `trigger=1` (wake), accumulating agents get
* `trigger=0` (store as context, don't wake), drop-policy agents are
* skipped (see `src/router.ts` routeInbound fan-out).
*
* `stickySubscribe` is only set when an actual engage happens (not just
* accumulate) — subscribing a thread we'd only silently accumulate on would
* misrepresent the bot's presence to other users.
*/
export function shouldEngage(
conversations: Map<string, ConversationConfig[]>,
channelId: string,
source: EngageSource,
text: string,
): { forward: boolean; stickySubscribe: boolean } {
const configs = conversations.get(channelId);
// Unknown conversation — behavior diverges by source:
// - subscribed/mention/dm: forward anyway. These paths imply some
// prior engagement (subscribe, @mention, DM open) and may be a new
// group that hasn't been registered yet; central routing will log +
// drop cleanly.
// - new-message: DROP. `onNewMessage(/./, …)` fires for every message
// in every unsubscribed thread the bot can see — including channels
// the bot is merely *present* in but not wired to. Forwarding
// everything would flood the host.
if (!configs || configs.length === 0) {
return { forward: source !== 'new-message', stickySubscribe: false };
}
let engage = false;
let accumulate = false;
let stickySubscribe = false;
const stickySubscribe =
(source === 'mention' || source === 'dm') && configs.some((cfg) => cfg.engageMode === 'mention-sticky');
for (const cfg of configs) {
let cfgEngages = false;
switch (cfg.engageMode) {
case 'mention':
if (source === 'mention' || source === 'dm') cfgEngages = true;
break;
case 'mention-sticky':
if (source === 'mention' || source === 'dm') {
cfgEngages = true;
stickySubscribe = true;
} else if (source === 'subscribed') {
// Thread was already subscribed on a prior mention — treat as
// engage-all so follow-ups in the thread reach the agent.
cfgEngages = true;
}
// source='new-message' → does not engage (requires explicit mention
// to start). Accumulate policy is evaluated below if set.
break;
case 'pattern': {
// Pattern evaluates on any source that delivers a plain message —
// including new-message, which is the whole reason we registered
// onNewMessage(/./). For mention/dm-delivered messages we still
// test the regex (historical behavior), so pattern='foo' wirings
// only fire on mentions whose text contains 'foo'.
const pattern = cfg.engagePattern ?? '.';
try {
if (pattern === '.' || new RegExp(pattern).test(text)) cfgEngages = true;
} catch {
// Invalid regex → fail open so the admin can see something is
// happening and fix the pattern.
cfgEngages = true;
}
break;
}
}
if (cfgEngages) {
engage = true;
} else if (cfg.ignoredMessagePolicy === 'accumulate') {
// Wiring doesn't engage on this message but wants it captured as
// context for its session — forward so the router can write it with
// trigger=0.
accumulate = true;
}
}
return { forward: engage || accumulate, stickySubscribe };
return { forward: true, stickySubscribe };
}
export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter {
@@ -202,12 +154,8 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
return map;
}
function engageDecision(
channelId: string,
source: EngageSource,
text: string,
): { forward: boolean; stickySubscribe: boolean } {
return shouldEngage(conversations, channelId, source, text);
function engageDecision(channelId: string, source: EngageSource): { forward: boolean; stickySubscribe: boolean } {
return shouldEngage(conversations, channelId, source);
}
async function messageToInbound(message: ChatMessage, isMention: boolean): Promise<InboundMessage> {
@@ -289,46 +237,38 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
logger: 'silent',
});
// Subscribed threads — the conversation is already active (via prior
// mention-sticky engagement or admin wiring). Gate on engageMode so a
// plain 'mention' wiring doesn't keep firing after a one-off mention.
// Four SDK dispatch paths — bridge just forwards; router does all
// per-wiring engage / accumulate / drop decisions. isMention is the
// load-bearing signal (see evaluateEngage in src/router.ts).
// Subscribed threads — every message in a thread we've previously
// engaged. Carry the SDK's `message.isMention` through so mention-mode
// wirings still fire on in-thread mentions.
chat.onSubscribedMessage(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id);
const text = typeof message.text === 'string' ? message.text : '';
const decision = engageDecision(channelId, 'subscribed', text);
const decision = engageDecision(channelId, 'subscribed');
if (!decision.forward) return;
// Subscribed path: the SDK sets message.isMention when the bot was
// @-mentioned in an already-subscribed thread (docs at
// handling-events.mdx). Forward it verbatim.
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, message.isMention === true));
});
// @mention in an unsubscribed thread — always engage; subscribe only
// if the wiring is 'mention-sticky'.
// @mention in an unsubscribed thread — SDK-confirmed bot mention.
chat.onNewMention(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id);
const text = typeof message.text === 'string' ? message.text : '';
const decision = engageDecision(channelId, 'mention', text);
const decision = engageDecision(channelId, 'mention');
if (!decision.forward) return;
// onNewMention only fires when the SDK confirms the bot was mentioned.
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true));
if (decision.stickySubscribe) {
await thread.subscribe();
}
});
// DMs — apply engage rules too, but DMs typically default to pattern='.'
// at setup time so this is a pass-through in practice. sticky subscribe
// follows the same rule as a group mention.
//
// Thread id is passed through so sub-thread context reaches delivery
// (Slack users can open threads inside a DM). The router collapses DM
// sub-threads to one session (is_group=0 short-circuits the per-thread
// escalation).
// DMs — by definition addressed to the bot. Thread id flows through
// so sub-thread context reaches delivery (Slack users can open threads
// inside a DM). Router collapses DM sub-threads to one session via
// is_group=0 short-circuit.
chat.onDirectMessage(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id);
const text = typeof message.text === 'string' ? message.text : '';
const decision = engageDecision(channelId, 'dm', text);
const decision = engageDecision(channelId, 'dm');
log.info('Inbound DM received', {
adapter: adapter.name,
channelId,
@@ -337,35 +277,25 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
forward: decision.forward,
});
if (!decision.forward) return;
// A DM is by definition addressed to the bot — treat as a mention
// for routing purposes. `mention` / `mention-sticky` wirings fire.
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true));
if (decision.stickySubscribe) {
await thread.subscribe();
}
});
// Plain (non-mention, non-DM) messages in unsubscribed threads.
// Plain messages in unsubscribed threads.
//
// Chat SDK dispatch (handling-events.mdx §"Handler dispatch order"):
// subscribed threads → onSubscribedMessage; unsubscribed + mention →
// onNewMention; unsubscribed + pattern match → onNewMessage. Dispatch
// is exclusive — at most one handler fires per message.
//
// Without this handler, `engage_mode='pattern'` is silently dropped in
// unsubscribed group threads because the SDK never surfaces the
// message anywhere else. Registering with `/./` lets every wired
// conversation's regex be evaluated in our `shouldEngage` — unknown
// conversations are dropped there (see the source='new-message'
// branch) so this doesn't flood the host on channels the bot isn't
// wired to.
// Chat SDK dispatch (handling-events.mdx §"Handler dispatch order") is
// exclusive: subscribed → onSubscribedMessage; unsubscribed+mention →
// onNewMention; unsubscribed+pattern-match → onNewMessage. Registering
// with `/./` lets the router see every plain message on wired channels
// (needed for engage_mode='pattern' + ignored_message_policy='accumulate'
// wirings). `shouldEngage` drops unknown channels on this source
// specifically so we don't flood from channels the bot merely joined.
chat.onNewMessage(/./, async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id);
const text = typeof message.text === 'string' ? message.text : '';
const decision = engageDecision(channelId, 'new-message', text);
const decision = engageDecision(channelId, 'new-message');
if (!decision.forward) return;
// SDK dispatch guarantees this is a non-mention non-DM message in an
// unsubscribed thread — isMention is definitively false here.
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, false));
});