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

refactor(channels,router): move all policy to router; bridge is transport

Browse Source 
Follow-up to b159722. That shrank the bridge's shouldEngage to a flood
gate + coarse sticky-subscribe signal. This completes the move —
policy lives exclusively in the router, the bridge is transport-only,
and the conversations map + ChannelSetup.conversations +
ChannelAdapter.updateConversations are all gone.

Key shifts:

1. Subscribe moves from bridge to router.

   Bridge used to call `thread.subscribe()` from its onNewMention /
   onDirectMessage handlers based on a coarse "any mention-sticky wiring
   exists on this channel" check. That forced the decision before the
   router could apply per-wiring engage logic, and it relied on the
   conversations map being current (staleness risk).

   ChannelAdapter gains `subscribe?(platformId, threadId)`. The Chat
   SDK bridge implements it via SqliteStateAdapter.subscribe(threadId)
   (idempotent — a repeat call on an already-subscribed thread is a
   no-op). The router's fan-out loop calls it once per message when
   the first mention-sticky wiring actually engages. Precise, not
   coarse.

2. Short-circuit the drop path with one combined query.

   New `getMessagingGroupWithAgentCount(channelType, platformId)` does
   the messaging_groups lookup AND counts wirings in a single SELECT,
   using the existing UNIQUE(channel_type, platform_id) index on
   messaging_groups and UNIQUE(messaging_group_id, agent_group_id) on
   messaging_group_agents for the JOIN. No new indexes needed.

   routeInbound now short-circuits:
     - No messaging_groups row AND not addressed (no mention/DM)
       → return silently. One DB read, nothing written. This is the
       Discord-bot-in-a-big-guild case; we no longer auto-create rows
       for every plain message in every channel the bot can see.
     - Messaging group exists but no wirings AND not addressed
       → return silently. One DB read.
     - Otherwise fall through to sender resolution + fan-out as before.

   Behavioral change: plain chatter on unwired channels no longer gets
   dropped_messages audit rows, which used to bloat the table. Audit
   still fires on addressed-to-bot drops where the admin cares
   ("someone @-mentioned us but nobody's wired").

3. Bridge is now purely transport.

   Deleted entirely: ConversationConfig, ChannelSetup.conversations,
   ChannelAdapter.updateConversations?, bridge's `conversations` map,
   buildConversationMap, shouldEngage, EngageSource, engageDecision,
   bridge.updateConversations method, src/index.ts
   buildConversationConfigs. Four handlers reduce to "resolve channel
   id, build InboundMessage with isMention, call onInbound". Net
   ~130 LOC deleted from the bridge.

   Collateral: the conversations-map staleness problem is gone. The
   upcoming channel-registration feature doesn't need any map-refresh
   plumbing — when an approval creates a new wiring, the next message
   hits the DB fresh and just works.

Bridge tests prune to the narrow platform-adjacent surface (openDM
delegation, subscribe presence). Host-core test that asserted the
old "auto-create on every unknown message" behavior updates to
reflect the new escalation-gated semantics: plain messages on
unknown channels don't auto-create, mentions do.

159 tests pass (was 172 — net -13, almost entirely from
bridge-engage-mode tests that covered logic now owned by the router
and exercised through host-core.test.ts).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gavrielc
2026-04-20 13:55:49 +03:00
parent b15972284b
commit a4061a0012
8 changed files with 173 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src/channels/adapter.ts
View File

