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

feat(routing): engage modes + sender scope + accumulate/drop + per-agent fan-out

Browse Source 
Replaces the opaque trigger_rules JSON + response_scope enum on
messaging_group_agents with four explicit orthogonal columns:

    engage_mode            'pattern' | 'mention' | 'mention-sticky'
    engage_pattern         regex source; required when mode='pattern';
                           '.' is the "always" sentinel
    sender_scope           'all' | 'known'
    ignored_message_policy 'drop' | 'accumulate'

Inbound routing becomes a fan-out — every wired agent is evaluated
independently. A match gets its own session + container wake. A miss
with accumulate keeps the message as context-only (trigger=0) in that
agent's session, so when the agent does eventually engage it sees the
prior chatter.

## Schema

- Migration 010 (`engage-modes`): adds the 4 new columns, backfills
  from trigger_rules.pattern + requiresTrigger + response_scope, drops
  the legacy columns.
- messages_in gains `trigger INTEGER NOT NULL DEFAULT 1` (session DB
  schema + `migrateMessagesInTable` forward-compat).
- countDueMessages gates waking on `trigger = 1`.

## Routing

- `pickAgent` (returns one) → loop over all wired agents. Per agent:
  evaluate engage_mode; run access gate + sender-scope gate; on full
  match → resolveSession + writeSessionMessage(trigger=1) + wake. On
  miss with accumulate → writeSessionMessage(trigger=0), no wake. On
  miss with drop → skip.
- New `findSessionForAgent(agentGroupId, mgId, threadId)` scopes
  session lookup by agent so fan-out doesn't cross sessions.
- `messageIdForAgent` namespaces inbound message ids by agent_group_id
  so PRIMARY KEY doesn't collide across per-agent session DBs.

## Adapter layer

- `ConversationConfig` replaces `triggerPattern` + `requiresTrigger`
  with `engageMode` + `engagePattern`.
- Chat SDK bridge stores `Map<platformId, ConversationConfig[]>` (multi-
  agent per conversation) and applies union gating pre-onInbound:
    * onSubscribedMessage: engage if any wiring keeps firing in
      subscribed state (mention-sticky or pattern)
    * onNewMention: engage on mention; only subscribes the thread if
      at least one wiring is `mention-sticky`
    * onDirectMessage: engage per mode; sticky follows same rule
- Bridge no longer unconditionally calls `thread.subscribe()`.

## Sender scope

- Permissions module registers a second hook `setSenderScopeGate` that
  runs per-wiring after the existing access gate. `sender_scope='known'`
  requires canAccessAgentGroup(); `'all'` is a no-op. Not installed →
  no-op everywhere (default allow).

## Container side

- Host passes `NANOCLAW_MAX_MESSAGES_PER_PROMPT` (reuses existing
  MAX_MESSAGES_PER_PROMPT config; was dead code from v1).
- `getPendingMessages` queries `ORDER BY seq DESC LIMIT N`, reverses to
  chronological order for the prompt — accumulated context rides along
  with trigger rows up to the cap.
