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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PJkDAnLGWJUqjgJNXpyqGZ
2026-07-12 19:00:58 +08:00 · 2026-07-10 21:12:12 +03:00
parent 8d39368352
commit 53ffa3d347
2 changed files with 277 additions and 25 deletions
@@ -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<void>): Promise<void> {
+    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 ?')
@@ -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<void> {
      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<void> {
  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<void> {
    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<void> {
      // 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<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).
+  // 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);