@@ -5,45 +5,8 @@
* Two patterns: native adapters (implement directly) or Chat SDK bridge (wrap a Chat SDK adapter). * Two patterns: native adapters (implement directly) or Chat SDK bridge (wrap a Chat SDK adapter).
*/ */
/** Configuration for a registered conversation (messaging group + agent wiring). */
export interface ConversationConfig {
platformId: string;
agentGroupId: string;
/**
* When does the agent engage on messages from this conversation?
*
* 'pattern' — regex test against message text; engagePattern='.'
* means "always" (match everything)
* 'mention' — fires only on @mention
* 'mention-sticky' — fires on @mention, then auto-subscribes to the thread
* and treats subsequent messages as engage-all.
* Threaded platforms only (Slack/Discord/Linear).
*/
engageMode: 'pattern' | 'mention' | 'mention-sticky';
/** Regex source when engageMode='pattern'. '.' is the "always" sentinel. */
engagePattern?: string | null;
/**
* What to do with messages this wiring doesn't engage on.
*
* 'drop' — discard silently
* 'accumulate' — still forward to the host so the router can store the
* message in this agent's session with trigger=0. It
* rides along as context when the agent next wakes, but
* doesn't wake it on its own.
*
* The bridge reads this to decide whether to forward a non-engaging
* message at all — if any wiring on a conversation has 'accumulate', the
* bridge forwards and lets the router apply the per-wiring decision.
*/
ignoredMessagePolicy?: 'drop' | 'accumulate';
sessionMode: 'shared' | 'per-thread' | 'agent-shared';
}
/** Passed to the adapter at setup time. */ /** Passed to the adapter at setup time. */
export interface ChannelSetup { export interface ChannelSetup {
/** Known conversations from central DB. */
conversations: ConversationConfig[];
/** Called when an inbound message arrives from the platform. */ /** Called when an inbound message arrives from the platform. */
onInbound(platformId: string, threadId: string | null, message: InboundMessage): void | Promise<void>; onInbound(platformId: string, threadId: string | null, message: InboundMessage): void | Promise<void>;
@@ -125,7 +88,17 @@ export interface ChannelAdapter {
// Optional // Optional
setTyping?(platformId: string, threadId: string | null): Promise<void>; setTyping?(platformId: string, threadId: string | null): Promise<void>;
syncConversations?(): Promise<ConversationInfo[]>; syncConversations?(): Promise<ConversationInfo[]>;
updateConversations?(conversations: ConversationConfig[]): void;
/**
* Subscribe the bot to a thread so follow-up messages route via the
* platform's "subscribed message" path (onSubscribedMessage in Chat SDK).
* Called by the router when a mention-sticky wiring first engages in a
* thread. Idempotent: calling twice on the same thread is a no-op.
*
* Platforms without a subscription concept can omit this; the router
* treats absence as a no-op.
*/
subscribe?(platformId: string, threadId: string): Promise<void>;
/** /**
* Open (or fetch) a DM with this user, returning the platform_id of the * Open (or fetch) a DM with this user, returning the platform_id of the

2
src/channels/channel-registry.test.ts
View File

@@ -64,8 +64,6 @@ function createMockAdapter(
}, },
async setTyping() {}, async setTyping() {},
updateConversations() {},
}; };
} }

106
src/channels/chat-sdk-bridge.test.ts
View File

@@ -2,37 +2,19 @@ import { describe, expect, it } from 'vitest';
import type { Adapter } from 'chat'; import type { Adapter } from 'chat';
import type { ConversationConfig } from './adapter.js'; import { createChatSdkBridge } from './chat-sdk-bridge.js';
import { createChatSdkBridge, shouldEngage, type EngageSource } from './chat-sdk-bridge.js';
function stubAdapter(partial: Partial<Adapter>): Adapter { function stubAdapter(partial: Partial<Adapter>): Adapter {
return { name: 'stub', ...partial } as unknown as Adapter; return { name: 'stub', ...partial } as unknown as Adapter;
} }
function cfg(
partial: Partial<ConversationConfig> & { engageMode: ConversationConfig['engageMode'] },
): ConversationConfig {
return {
platformId: partial.platformId ?? 'C1',
agentGroupId: partial.agentGroupId ?? 'ag-1',
engageMode: partial.engageMode,
engagePattern: partial.engagePattern ?? null,
ignoredMessagePolicy: partial.ignoredMessagePolicy ?? 'drop',
sessionMode: partial.sessionMode ?? 'shared',
};
}
function mapFor(...configs: ConversationConfig[]): Map<string, ConversationConfig[]> {
const map = new Map<string, ConversationConfig[]>();
for (const c of configs) {
const existing = map.get(c.platformId);
if (existing) existing.push(c);
else map.set(c.platformId, [c]);
}
return map;
}
describe('createChatSdkBridge', () => { describe('createChatSdkBridge', () => {
// The bridge is now transport-only: forward inbound events, relay outbound
// ops. All per-wiring engage / accumulate / drop / subscribe decisions live
// in the router (src/router.ts routeInbound / evaluateEngage) and are
// exercised by host-core.test.ts end-to-end. These tests only cover the
// bridge's narrow, platform-adjacent surface.
it('omits openDM when the underlying Chat SDK adapter has none', () => { it('omits openDM when the underlying Chat SDK adapter has none', () => {
const bridge = createChatSdkBridge({ const bridge = createChatSdkBridge({
adapter: stubAdapter({}), adapter: stubAdapter({}),
@@ -59,76 +41,12 @@ describe('createChatSdkBridge', () => {
expect(openDMCalls).toEqual(['user-42']); expect(openDMCalls).toEqual(['user-42']);
expect(platformId).toBe('stub:user-42'); expect(platformId).toBe('stub:user-42');
}); });
});
describe('shouldEngage (bridge-level flood gate + subscribe signal)', () => { it('exposes subscribe (lets the router initiate thread subscription on mention-sticky engage)', () => {
// Per-wiring engage_mode / engage_pattern / ignored_message_policy const bridge = createChatSdkBridge({
// semantics live in the router (evaluateEngage / routeInbound fan-out). adapter: stubAdapter({}),
// These tests only cover the bridge's two responsibilities: should we supportsThreads: true,
// forward at all, and should we call thread.subscribe().
describe('flood gate — unknown conversation', () => {
const empty = new Map<string, ConversationConfig[]>();
const carriedSources: EngageSource[] = ['subscribed', 'mention', 'dm'];
for (const source of carriedSources) {
it(`forwards for source='${source}' (may be a newly-auto-created channel or a channel-registration trigger)`, () => {
expect(shouldEngage(empty, 'C-new', source)).toEqual({ forward: true, stickySubscribe: false });
});
}
it("DROPS for source='new-message' (onNewMessage(/./) fires for every unsubscribed thread the bot can see — would flood)", () => {
expect(shouldEngage(empty, 'C-unwired', 'new-message')).toEqual({
forward: false,
stickySubscribe: false,
});
});
});
describe('known conversation — bridge forwards regardless of engage mode', () => {
// Policy lives in the router now. The bridge only knows "has any wiring".
const conv = mapFor(cfg({ engageMode: 'mention' }));
for (const source of ['subscribed', 'mention', 'dm', 'new-message'] as EngageSource[]) {
it(`forwards for source='${source}' — router will decide engage / accumulate / drop per wiring`, () => {
expect(shouldEngage(conv, 'C1', source).forward).toBe(true);
});
}
});
describe('stickySubscribe signal', () => {
it('true when any mention-sticky wiring exists AND source is mention', () => {
const conv = mapFor(cfg({ engageMode: 'mention-sticky' }));
expect(shouldEngage(conv, 'C1', 'mention').stickySubscribe).toBe(true);
});
it('true when any mention-sticky wiring exists AND source is dm', () => {
const conv = mapFor(cfg({ engageMode: 'mention-sticky' }));
expect(shouldEngage(conv, 'C1', 'dm').stickySubscribe).toBe(true);
});
it('false on subscribed — thread is already subscribed, no need to re-subscribe', () => {
const conv = mapFor(cfg({ engageMode: 'mention-sticky' }));
expect(shouldEngage(conv, 'C1', 'subscribed').stickySubscribe).toBe(false);
});
it('false on new-message — mention-sticky requires an explicit mention to start', () => {
const conv = mapFor(cfg({ engageMode: 'mention-sticky' }));
expect(shouldEngage(conv, 'C1', 'new-message').stickySubscribe).toBe(false);
});
it('false for plain mention / pattern wirings (not sticky)', () => {
const mentionConv = mapFor(cfg({ engageMode: 'mention' }));
const patternConv = mapFor(cfg({ engageMode: 'pattern', engagePattern: '.' }));
for (const s of ['subscribed', 'mention', 'dm', 'new-message'] as EngageSource[]) {
expect(shouldEngage(mentionConv, 'C1', s).stickySubscribe).toBe(false);
expect(shouldEngage(patternConv, 'C1', s).stickySubscribe).toBe(false);
}
});
it('fires on coarse union — mixed wirings where any one is mention-sticky', () => {
const conv = mapFor(
cfg({ agentGroupId: 'ag-a', engageMode: 'mention' }),
cfg({ agentGroupId: 'ag-b', engageMode: 'mention-sticky' }),
);
expect(shouldEngage(conv, 'C1', 'mention').stickySubscribe).toBe(true);
}); });
expect(typeof bridge.subscribe).toBe('function');
}); });
}); });

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

@@ -21,7 +21,7 @@ import { SqliteStateAdapter } from '../state-sqlite.js';
import { registerWebhookAdapter } from '../webhook-server.js'; import { registerWebhookAdapter } from '../webhook-server.js';
import { getAskQuestionRender } from '../db/sessions.js'; import { getAskQuestionRender } from '../db/sessions.js';
import { normalizeOptions, type NormalizedOption } from './ask-question.js'; import { normalizeOptions, type NormalizedOption } from './ask-question.js';
import type { ChannelAdapter, ChannelSetup, ConversationConfig, InboundMessage } from './adapter.js'; import type { ChannelAdapter, ChannelSetup, InboundMessage } from './adapter.js';
/** Adapter with optional gateway support (e.g., Discord). */ /** Adapter with optional gateway support (e.g., Discord). */
interface GatewayAdapter extends Adapter { interface GatewayAdapter extends Adapter {
@@ -65,99 +65,14 @@ export interface ChatSdkBridgeConfig {
transformOutboundText?: (text: string) => string; transformOutboundText?: (text: string) => string;
} }
/**
* Which Chat SDK handler delivered this message. Determines which engage modes
* can fire.
*
* - `subscribed` — `onSubscribedMessage`. Thread is already subscribed.
* Every wiring mode (mention / mention-sticky / pattern)
* evaluates normally.
* - `mention` — `onNewMention`. Bot was @-mentioned in an unsubscribed
* thread. mention + mention-sticky engage; pattern runs
* the regex.
* - `dm` — `onDirectMessage`. Unsubscribed DM. Treated like a
* mention for engagement purposes.
* - `new-message` — `onNewMessage(/./, …)`. Plain non-mention non-DM
* message in an unsubscribed thread. Only `pattern`
* wirings can fire here. mention / mention-sticky ignore
* this source (they require an explicit mention).
*/
export type EngageSource = 'subscribed' | 'mention' | 'dm' | 'new-message';
/**
* 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`.
*/
export function shouldEngage(
conversations: Map<string, ConversationConfig[]>,
channelId: string,
source: EngageSource,
): { forward: boolean; stickySubscribe: boolean } {
const configs = conversations.get(channelId);
if (!configs || configs.length === 0) {
return { forward: source !== 'new-message', stickySubscribe: false };
}
const stickySubscribe =
(source === 'mention' || source === 'dm') && configs.some((cfg) => cfg.engageMode === 'mention-sticky');
return { forward: true, stickySubscribe };
}
export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter { export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter {
const { adapter } = config; const { adapter } = config;
const transformText = (t: string): string => (config.transformOutboundText ? config.transformOutboundText(t) : t); const transformText = (t: string): string => (config.transformOutboundText ? config.transformOutboundText(t) : t);
let chat: Chat; let chat: Chat;
let state: SqliteStateAdapter; let state: SqliteStateAdapter;
let setupConfig: ChannelSetup; let setupConfig: ChannelSetup;
// Keyed by platformId. Multiple agents may be wired to the same
// conversation — this holds all their configs so the bridge can apply the
// most-permissive engage rule at gate time and only subscribe when at
// least one wiring requested 'mention-sticky'.
//
// STALENESS: populated at setup() and updateConversations(). If wirings
// change after setup, updateConversations() must be called to refresh
// (ACTION-ITEMS item 17).
let conversations: Map<string, ConversationConfig[]>;
let gatewayAbort: AbortController | null = null; let gatewayAbort: AbortController | null = null;
function buildConversationMap(configs: ConversationConfig[]): Map<string, ConversationConfig[]> {
const map = new Map<string, ConversationConfig[]>();
for (const conv of configs) {
const existing = map.get(conv.platformId);
if (existing) existing.push(conv);
else map.set(conv.platformId, [conv]);
}
return map;
}
function engageDecision(channelId: string, source: EngageSource): { forward: boolean; stickySubscribe: boolean } {
return shouldEngage(conversations, channelId, source);
}
async function messageToInbound(message: ChatMessage, isMention: boolean): Promise<InboundMessage> { async function messageToInbound(message: ChatMessage, isMention: boolean): Promise<InboundMessage> {
// eslint-disable-next-line @typescript-eslint/no-explicit-any // eslint-disable-next-line @typescript-eslint/no-explicit-any
const serialized = message.toJSON() as Record<string, any>; const serialized = message.toJSON() as Record<string, any>;
@@ -225,7 +140,6 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
async setup(hostConfig: ChannelSetup) { async setup(hostConfig: ChannelSetup) {
setupConfig = hostConfig; setupConfig = hostConfig;
conversations = buildConversationMap(hostConfig.conversations);
state = new SqliteStateAdapter(); state = new SqliteStateAdapter();
@@ -237,29 +151,25 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
logger: 'silent', logger: 'silent',
}); });
// Four SDK dispatch paths — bridge just forwards; router does all // Four SDK dispatch paths — bridge just forwards. All per-wiring
// per-wiring engage / accumulate / drop decisions. isMention is the // engage / accumulate / drop / subscribe decisions live in the host
// load-bearing signal (see evaluateEngage in src/router.ts). // router (src/router.ts routeInbound / evaluateEngage). The bridge
// only resolves channel ids and sets the platform-confirmed isMention
// flag that routeInbound evaluates; the router calls back into
// bridge.subscribe(...) when a mention-sticky wiring engages.
// Subscribed threads — every message in a thread we've previously // Subscribed threads — every message in a thread we've previously
// engaged. Carry the SDK's `message.isMention` through so mention-mode // engaged. Carry the SDK's `message.isMention` through so mention-mode
// wirings still fire on in-thread mentions. // wirings still fire on in-thread mentions.
chat.onSubscribedMessage(async (thread, message) => { chat.onSubscribedMessage(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id); const channelId = adapter.channelIdFromThreadId(thread.id);
const decision = engageDecision(channelId, 'subscribed');
if (!decision.forward) return;
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, message.isMention === true)); await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, message.isMention === true));
}); });
// @mention in an unsubscribed thread — SDK-confirmed bot mention. // @mention in an unsubscribed thread — SDK-confirmed bot mention.
chat.onNewMention(async (thread, message) => { chat.onNewMention(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id); const channelId = adapter.channelIdFromThreadId(thread.id);
const decision = engageDecision(channelId, 'mention');
if (!decision.forward) return;
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true)); await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true));
if (decision.stickySubscribe) {
await thread.subscribe();
}
}); });
// DMs — by definition addressed to the bot. Thread id flows through // DMs — by definition addressed to the bot. Thread id flows through
@@ -268,19 +178,13 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
// is_group=0 short-circuit. // is_group=0 short-circuit.
chat.onDirectMessage(async (thread, message) => { chat.onDirectMessage(async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id); const channelId = adapter.channelIdFromThreadId(thread.id);
const decision = engageDecision(channelId, 'dm');
log.info('Inbound DM received', { log.info('Inbound DM received', {
adapter: adapter.name, adapter: adapter.name,
channelId, channelId,
sender: (message.author as any)?.fullName ?? (message.author as any)?.userId ?? 'unknown', sender: (message.author as any)?.fullName ?? (message.author as any)?.userId ?? 'unknown',
threadId: thread.id, threadId: thread.id,
forward: decision.forward,
}); });
if (!decision.forward) return;
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true)); await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, true));
if (decision.stickySubscribe) {
await thread.subscribe();
}
}); });
// Plain messages in unsubscribed threads. // Plain messages in unsubscribed threads.
@@ -288,14 +192,13 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
// Chat SDK dispatch (handling-events.mdx §"Handler dispatch order") is // Chat SDK dispatch (handling-events.mdx §"Handler dispatch order") is
// exclusive: subscribed → onSubscribedMessage; unsubscribed+mention → // exclusive: subscribed → onSubscribedMessage; unsubscribed+mention →
// onNewMention; unsubscribed+pattern-match → onNewMessage. Registering // onNewMention; unsubscribed+pattern-match → onNewMessage. Registering
// with `/./` lets the router see every plain message on wired channels // with `/./` lets the router see every plain message on every
// (needed for engage_mode='pattern' + ignored_message_policy='accumulate' // unsubscribed thread the bot can see. The router short-circuits via
// wirings). `shouldEngage` drops unknown channels on this source // getMessagingGroupWithAgentCount (~1 DB read) for unwired channels,
// specifically so we don't flood from channels the bot merely joined. // so forwarding every one is cheap enough to not need a bridge-side
// flood gate.
chat.onNewMessage(/./, async (thread, message) => { chat.onNewMessage(/./, async (thread, message) => {
const channelId = adapter.channelIdFromThreadId(thread.id); const channelId = adapter.channelIdFromThreadId(thread.id);
const decision = engageDecision(channelId, 'new-message');
if (!decision.forward) return;
await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, false)); await setupConfig.onInbound(channelId, thread.id, await messageToInbound(message, false));
}); });
@@ -468,8 +371,13 @@ export function createChatSdkBridge(config: ChatSdkBridgeConfig): ChannelAdapter
return true; return true;
}, },
updateConversations(configs: ConversationConfig[]) { async subscribe(_platformId: string, threadId: string) {
conversations = buildConversationMap(configs); // Chat SDK's subscription state lives on the StateAdapter (not on the
// Chat instance itself). SqliteStateAdapter.subscribe is idempotent —
// a second call on an already-subscribed thread is a no-op. threadId
// is the SDK's thread id, which is what the router already has from
// the original inbound event.
await state.subscribe(threadId);
}, },
}; };