- `MessageInRow` gains `trigger: number` so the container can tell them
  apart in downstream code (container still processes both; only the
  host uses `trigger=0` for don't-wake).

## Defaults (per ACTION-ITEMS item 1 decision)

- DM (is_group=0): `engage_mode='pattern'`, `engage_pattern='.'` (always)
- Threaded group: `engage_mode='mention-sticky'` (seed-discord)
- Non-threaded group / CLI: pattern '.' in bootstrap scripts

## Tests

- src/host-core.test.ts: 3 new cases — fan-out (2 agents, 2 sessions,
  2 wakes), accumulate (trigger=0 + no wake), drop (no session created).
- Existing 10 host-core tests still pass.
- Migration 010 runs on an empty DB in 0-row path — verified.

Closes: ACTION-ITEMS items 1, 4.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gavrielc
2026-04-20 01:30:04 +03:00
parent 6a815190c0
commit 16b9499532
22 changed files with 688 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
src/router.ts
View File

@@ -18,14 +18,16 @@
* for policy refusals.
*/
import { getChannelAdapter } from './channels/channel-registry.js';
import { getAgentGroup } from './db/agent-groups.js';
import { recordDroppedMessage } from './db/dropped-messages.js';
import { getMessagingGroupByPlatform, createMessagingGroup, getMessagingGroupAgents } from './db/messaging-groups.js';
import { findSessionForAgent } from './db/sessions.js';
import { startTypingRefresh } from './modules/typing/index.js';
import { log } from './log.js';
import { resolveSession, writeSessionMessage } from './session-manager.js';
import { wakeContainer } from './container-runner.js';
import { getSession } from './db/sessions.js';
import type { MessagingGroup, MessagingGroupAgent } from './types.js';
import type { AgentGroup, MessagingGroup, MessagingGroupAgent } from './types.js';
function generateId(): string {
return `msg-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
@@ -89,6 +91,29 @@ export function setAccessGate(fn: AccessGateFn): void {
accessGate = fn;
}
/**
* Per-wiring sender-scope hook. Runs alongside the access gate for each
* agent that would otherwise engage — lets the permissions module enforce
* `sender_scope='known'` on wirings that are stricter than the messaging
* group's `unknown_sender_policy`. When the hook isn't registered (module
* not installed), sender_scope is a no-op.
*/
export type SenderScopeGateFn = (
event: InboundEvent,
userId: string | null,
mg: MessagingGroup,
agent: MessagingGroupAgent,
) => AccessGateResult;
let senderScopeGate: SenderScopeGateFn | null = null;
export function setSenderScopeGate(fn: SenderScopeGateFn): void {
if (senderScopeGate) {
log.warn('Sender-scope gate overwritten');
}
senderScopeGate = fn;
}
function safeParseContent(raw: string): { text?: string; sender?: string; senderId?: string } {
try {
return JSON.parse(raw);
@@ -158,91 +183,167 @@ export async function routeInbound(event: InboundEvent): Promise<void> {
return;
}
const match = pickAgent(agents, event);
if (!match) {
log.warn('MESSAGE DROPPED — no agent matched trigger rules', {
messagingGroupId: mg.id,
channelType: event.channelType,
});
const parsed = safeParseContent(event.message.content);
// 4. Fan-out: evaluate each wired agent independently against engage_mode,
// sender_scope, and access gate. An agent that engages gets its own
// session and container wake. An agent that declines but has
// ignored_message_policy='accumulate' still gets the message stored in
// its session (trigger=0) so the context is available when it does
// engage later. Drop policy = skip silently.
const parsed = safeParseContent(event.message.content);
const messageText = parsed.text ?? '';
let engagedCount = 0;
let accumulatedCount = 0;
for (const agent of agents) {
const agentGroup = getAgentGroup(agent.agent_group_id);
if (!agentGroup) continue;
const engages = evaluateEngage(agent, agentGroup, messageText, mg, event.threadId);
const accessOk = engages && (!accessGate || accessGate(event, userId, mg, agent.agent_group_id).allowed);
const scopeOk = engages && (!senderScopeGate || senderScopeGate(event, userId, mg, agent).allowed);
if (engages && accessOk && scopeOk) {
await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, true);
engagedCount++;
} else if (agent.ignored_message_policy === 'accumulate') {
await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, false);
accumulatedCount++;
} else {
log.debug('Message not engaged for agent (drop policy)', {
agentGroupId: agent.agent_group_id,
engage_mode: agent.engage_mode,
engages,
accessOk,
scopeOk,
});
}
}
if (engagedCount + accumulatedCount === 0) {
recordDroppedMessage({
channel_type: event.channelType,
platform_id: event.platformId,
user_id: userId,
sender_name: parsed.sender ?? null,
reason: 'no_trigger_match',
reason: 'no_agent_engaged',
messaging_group_id: mg.id,
agent_group_id: null,
});
return;
}
}
// 4. Access gate (if the permissions module is loaded). Otherwise
// allow-all.
if (accessGate) {
const result = accessGate(event, userId, mg, match.agent_group_id);
if (!result.allowed) {
log.info('MESSAGE DROPPED — access gate refused', {
messagingGroupId: mg.id,
agentGroupId: match.agent_group_id,
userId,
reason: result.reason,
});
return;
/**
* Decide whether a given wired agent should engage on this message.
*
* 'pattern' — regex test on text; '.' = always
* 'mention' — bot must be @-mentioned by its agent-group name
* 'mention-sticky' — @mention OR an active per-thread session already
* exists for this (agent, mg, thread). The session
* existence IS our subscription state; once a thread
* has engaged us once, follow-ups arrive with no
* mention and should still fire.
*/
function evaluateEngage(
agent: MessagingGroupAgent,
agentGroup: AgentGroup,
text: string,
mg: MessagingGroup,
threadId: string | null,
): boolean {
switch (agent.engage_mode) {
case 'pattern': {
const pat = agent.engage_pattern ?? '.';
if (pat === '.') return true;
try {
return new RegExp(pat).test(text);
} catch {
// Bad regex: fail open so admin sees the agent responding + can fix.
return true;
}
}
case 'mention':
return hasMention(text, agentGroup.name);
case 'mention-sticky': {
if (hasMention(text, agentGroup.name)) return true;
// Sticky follow-up: session already exists for this (agent, mg, thread)
// — the thread was activated before, keep firing.
if (mg.is_group === 0) return false; // DMs never use mention-sticky sensibly
const existing = findSessionForAgent(agent.agent_group_id, mg.id, threadId);
return existing !== undefined;
}
default:
return false;
}
}
// 5. Resolve or create session.
//
// Adapter thread policy overrides the wiring's session_mode: if the adapter
// is threaded, each thread gets its own session regardless of what the
// wiring says. Agent-shared is preserved because it expresses a
// cross-channel intent the adapter can't know about.
//
// Exception: DMs (is_group=0). Sub-threads within a DM are a UX affordance,
// not a conversation boundary — treat the whole DM as one session and let
// threadId flow through to delivery so replies land in the right sub-thread.
let effectiveSessionMode = match.session_mode;
if (adapter && adapter.supportsThreads && effectiveSessionMode !== 'agent-shared' && mg.is_group !== 0) {
function hasMention(text: string, agentName: string): boolean {
if (!agentName) return false;
const escaped = agentName.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
return new RegExp(`@${escaped}\\b`, 'i').test(text);
}
async function deliverToAgent(
agent: MessagingGroupAgent,
agentGroup: AgentGroup,
mg: MessagingGroup,
event: InboundEvent,
userId: string | null,
adapterSupportsThreads: boolean,
wake: boolean,
): Promise<void> {
// Apply the adapter thread policy: threaded adapter in a group chat →
// per-thread session regardless of wiring. agent-shared preserved (it's
// a cross-channel directive the adapter doesn't know about). DMs collapse
// sub-threads to one session (is_group=0 short-circuit).
let effectiveSessionMode = agent.session_mode;
if (adapterSupportsThreads && effectiveSessionMode !== 'agent-shared' && mg.is_group !== 0) {
effectiveSessionMode = 'per-thread';
}
const { session, created } = resolveSession(match.agent_group_id, mg.id, event.threadId, effectiveSessionMode);
// 6. Write message to session DB
const { session, created } = resolveSession(agent.agent_group_id, mg.id, event.threadId, effectiveSessionMode);
writeSessionMessage(session.agent_group_id, session.id, {
id: event.message.id || generateId(),
id: messageIdForAgent(event.message.id, agent.agent_group_id),
kind: event.message.kind,
timestamp: event.message.timestamp,
platformId: event.platformId,
channelType: event.channelType,
threadId: event.threadId,
content: event.message.content,
trigger: wake ? 1 : 0,
});
log.info('Message routed', {
sessionId: session.id,
agentGroup: match.agent_group_id,
agentGroup: agent.agent_group_id,
engage_mode: agent.engage_mode,
kind: event.message.kind,
userId,
wake,
created,
agentGroupName: agentGroup.name,
});
// 7. Show typing indicator while the agent processes.
startTypingRefresh(session.id, session.agent_group_id, event.channelType, event.platformId, event.threadId);
// 8. Wake container
const freshSession = getSession(session.id);
if (freshSession) {
await wakeContainer(freshSession);
if (wake) {
// Typing indicator + wake are only for the engaged branch; accumulated
// messages sit silently until a real trigger fires.
startTypingRefresh(session.id, session.agent_group_id, event.channelType, event.platformId, event.threadId);
const freshSession = getSession(session.id);
if (freshSession) {
await wakeContainer(freshSession);
}
}
}
/**
* Pick the matching agent for an inbound event.
* Currently: highest priority agent. Future: trigger rule matching.
* When fanning out, the same inbound message lands in multiple per-agent
* session DBs. messages_in.id is PRIMARY KEY, so reuse of the raw id would
* collide across sessions (or, more subtly, within one session if re-routed
* after a retry). Namespace by agent_group_id to keep ids unique per session.
*/
function pickAgent(agents: MessagingGroupAgent[], _event: InboundEvent): MessagingGroupAgent | null {
// Agents are already ordered by priority DESC from the DB query
// TODO: apply trigger_rules matching (pattern, mentionOnly, etc.)
return agents[0] ?? null;
function messageIdForAgent(baseId: string | undefined, agentGroupId: string): string {
const id = baseId && baseId.length > 0 ? baseId : generateId();
return `${id}:${agentGroupId}`;
}