From 53ffa3d34753a07fd869d07d29f52e8bceaeface Mon Sep 17 00:00:00 2001 From: gavrielc Date: Fri, 10 Jul 2026 21:12:12 +0300 Subject: [PATCH] PR3: apply per-wiring thread policy and declared auto-create defaults in router - Auto-created messaging_groups take unknown_sender_policy from the channel declaration (DM vs group context); undeclared adapters resolve through the behavior-faithful fallback, identical to the old hardcoded value - Fanout computes each wiring's effectiveThreadId via resolveThreadPolicy (threads override, declaration, live capability) and threads it through sticky engage lookup, session-mode force, resolveSession, delivery address, typing, and the subscribe gate; event.replyTo is operator intent and never policy-stripped - Capability pre-strip for non-threaded adapters stays unchanged - Tests: NULL-threads inherit parity, threads=0 opt-out, replyTo exemption, declared vs fallback auto-create policy; metadata-preservation tests now register a live threaded adapter (the realistic config the router sees) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PJkDAnLGWJUqjgJNXpyqGZ --- src/host-core.test.ts | 223 +++++++++++++++++++++++++++++++++++++++++- src/router.ts | 79 ++++++++++----- 2 files changed, 277 insertions(+), 25 deletions(-) diff --git a/src/host-core.test.ts b/src/host-core.test.ts index 5bee875f4..a381d72e1 100644 --- a/src/host-core.test.ts +++ b/src/host-core.test.ts @@ -780,8 +780,198 @@ describe('router — channel instances', () => { }); }); -describe('routing metadata preservation', () => { +describe('router — per-wiring thread policy', () => { + // Slack-like threaded adapter on a unique channel type (the registry maps + // are module-global; unique names avoid cross-test collisions). + const makeThreadedAdapter = () => ({ + name: 'tp-slack', + channelType: 'tp-slack', + supportsThreads: true, + async setup() {}, + async teardown() {}, + isConnected: () => true, + async deliver() { + return undefined; + }, + }); + beforeEach(() => { + createAgentGroup({ + id: 'ag-tp', + name: 'Thread Agent', + folder: 'thread-agent', + agent_provider: null, + created_at: now(), + }); + createMessagingGroup({ + id: 'mg-tp', + channel_type: 'tp-slack', + platform_id: 'tp:C1', + name: 'Threaded chat', + is_group: 1, + unknown_sender_policy: 'public', + created_at: now(), + }); + createMessagingGroupAgent({ + id: 'mga-tp', + messaging_group_id: 'mg-tp', + agent_group_id: 'ag-tp', + engage_mode: 'pattern', + engage_pattern: '.', + sender_scope: 'all', + ignored_message_policy: 'drop', + session_mode: 'shared', + priority: 0, + created_at: now(), + }); + }); + + async function withThreadedAdapter(fn: () => Promise): Promise { + const { registerChannelAdapter, initChannelAdapters, teardownChannelAdapters } = + await import('./channels/channel-registry.js'); + registerChannelAdapter('tp-slack', { factory: makeThreadedAdapter }); + await initChannelAdapters(() => ({ + onInbound: () => {}, + onInboundEvent: () => {}, + onMetadata: () => {}, + onAction: () => {}, + })); + try { + await fn(); + } finally { + await teardownChannelAdapters(); + } + } + + const threadedEvent = (id: string): InboundEvent => ({ + channelType: 'tp-slack', + platformId: 'tp:C1', + threadId: 'thread-42', + message: { + id, + kind: 'chat', + content: JSON.stringify({ sender: 'U', text: 'hi' }), + timestamp: now(), + isGroup: true, + }, + }); + + it('NULL-threads wiring (inherit) on a threaded adapter keeps thread routing as before', async () => { + await withThreadedAdapter(async () => { + const { routeInbound } = await import('./router.js'); + const { getSessionsByAgentGroup } = await import('./db/sessions.js'); + + await routeInbound(threadedEvent('msg-null-threads')); + + // threads=NULL inherits the (fallback) declaration → supportsThreads → + // per-thread session with the platform thread id, message addressed + // in-thread. Identical to pre-declaration routing. + const sessions = getSessionsByAgentGroup('ag-tp'); + expect(sessions).toHaveLength(1); + expect(sessions[0].thread_id).toBe('thread-42'); + + const db = new Database(inboundDbPath('ag-tp', sessions[0].id)); + const row = db.prepare('SELECT thread_id FROM messages_in').get() as { thread_id: string | null }; + db.close(); + expect(row.thread_id).toBe('thread-42'); + }); + }); + + it('wiring threads=0 nulls the event-derived thread for session and delivery', async () => { + getDb().prepare("UPDATE messaging_group_agents SET threads = 0 WHERE id = 'mga-tp'").run(); + + await withThreadedAdapter(async () => { + const { routeInbound } = await import('./router.js'); + const { getSessionsByAgentGroup } = await import('./db/sessions.js'); + + await routeInbound(threadedEvent('msg-opt-out')); + + // Session collapses (no per-thread force, thread id stripped) and the + // reply address is top-level. + const sessions = getSessionsByAgentGroup('ag-tp'); + expect(sessions).toHaveLength(1); + expect(sessions[0].thread_id).toBeNull(); + + const db = new Database(inboundDbPath('ag-tp', sessions[0].id)); + const row = db.prepare('SELECT thread_id FROM messages_in').get() as { thread_id: string | null }; + db.close(); + expect(row.thread_id).toBeNull(); + }); + }); + + it('wiring threads=0 never strips replyTo (operator intent)', async () => { + getDb().prepare("UPDATE messaging_group_agents SET threads = 0 WHERE id = 'mga-tp'").run(); + + await withThreadedAdapter(async () => { + const { routeInbound } = await import('./router.js'); + const { getSessionsByAgentGroup } = await import('./db/sessions.js'); + + await routeInbound({ + ...threadedEvent('msg-replyto'), + replyTo: { channelType: 'cli', platformId: 'cli:operator', threadId: 'term-1' }, + }); + + const sessions = getSessionsByAgentGroup('ag-tp'); + expect(sessions).toHaveLength(1); + const db = new Database(inboundDbPath('ag-tp', sessions[0].id)); + const row = db.prepare('SELECT channel_type, thread_id FROM messages_in').get() as { + channel_type: string; + thread_id: string | null; + }; + db.close(); + // The reply address is the operator's, thread id intact — only the + // event-derived address is policy-stripped. + expect(row.channel_type).toBe('cli'); + expect(row.thread_id).toBe('term-1'); + }); + }); + + it('auto-create takes unknown_sender_policy from the declaration and falls back faithfully', async () => { + const { registerChannelAdapter } = await import('./channels/channel-registry.js'); + const { routeInbound } = await import('./router.js'); + const { getMessagingGroupByPlatform } = await import('./db/messaging-groups.js'); + + // Registration-tier declaration is enough — no live adapter needed. + registerChannelAdapter('tp-declared', { + factory: () => null, + defaults: { + dm: { engageMode: 'pattern', engagePattern: '.', threads: false, unknownSenderPolicy: 'strict' }, + group: { engageMode: 'mention', threads: false, unknownSenderPolicy: 'public' }, + mentions: 'platform', + }, + }); + + const mention = (channelType: string, platformId: string, isGroup: boolean): InboundEvent => ({ + channelType, + platformId, + threadId: null, + message: { + id: `msg-${platformId}`, + kind: 'chat', + content: JSON.stringify({ sender: 'U', text: '@bot hi' }), + timestamp: now(), + isMention: true, + isGroup, + }, + }); + + // Declared adapter: group context reads the group declaration... + await routeInbound(mention('tp-declared', 'tp:G1', true)); + expect(getMessagingGroupByPlatform('tp-declared', 'tp:G1')!.unknown_sender_policy).toBe('public'); + + // ...and DM context reads the dm declaration. + await routeInbound(mention('tp-declared', 'tp:D1', false)); + expect(getMessagingGroupByPlatform('tp-declared', 'tp:D1')!.unknown_sender_policy).toBe('strict'); + + // Undeclared channel: the behavior-faithful fallback reproduces the + // historical hardcoded 'request_approval'. + await routeInbound(mention('tp-undeclared', 'tp:U1', true)); + expect(getMessagingGroupByPlatform('tp-undeclared', 'tp:U1')!.unknown_sender_policy).toBe('request_approval'); + }); +}); + +describe('routing metadata preservation', () => { + beforeEach(async () => { createAgentGroup({ id: 'ag-1', name: 'Test Agent', @@ -810,6 +1000,34 @@ describe('routing metadata preservation', () => { priority: 0, created_at: now(), }); + // A live threaded adapter, matching real Discord routing — inbound + // platform events always have their receiving adapter live, and the + // per-wiring thread policy hard-ANDs the live capability. + const { registerChannelAdapter, initChannelAdapters } = await import('./channels/channel-registry.js'); + registerChannelAdapter('discord', { + factory: () => ({ + name: 'discord', + channelType: 'discord', + supportsThreads: true, + async setup() {}, + async teardown() {}, + isConnected: () => true, + async deliver() { + return undefined; + }, + }), + }); + await initChannelAdapters(() => ({ + onInbound: () => {}, + onInboundEvent: () => {}, + onMetadata: () => {}, + onAction: () => {}, + })); + }); + + afterEach(async () => { + const { teardownChannelAdapters } = await import('./channels/channel-registry.js'); + await teardownChannelAdapters(); }); it('routed message carries platformId, channelType, threadId on the messages_in row', async () => { @@ -822,7 +1040,8 @@ describe('routing metadata preservation', () => { message: { id: 'msg-r1', kind: 'chat', content: JSON.stringify({ sender: 'A', text: 'hi' }), timestamp: now() }, }); - const session = findSession('mg-1', null); + // Threaded adapter in a group chat forces a per-thread session. + const session = findSession('mg-1', 'thread-42'); const db = new Database(inboundDbPath('ag-1', session!.id)); const row = db .prepare('SELECT platform_id, channel_type, thread_id FROM messages_in WHERE id LIKE ?') diff --git a/src/router.ts b/src/router.ts index d39c6a064..7a23ffcf6 100644 --- a/src/router.ts +++ b/src/router.ts @@ -17,7 +17,8 @@ * drops (no agent wired, no trigger match); the access gate writes rows * for policy refusals. */ -import { getChannelAdapter } from './channels/channel-registry.js'; +import { getChannelAdapter, getChannelDefaults } from './channels/channel-registry.js'; +import { resolveThreadPolicy, resolveUnknownSenderPolicy } from './channels/channel-defaults.js'; import { gateCommand } from './command-gate.js'; import { getAgentGroup } from './db/agent-groups.js'; import { recordDroppedMessage } from './db/dropped-messages.js'; @@ -208,7 +209,15 @@ export async function routeInbound(event: InboundEvent): Promise { instance: event.instance ?? event.channelType, name: null, is_group: event.message.isGroup ? 1 : 0, - unknown_sender_policy: 'request_approval', + // Policy from the receiving channel's declared defaults (DM vs group + // context); undeclared adapters resolve through the behavior-faithful + // fallback, which is 'request_approval' in both contexts — identical + // to the historical hardcode. + unknown_sender_policy: resolveUnknownSenderPolicy( + event.instance ?? event.channelType, + event.message.isGroup === true, + event.channelType, + ), denied_at: null, created_at: new Date().toISOString(), }; @@ -289,6 +298,14 @@ export async function routeInbound(event: InboundEvent): Promise { const parsed = safeParseContent(event.message.content); const messageText = parsed.text ?? ''; + // Per-wiring thread policy inputs, resolved once per event. Each wiring's + // threads override (NULL = inherit) resolves against the channel's declared + // defaults, hard-bounded by the live adapter's raw capability. Undeclared + // adapters resolve through the behavior-faithful fallback, so a NULL-threads + // wiring reproduces the historical supportsThreads-derived routing exactly. + const channelDefaults = getChannelDefaults(mg.instance ?? mg.channel_type, mg.channel_type); + const supportsThreads = adapter?.supportsThreads === true; + let engagedCount = 0; let accumulatedCount = 0; let subscribed = false; @@ -297,33 +314,44 @@ export async function routeInbound(event: InboundEvent): Promise { const agentGroup = getAgentGroup(agent.agent_group_id); if (!agentGroup) continue; - const engages = evaluateEngage(agent, messageText, isMention, mg, event.threadId); + // Effective thread id for THIS wiring: the event-derived address is + // policy-stripped when the wiring (or its channel declaration) opts out + // of threads. event.replyTo is operator intent from the CLI admin + // transport and is never nulled. Guard: platform thread ids must never + // collide with the reserved 'system:%' session namespace + // (src/db/sessions.ts) — they are platform-native identifiers, and this + // is the only place an inbound thread id enters session resolution. + const threadsEnabled = resolveThreadPolicy(agent.threads ?? null, channelDefaults, mg.is_group === 1, supportsThreads); + const effectiveThreadId = threadsEnabled ? event.threadId : null; + + const engages = evaluateEngage(agent, messageText, isMention, mg, effectiveThreadId); 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); + await deliverToAgent(agent, agentGroup, mg, event, userId, threadsEnabled, effectiveThreadId, true); 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. + // requiring another @mention. Uses this wiring's OWN effective thread + // id — a non-null value already implies the adapter supports threads + // (resolveThreadPolicy hard-ANDs the capability). DMs, non-threaded + // platforms, and thread-opted-out wirings skip. if ( !subscribed && agent.engage_mode === 'mention-sticky' && - adapter?.supportsThreads && - adapter.subscribe && - event.threadId !== null && + adapter?.subscribe && + effectiveThreadId !== 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 }); + void adapter.subscribe(event.platformId, effectiveThreadId).catch((err) => { + log.warn('adapter.subscribe failed', { channelType: event.channelType, threadId: effectiveThreadId, err }); }); } } else if (agent.ignored_message_policy === 'accumulate' && !(engages && (!accessOk || !scopeOk))) { @@ -334,7 +362,7 @@ export async function routeInbound(event: InboundEvent): Promise { // message (which also stages their attachments to disk via // writeSessionMessage → extractAttachmentFiles) is exactly what the // gate is meant to prevent. - await deliverToAgent(agent, agentGroup, mg, event, userId, adapter?.supportsThreads === true, false); + await deliverToAgent(agent, agentGroup, mg, event, userId, threadsEnabled, effectiveThreadId, false); accumulatedCount++; } else { log.debug('Message not engaged for agent (drop policy)', { @@ -419,28 +447,33 @@ async function deliverToAgent( mg: MessagingGroup, event: InboundEvent, userId: string | null, - adapterSupportsThreads: boolean, + threadsEnabled: boolean, + effectiveThreadId: string | null, wake: boolean, ): Promise { - // 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). + // Apply the resolved thread policy (wiring override AND channel declaration + // AND adapter capability — resolveThreadPolicy at fanout): thread-enabled + // wiring in a group chat → per-thread session regardless of wiring + // session_mode. 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) { + if (threadsEnabled && effectiveSessionMode !== 'agent-shared' && mg.is_group !== 0) { effectiveSessionMode = 'per-thread'; } - const { session, created } = resolveSession(agent.agent_group_id, mg.id, event.threadId, effectiveSessionMode); + const { session, created } = resolveSession(agent.agent_group_id, mg.id, effectiveThreadId, effectiveSessionMode); // The inbound row's (channel_type, platform_id, thread_id) is the address // the agent's reply will be delivered to. Normally it mirrors the source - // (stamped from the event). When the caller supplied `replyTo` (CLI admin - // transport acting on operator intent), the reply is redirected there. + // (stamped from the event, with the wiring's thread policy applied). When + // the caller supplied `replyTo` (CLI admin transport acting on operator + // intent), the reply is redirected there — replyTo is exempt from + // thread-policy stripping. const deliveryAddr = event.replyTo ?? { channelType: event.channelType, platformId: event.platformId, - threadId: event.threadId, + threadId: effectiveThreadId, }; // Command gate: classify slash commands before they reach the container. @@ -497,7 +530,7 @@ async function deliverToAgent( session.agent_group_id, event.channelType, event.platformId, - event.threadId, + effectiveThreadId, mg.instance, ); const freshSession = getSession(session.id);