31
src/db/messaging-groups.ts
View File

@@ -37,6 +37,37 @@ export function getMessagingGroupByPlatform(channelType: string, platformId: str
.get(channelType, platformId) as MessagingGroup | undefined; .get(channelType, platformId) as MessagingGroup | undefined;
} }
/**
* Combined lookup for the router's fast-drop path. Returns the messaging
* group (if it exists) and a count of wired agents in one query — lets
* `routeInbound` short-circuit messages for unwired / unknown channels
* with a single DB read instead of four (mg lookup, sender upsert, agents
* lookup, dropped_messages insert).
*
* Returns `null` when no messaging_groups row exists for this channel.
* Returns `{ mg, agentCount: 0 }` when the row exists but has no wired
* agents. Uses the `UNIQUE(channel_type, platform_id)` index plus the
* `UNIQUE(messaging_group_id, agent_group_id)` index for the JOIN — both
* covered by existing SQLite auto-indexes from the UNIQUE constraints.
*/
export function getMessagingGroupWithAgentCount(
channelType: string,
platformId: string,
): { mg: MessagingGroup; agentCount: number } | null {
const row = getDb()
.prepare(
`SELECT mg.*, COUNT(mga.id) AS agent_count
FROM messaging_groups mg
LEFT JOIN messaging_group_agents mga ON mga.messaging_group_id = mg.id
WHERE mg.channel_type = ? AND mg.platform_id = ?
GROUP BY mg.id`,
)
.get(channelType, platformId) as (MessagingGroup & { agent_count: number }) | undefined;
if (!row) return null;
const { agent_count, ...mg } = row;
return { mg: mg as MessagingGroup, agentCount: agent_count };
}
export function getAllMessagingGroups(): MessagingGroup[] { export function getAllMessagingGroups(): MessagingGroup[] {
return getDb().prepare('SELECT * FROM messaging_groups ORDER BY name').all() as MessagingGroup[]; return getDb().prepare('SELECT * FROM messaging_groups ORDER BY name').all() as MessagingGroup[];
} }

36
src/host-core.test.ts
View File

@@ -244,26 +244,42 @@ describe('router', () => {
expect(wakeContainer).toHaveBeenCalled(); expect(wakeContainer).toHaveBeenCalled();
}); });
it('should auto-create messaging group for unknown platform', async () => { it('auto-creates messaging group only when the bot is addressed (mention/DM)', async () => {
// The router's no-mg branch is escalation-gated: plain chatter on an
// unknown channel stays silent (no DB writes) so a bot that sits in
// many unwired channels doesn't bloat messaging_groups. Only explicit
// mentions and DMs trigger auto-create.
const { routeInbound } = await import('./router.js'); const { routeInbound } = await import('./router.js');
const { getMessagingGroupByPlatform } = await import('./db/messaging-groups.js');
const event: InboundEvent = { // Plain message on unknown channel — should NOT auto-create.
await routeInbound({
channelType: 'slack', channelType: 'slack',
platformId: 'C-NEW-CHANNEL', platformId: 'C-PLAIN',
threadId: null, threadId: null,
message: { message: {
id: 'msg-2', id: 'msg-plain',
kind: 'chat', kind: 'chat',
content: JSON.stringify({ sender: 'User', text: 'Hi' }), content: JSON.stringify({ sender: 'User', text: 'Hi' }),
timestamp: now(), timestamp: now(),
}, },
}; });
expect(getMessagingGroupByPlatform('slack', 'C-PLAIN')).toBeUndefined();
await routeInbound(event); // Mention on unknown channel — SHOULD auto-create (next step: channel-registration flow).
await routeInbound({
const { getMessagingGroupByPlatform } = await import('./db/messaging-groups.js'); channelType: 'slack',
const mg = getMessagingGroupByPlatform('slack', 'C-NEW-CHANNEL'); platformId: 'C-MENTIONED',
expect(mg).toBeDefined(); threadId: null,
message: {
id: 'msg-mentioned',
kind: 'chat',
content: JSON.stringify({ sender: 'User', text: '@bot hi' }),
timestamp: now(),
isMention: true,
},
});
expect(getMessagingGroupByPlatform('slack', 'C-MENTIONED')).toBeDefined();
}); });
it('should route multiple messages to the same session', async () => { it('should route multiple messages to the same session', async () => {

27
src/index.ts
View File

@@ -9,7 +9,6 @@ import path from 'path';
import { DATA_DIR } from './config.js'; import { DATA_DIR } from './config.js';
import { initDb } from './db/connection.js'; import { initDb } from './db/connection.js';
import { runMigrations } from './db/migrations/index.js'; import { runMigrations } from './db/migrations/index.js';
import { getMessagingGroupsByChannel, getMessagingGroupAgents } from './db/messaging-groups.js';
import { ensureContainerRuntimeRunning, cleanupOrphans } from './container-runtime.js'; import { ensureContainerRuntimeRunning, cleanupOrphans } from './container-runtime.js';
import { startActiveDeliveryPoll, startSweepDeliveryPoll, setDeliveryAdapter, stopDeliveryPolls } from './delivery.js'; import { startActiveDeliveryPoll, startSweepDeliveryPoll, setDeliveryAdapter, stopDeliveryPolls } from './delivery.js';
import { startHostSweep, stopHostSweep } from './host-sweep.js'; import { startHostSweep, stopHostSweep } from './host-sweep.js';
@@ -52,7 +51,7 @@ import './channels/index.js';
// append registry-based modules. Imported for side effects (registrations). // append registry-based modules. Imported for side effects (registrations).
import './modules/index.js'; import './modules/index.js';
import type { ChannelAdapter, ChannelSetup, ConversationConfig } from './channels/adapter.js'; import type { ChannelAdapter, ChannelSetup } from './channels/adapter.js';
import { initChannelAdapters, teardownChannelAdapters, getChannelAdapter } from './channels/channel-registry.js'; import { initChannelAdapters, teardownChannelAdapters, getChannelAdapter } from './channels/channel-registry.js';
async function main(): Promise<void> { async function main(): Promise<void> {
@@ -70,9 +69,7 @@ async function main(): Promise<void> {
// 3. Channel adapters // 3. Channel adapters
await initChannelAdapters((adapter: ChannelAdapter): ChannelSetup => { await initChannelAdapters((adapter: ChannelAdapter): ChannelSetup => {
const conversations = buildConversationConfigs(adapter.channelType);
return { return {
conversations,
onInbound(platformId, threadId, message) { onInbound(platformId, threadId, message) {
routeInbound({ routeInbound({
channelType: adapter.channelType, channelType: adapter.channelType,
@@ -151,28 +148,6 @@ async function main(): Promise<void> {
log.info('NanoClaw running'); log.info('NanoClaw running');
} }
/** Build ConversationConfig[] for a channel type from the central DB. */
function buildConversationConfigs(channelType: string): ConversationConfig[] {
const groups = getMessagingGroupsByChannel(channelType);
const configs: ConversationConfig[] = [];
for (const mg of groups) {
const agents = getMessagingGroupAgents(mg.id);
for (const agent of agents) {
configs.push({
platformId: mg.platform_id,
agentGroupId: agent.agent_group_id,
engageMode: agent.engage_mode,
engagePattern: agent.engage_pattern,
ignoredMessagePolicy: agent.ignored_message_policy,
sessionMode: agent.session_mode,
});
}
}
return configs;
}
/** Graceful shutdown. */ /** Graceful shutdown. */
async function shutdown(signal: string): Promise<void> { async function shutdown(signal: string): Promise<void> {
log.info('Shutdown signal received', { signal }); log.info('Shutdown signal received', { signal });

85
src/router.ts
View File

@@ -20,7 +20,11 @@
import { getChannelAdapter } from './channels/channel-registry.js'; import { getChannelAdapter } from './channels/channel-registry.js';
import { getAgentGroup } from './db/agent-groups.js'; import { getAgentGroup } from './db/agent-groups.js';
import { recordDroppedMessage } from './db/dropped-messages.js'; import { recordDroppedMessage } from './db/dropped-messages.js';
import { getMessagingGroupByPlatform, createMessagingGroup, getMessagingGroupAgents } from './db/messaging-groups.js'; import {
createMessagingGroup,
getMessagingGroupAgents,
getMessagingGroupWithAgentCount,
} from './db/messaging-groups.js';
import { findSessionForAgent } from './db/sessions.js'; import { findSessionForAgent } from './db/sessions.js';
import { startTypingRefresh } from './modules/typing/index.js'; import { startTypingRefresh } from './modules/typing/index.js';
import { log } from './log.js'; import { log } from './log.js';
@@ -143,10 +147,21 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
event = { ...event, threadId: null }; event = { ...event, threadId: null };
} }
// 1. Resolve messaging group const isMention = event.message.isMention === true;
let mg = getMessagingGroupByPlatform(event.channelType, event.platformId);
// 1. Combined lookup: messaging_group row + count of wired agents in a
// single query. Cheap short-circuit for the common "unwired channel"
// case — one DB read and we're out, no auto-create, no sender
// resolution, no log spam.
const found = getMessagingGroupWithAgentCount(event.channelType, event.platformId);
let mg: MessagingGroup;
if (!found) {
// No messaging_groups row. Auto-create only when the message warrants
// attention (the bot was addressed — @mention or DM). Plain chatter in
// channels we merely sit in stays silent — no row, no DB writes.
if (!isMention) return;
if (!mg) {
const mgId = `mg-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; const mgId = `mg-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
mg = { mg = {
id: mgId, id: mgId,
@@ -154,9 +169,6 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
platform_id: event.platformId, platform_id: event.platformId,
name: null, name: null,
is_group: 0, is_group: 0,
// Let the schema default (currently 'request_approval') apply rather
// than hardcoding 'strict' — the schema is the source of truth for
// the default policy. See migration 011.
unknown_sender_policy: 'request_approval', unknown_sender_policy: 'request_approval',
created_at: new Date().toISOString(), created_at: new Date().toISOString(),
}; };
@@ -166,17 +178,13 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
channelType: event.channelType, channelType: event.channelType,
platformId: event.platformId, platformId: event.platformId,
}); });
} } else {
mg = found.mg;
// 2. Sender resolution (permissions module upserts the users row as a if (found.agentCount === 0) {
// side effect so later role/access lookups find a real record). // Messaging group exists but has no wirings. Stay silent for plain
// Without the module, userId is null — downstream tolerates it. // messages; only log + record on explicit mention/DM so admins can
const userId: string | null = senderResolver ? senderResolver(event) : null; // see that someone tried to reach the bot on an unwired channel.
if (!isMention) return;
// 3. Resolve agent groups wired to this messaging group. Structural
// drops record to dropped_messages for audit.
const agents = getMessagingGroupAgents(mg.id);
if (agents.length === 0) {
log.warn('MESSAGE DROPPED — no agent groups wired to this channel. Run setup register step to configure.', { log.warn('MESSAGE DROPPED — no agent groups wired to this channel. Run setup register step to configure.', {
messagingGroupId: mg.id, messagingGroupId: mg.id,
channelType: event.channelType, channelType: event.channelType,
@@ -186,7 +194,7 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
recordDroppedMessage({ recordDroppedMessage({
channel_type: event.channelType, channel_type: event.channelType,
platform_id: event.platformId, platform_id: event.platformId,
user_id: userId, user_id: null,
sender_name: parsed.sender ?? null, sender_name: parsed.sender ?? null,
reason: 'no_agent_wired', reason: 'no_agent_wired',
messaging_group_id: mg.id, messaging_group_id: mg.id,
@@ -194,6 +202,16 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
}); });
return; return;
} }
}
// 2. Sender resolution (permissions module upserts the users row as a
// side effect so later role/access lookups find a real record).
// Without the module, userId is null — downstream tolerates it.
const userId: string | null = senderResolver ? senderResolver(event) : null;
// 3. Fetch wired agents in full (we already know the count is > 0; now
// we need their actual rows for fan-out).
const agents = getMessagingGroupAgents(mg.id);
// 4. Fan-out: evaluate each wired agent independently against engage_mode, // 4. Fan-out: evaluate each wired agent independently against engage_mode,
// sender_scope, and access gate. An agent that engages gets its own // sender_scope, and access gate. An agent that engages gets its own
@@ -201,12 +219,18 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
// ignored_message_policy='accumulate' still gets the message stored in // ignored_message_policy='accumulate' still gets the message stored in
// its session (trigger=0) so the context is available when it does // its session (trigger=0) so the context is available when it does
// engage later. Drop policy = skip silently. // engage later. Drop policy = skip silently.
//
// Subscribe (for mention-sticky wirings on threaded platforms) fires
// once per message from this loop — the first engaging mention-sticky
// wiring triggers adapter.subscribe(...); subsequent wirings don't
// re-subscribe (chat.subscribe is idempotent anyway, but the flag
// avoids the extra await).
const parsed = safeParseContent(event.message.content); const parsed = safeParseContent(event.message.content);
const messageText = parsed.text ?? ''; const messageText = parsed.text ?? '';
const isMention = event.message.isMention === true;
let engagedCount = 0; let engagedCount = 0;
let accumulatedCount = 0; let accumulatedCount = 0;
let subscribed = false;
for (const agent of agents) { for (const agent of agents) {
const agentGroup = getAgentGroup(agent.agent_group_id); const agentGroup = getAgentGroup(agent.agent_group_id);
@@ -220,6 +244,27 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
if (engages && accessOk && scopeOk) { if (engages && accessOk && scopeOk) {
await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, true); await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, true);
engagedCount++; engagedCount++;
// Mention-sticky: ask the adapter to subscribe the thread so the
// platform's subscribed-message path carries follow-ups without
// requiring another @mention. Threaded-adapter only; DMs and
// non-threaded platforms skip.
if (
!subscribed &&
agent.engage_mode === 'mention-sticky' &&
adapter?.supportsThreads &&
adapter.subscribe &&
event.threadId !== null &&
mg.is_group !== 0
) {
subscribed = true;
// Fire-and-forget — subscribe is platform-side bookkeeping and
// shouldn't block message routing. Errors are logged inside the
// adapter (or by the promise rejection handler below).
void adapter.subscribe(event.platformId, event.threadId).catch((err) => {
log.warn('adapter.subscribe failed', { channelType: event.channelType, threadId: event.threadId, err });
});
}
} else if (agent.ignored_message_policy === 'accumulate') { } else if (agent.ignored_message_policy === 'accumulate') {
await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, false); await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, false);
accumulatedCount++; accumulatedCount++;