feat(inbound): listen for backend-pushed channel.joined/left events

Companion to nav/Fabric.Backend.Guild#<TBD> which adds the server-side
emitToUser broadcast on channel membership changes. Before, the inbound
only learned about new channels via the 60s polling resync (worst-case
60s lag). Now the backend tells us directly so sub/unsub is realtime.

socket.on('channel.joined', evt) → join the socket.io room for evt.channelId
                                    and add to the local 'joined' set.
socket.on('channel.left',   evt) → leave + remove from 'joined'.

Both events are idempotent (`if (joined.has(id))` / `if (!joined.has(id))`)
so duplicate emits from server are safe. Polling resync still runs every
60s as a safety net for transient socket drops between emit and
reconnect, partial server failures, etc.

When backend lacks this support (older deployments), nothing breaks —
the event simply never fires and polling carries the load as before.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

README.md

@@ -72,6 +72,12 @@ Two ways, both write the same identity registry the transport reads:
   `FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY` (Guild C-2). Read it from the
   guild with `docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js`.
   Sourced from config only — never from the environment.
+- `channels.fabric.coalesce` — default `true`. OpenClaw splits one agent
+  turn whose blocks are `text → thinking/tool → text` into multiple
+  `deliver()` calls; this buffers them and posts ONE Fabric message at the
+  deterministic turn boundary (right after the inbound reply dispatch
+  resolves — no hooks, no timers, no idle guessing). `false` = raw
+  per-segment posting.
 - `channels.fabric.accounts.<agentId>` = `{ fabricApiKey, enabled }`
   (**agent = account**; the account id is the OpenClaw agentId)
 - plugin `identityFilePath` — default `~/.openclaw/fabric-identity.json`

dist/fabric/index.js

@@ -5,6 +5,7 @@
 // the OpenClawPluginApi for runtime startup (transport + tools).
 import { defineChannelPluginEntry } from 'openclaw/plugin-sdk/core';
 import { fabricChannelPlugin } from './src/channel.js';
+import { flushAllFabric } from './src/coalesce.js';
 import { FabricInbound } from './src/inbound.js';
 import { listEnabledFabricAccounts } from './src/accounts.js';
 import { registerFabricTools } from './src/tools.js';
@@ -60,7 +61,12 @@ export default defineChannelPluginEntry({
             api.logger.info(`fabric: inbound started for ${accounts.length} account(s)`);
             void syncFabricCommands(client, cfg, accounts, api.logger);
         });
+        // Note: the per-turn coalesce flush happens deterministically in
+        // inbound.ts right after dispatchInboundReplyWithBase resolves (that
+        // is the real "all deliveries done" boundary; the agent_end hook fires
+        // BEFORE deliver()). gateway_stop only flushes any leftover buffer.
         api.on('gateway_stop', () => {
+            void flushAllFabric();
             inbound?.stop();
             inbound = null;
         });

dist/fabric/src/accounts.js

@@ -13,6 +13,11 @@ function section(cfg) {
 export function resolveCommandsSyncKey(cfg) {
     return (section(cfg).commandsSyncKey ?? '').trim();
 }
+// Whether to coalesce a split agent turn into one Fabric message
+// (channel-level). Default true.
+export function resolveCoalesce(cfg) {
+    return (cfg.channels?.fabric ?? {}).coalesce !== false;
+}
 export function listFabricAccountIds(cfg) {
     const accts = section(cfg).accounts ?? {};
     const ids = Object.keys(accts);

dist/fabric/src/channel.js

@@ -137,7 +137,10 @@ export const fabricChannelPlugin = createChatChannelPlugin({
         attachedResults: {
             channel: 'fabric',
             sendText: async (ctx) => {
-                // openclaw passes config under cfg or config depending on path
+                // openclaw passes config under cfg or config depending on path.
+                // Note: inbound agent replies go through inbound.ts `deliver`
+                // (where turn coalescing happens). This path is for any direct
+                // outbound sends and posts immediately.
                 const cfg = (ctx.cfg ?? ctx.config ?? {});
                 try {
                     const r = await sendToFabric(cfg, ctx.accountId ?? null, ctx.to, ctx.text);

dist/fabric/src/coalesce.js

@@ -0,0 +1,75 @@
+// Deterministic turn coalescer.
+//
+// OpenClaw calls the Fabric `deliver` callback once per assistant text
+// segment; a thinking/tool block between two text blocks is a delivery
+// boundary, so one agent turn of `text → thinking/tool → text` arrives as
+// multiple deliver() calls. There is no turn id on the delivery, so we
+// BUFFER segments by Fabric channelId and post the merged message when the
+// turn truly ends. The flush is driven by inbound.ts right after
+// `dispatchInboundReplyWithBase` resolves — that only happens AFTER every
+// deliver() of the turn, a deterministic boundary (NOT a timer, NOT the
+// agent_end hook, which fires before deliver()). `coalesce=false` posts
+// each segment immediately.
+const SAFETY_FLUSH_MS = 120_000; // leak-guard only; not the flush mechanism
+export function normChannelId(x) {
+    const s = String(x ?? '');
+    return s.startsWith('fabric:') ? s.slice('fabric:'.length) : s;
+}
+const pendingByChannel = new Map();
+async function flushChannel(channelId, reason) {
+    const p = pendingByChannel.get(channelId);
+    if (!p)
+        return;
+    pendingByChannel.delete(channelId);
+    clearTimeout(p.safety);
+    const text = p.parts.join('\n\n').trim();
+    if (!text)
+        return;
+    try {
+        await p.post(text);
+        p.log?.(`fabric: flushed ${p.parts.length} segment(s) channel=${channelId} (${reason})`);
+    }
+    catch (e) {
+        p.log?.(`fabric: flush FAILED channel=${channelId} (${reason}): ${String(e)}`);
+    }
+}
+// Buffer one delivered segment (or send immediately when coalesce=false).
+// `post` performs the real Fabric postMessage with the caller's already
+// resolved guild/token; on flush it is called once with the merged text.
+export async function enqueueDelivery(params) {
+    const cid = normChannelId(params.channelId);
+    const text = (params.text ?? '').trim();
+    if (!text)
+        return;
+    if (!params.coalesce) {
+        await params.post(text);
+        return;
+    }
+    const existing = pendingByChannel.get(cid);
+    if (existing) {
+        existing.parts.push(text);
+        existing.post = params.post; // freshest guild/token closure
+        existing.log = params.log;
+    }
+    else {
+        pendingByChannel.set(cid, {
+            parts: [text],
+            post: params.post,
+            log: params.log,
+            safety: setTimeout(() => void flushChannel(cid, 'safety-timeout'), SAFETY_FLUSH_MS),
+        });
+    }
+}
+// Called by the agent_end hook with the hook ctx's channelId (bare or
+// fabric:-prefixed). Deterministic per-turn boundary.
+export async function flushFabricForChannel(rawChannelId) {
+    const cid = normChannelId(rawChannelId);
+    if (cid)
+        await flushChannel(cid, 'dispatch-end');
+}
+// gateway_stop: flush anything still buffered.
+export async function flushAllFabric() {
+    for (const cid of [...pendingByChannel.keys()]) {
+        await flushChannel(cid, 'gateway_stop');
+    }
+}

dist/fabric/src/inbound.js

@@ -3,6 +3,8 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { io } from 'socket.io-client';
 import { dispatchInboundReplyWithBase } from 'openclaw/plugin-sdk/inbound-reply-dispatch';
+import { resolveCoalesce } from './accounts.js';
+import { enqueueDelivery, flushFabricForChannel } from './coalesce.js';
 export class FabricInbound {
     core;
     cfg;
@@ -12,6 +14,14 @@ export class FabricInbound {
     accounts;
     sockets = [];
     seen = new Set();
+    // Timers that periodically re-sync channel membership per (agent, guild).
+    // Without this, the agent's socket.io subscriptions are a snapshot taken
+    // at connect time — any channel the agent joins later (e.g. a fresh DM
+    // created by another user) is unreachable until the gateway restarts.
+    channelSyncTimers = [];
+    // Resync cadence. Backend doesn't push a `channel.joined` event, so we
+    // poll. 60s keeps the lag bounded without hammering the backend.
+    static CHANNEL_SYNC_INTERVAL_MS = 60_000;
     // Guild access tokens are short-lived (~15 min). The socket survives via
     // socket.io reconnect, but the token captured at connect time goes stale,
     // so HTTP calls (attachment download, posting the reply) start 401ing.
@@ -73,6 +83,9 @@ export class FabricInbound {
         }
     }
     stop() {
+        for (const t of this.channelSyncTimers)
+            clearInterval(t);
+        this.channelSyncTimers = [];
         for (const s of this.sockets)
             s.disconnect();
         this.sockets = [];
@@ -88,19 +101,83 @@ export class FabricInbound {
                 auth: { token: tok },
                 autoConnect: false,
             });
-            const joinAll = async () => {
+            // Tracked socket.io rooms for this (agent, guild). The initial fetch
+            // on `connect` seeds it; the periodic resync diffs against it so we
+            // only emit `join_channel` for genuinely new channels (and
+            // `leave_channel` for ones the agent is no longer in).
+            const joined = new Set();
+            const syncChannels = async (kind) => {
+                let freshTok;
                 try {
-                    const res = await fetch(`${g.endpoint}/api/channels?guildId=${encodeURIComponent(g.nodeId)}`, { headers: { authorization: `Bearer ${tok}` } });
-                    const channels = res.ok ? (await res.json()) : [];
-                    for (const c of channels)
-                        socket.emit('join_channel', { channelId: c.id });
-                    this.log.info(`fabric: agent ${agentId} joined ${channels.length} channel(s) on ${g.nodeId}`);
+                    freshTok = await this.freshGuildToken(agentId, g.nodeId, session);
                 }
                 catch {
-                    /* best effort */
+                    freshTok = tok;
+                }
+                const authTok = freshTok ?? tok;
+                try {
+                    const res = await fetch(`${g.endpoint}/api/channels?guildId=${encodeURIComponent(g.nodeId)}`, { headers: { authorization: `Bearer ${authTok}` } });
+                    if (!res.ok)
+                        return;
+                    const channels = (await res.json());
+                    const current = new Set(channels.map((c) => c.id));
+                    let added = 0;
+                    let removed = 0;
+                    for (const id of current) {
+                        if (!joined.has(id)) {
+                            socket.emit('join_channel', { channelId: id });
+                            joined.add(id);
+                            added++;
+                        }
+                    }
+                    for (const id of [...joined]) {
+                        if (!current.has(id)) {
+                            socket.emit('leave_channel', { channelId: id });
+                            joined.delete(id);
+                            removed++;
+                        }
+                    }
+                    if (kind === 'initial') {
+                        this.log.info(`fabric: agent ${agentId} joined ${current.size} channel(s) on ${g.nodeId}`);
+                    }
+                    else if (added > 0 || removed > 0) {
+                        this.log.info(`fabric: agent ${agentId} channel resync on ${g.nodeId}: +${added} -${removed} (now ${joined.size})`);
+                    }
+                }
+                catch {
+                    /* best effort — next tick will retry */
                 }
             };
-            socket.on('connect', () => void joinAll());
+            socket.on('connect', () => {
+                // On every (re)connect the server forgets prior subscriptions, so
+                // reset our local view and seed from a fresh fetch.
+                joined.clear();
+                void syncChannels('initial');
+            });
+            // Push-based membership events from the backend (companion to
+            // Fabric.Backend.Guild's RealtimeGateway.emitToUser). When the
+            // server tells us this user was added to / removed from a
+            // channel, we sub/unsub the socket.io room immediately — no
+            // 60s wait for the polling resync. Polling remains as a safety
+            // net for missed events.
+            socket.on('channel.joined', (evt) => {
+                const id = evt?.channelId;
+                if (!id || joined.has(id))
+                    return;
+                socket.emit('join_channel', { channelId: id });
+                joined.add(id);
+                this.log.info(`fabric: agent ${agentId} channel.joined push on ${g.nodeId}: ${id} (now ${joined.size})`);
+            });
+            socket.on('channel.left', (evt) => {
+                const id = evt?.channelId;
+                if (!id || !joined.has(id))
+                    return;
+                socket.emit('leave_channel', { channelId: id });
+                joined.delete(id);
+                this.log.info(`fabric: agent ${agentId} channel.left push on ${g.nodeId}: ${id} (now ${joined.size})`);
+            });
+            const syncTimer = setInterval(() => void syncChannels('resync'), FabricInbound.CHANNEL_SYNC_INTERVAL_MS);
+            this.channelSyncTimers.push(syncTimer);
             socket.on('message.created', (m) => {
                 const channelId = m.channelId ?? '';
                 if (!channelId)
@@ -199,7 +276,11 @@ export class FabricInbound {
             // the woken speaker emits a normal message or /no-reply). We still
             // record the message into the agent's session so it has the full
             // channel conversation as context whenever it IS later woken.
-            if (m.wakeup !== true) {
+            //
+            // Exception: dm channels are 1:1 — there is no turn/wakeup gating;
+            // any message that isn't the agent's own (already filtered above) is
+            // always delivered to the model.
+            if (m.xType !== 'dm' && m.wakeup !== true) {
                 const ctxPayload = core.channel.reply.finalizeInboundContext(baseCtx);
                 await core.channel.session.recordInboundSession({
                     storePath,
@@ -244,8 +325,16 @@ export class FabricInbound {
                     this.log.info(`fabric: deliver agent=${agentId} channel=${channelId} len=${text.length}`);
                     if (!text || !gt)
                         return;
-                    await this.client.postMessage(guild.endpoint, gt, channelId, text, session.user.id);
-                    this.log.info(`fabric: posted reply agent=${agentId} channel=${channelId}`);
+                    // Buffer segments; the merged message is posted right after
+                    // dispatch returns (the deterministic turn boundary, see the
+                    // finally below). Disable per channel: channels.fabric.coalesce.
+                    await enqueueDelivery({
+                        channelId,
+                        text,
+                        coalesce: resolveCoalesce(this.cfg),
+                        post: (t) => this.client.postMessage(guild.endpoint, gt, channelId, t, session.user.id),
+                        log: (m) => this.log.info(m),
+                    });
                 },
                 onRecordError: (err) => this.log.warn(`fabric: session record failed agent=${agentId}: ${String(err)}`),
                 onDispatchError: (err, info) => this.log.warn(`fabric: ${info.kind} dispatch failed agent=${agentId}: ${String(err)}`),
@@ -270,5 +359,12 @@ export class FabricInbound {
         catch (err) {
             this.log.warn(`fabric: dispatch failed agent=${agentId} channel=${channelId}: ${String(err)}`);
         }
+        finally {
+            // Deterministic per-turn boundary: dispatchInboundReplyWithBase only
+            // resolves AFTER every deliver() call of this turn has run, so the
+            // buffer now holds all segments — flush them as ONE Fabric message.
+            // No hooks, no timers, no idle guessing.
+            await flushFabricForChannel(channelId);
+        }
     }
 }

index.ts

@@ -6,6 +6,7 @@
 import { defineChannelPluginEntry } from 'openclaw/plugin-sdk/core';
 import type { OpenClawPluginApi } from 'openclaw/plugin-sdk/core';
 import { fabricChannelPlugin } from './src/channel.js';
+import { flushAllFabric } from './src/coalesce.js';
 import { FabricInbound } from './src/inbound.js';
 import { listEnabledFabricAccounts } from './src/accounts.js';
 import { registerFabricTools } from './src/tools.js';
@@ -37,7 +38,7 @@ export default defineChannelPluginEntry({
       config?: unknown;
       pluginConfig?: { identityFilePath?: string };
       logger: { info: (m: string) => void; warn: (m: string) => void };
-      on: (ev: string, fn: () => void) => void;
+      on: (ev: string, fn: (...args: unknown[]) => unknown) => void;
       registerTool: (d: unknown) => void;
     };
     const cfg = (api.config ?? {}) as { channels?: { fabric?: { centerApiBase?: string } } };
@@ -86,7 +87,12 @@ export default defineChannelPluginEntry({
       void syncFabricCommands(client, cfg, accounts, api.logger);
     });
 
+    // Note: the per-turn coalesce flush happens deterministically in
+    // inbound.ts right after dispatchInboundReplyWithBase resolves (that
+    // is the real "all deliveries done" boundary; the agent_end hook fires
+    // BEFORE deliver()). gateway_stop only flushes any leftover buffer.
     api.on('gateway_stop', () => {
+      void flushAllFabric();
       inbound?.stop();
       inbound = null;
     });

openclaw.plugin.json

@@ -44,6 +44,10 @@
             "minLength": 1,
             "description": "Shared secret that must equal the guild's FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY. Required to register the slash-command catalog (Guild C-2). Read it from the guild via: docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js"
           },
+          "coalesce": {
+            "type": "boolean",
+            "description": "Merge a split agent turn (text → thinking/tool → text) into ONE Fabric message. Flushed deterministically on the agent_end hook. Default true; false = raw per-segment posting."
+          },
           "dmSecurity": { "type": "string" },
           "dmPolicy": { "type": "string" },
           "enabled": { "type": "boolean" },

src/accounts.ts

@@ -18,6 +18,11 @@ export type FabricChannelConfig = {
   // (Guild C-2). Required by the channel config schema; sourced from config
   // only — never from the environment.
   commandsSyncKey?: string;
+  // Coalesce an agent turn that OpenClaw split into multiple deliveries
+  // (text → thinking/tool → text => N sendText calls) into ONE Fabric
+  // message. The flush boundary is the deterministic `agent_end` hook (not
+  // a timer). Default true; set false for raw per-segment posting.
+  coalesce?: boolean;
   accounts?: Record<string, FabricAccountConfig>;
   defaultAccount?: string;
 } & FabricAccountConfig;
@@ -46,6 +51,12 @@ export function resolveCommandsSyncKey(cfg: Cfg): string {
   return (section(cfg).commandsSyncKey ?? '').trim();
 }
 
+// Whether to coalesce a split agent turn into one Fabric message
+// (channel-level). Default true.
+export function resolveCoalesce(cfg: Cfg): boolean {
+  return (cfg.channels?.fabric ?? {}).coalesce !== false;
+}
+
 export function listFabricAccountIds(cfg: Cfg): string[] {
   const accts = section(cfg).accounts ?? {};
   const ids = Object.keys(accts);

src/channel.ts

@@ -159,7 +159,10 @@ export const fabricChannelPlugin = createChatChannelPlugin<ResolvedFabricAccount
         cfg?: unknown;
         config?: unknown;
       }) => {
-        // openclaw passes config under cfg or config depending on path
+        // openclaw passes config under cfg or config depending on path.
+        // Note: inbound agent replies go through inbound.ts `deliver`
+        // (where turn coalescing happens). This path is for any direct
+        // outbound sends and posts immediately.
         const cfg = (ctx.cfg ?? ctx.config ?? {}) as AnyCfg;
         try {
           const r = await sendToFabric(cfg, ctx.accountId ?? null, ctx.to, ctx.text);

src/coalesce.ts

@@ -0,0 +1,93 @@
+// Deterministic turn coalescer.
+//
+// OpenClaw calls the Fabric `deliver` callback once per assistant text
+// segment; a thinking/tool block between two text blocks is a delivery
+// boundary, so one agent turn of `text → thinking/tool → text` arrives as
+// multiple deliver() calls. There is no turn id on the delivery, so we
+// BUFFER segments by Fabric channelId and post the merged message when the
+// turn truly ends. The flush is driven by inbound.ts right after
+// `dispatchInboundReplyWithBase` resolves — that only happens AFTER every
+// deliver() of the turn, a deterministic boundary (NOT a timer, NOT the
+// agent_end hook, which fires before deliver()). `coalesce=false` posts
+// each segment immediately.
+
+const SAFETY_FLUSH_MS = 120_000; // leak-guard only; not the flush mechanism
+
+export function normChannelId(x: string | null | undefined): string {
+  const s = String(x ?? '');
+  return s.startsWith('fabric:') ? s.slice('fabric:'.length) : s;
+}
+
+type Pending = {
+  parts: string[];
+  post: (text: string) => Promise<void>;
+  log?: (m: string) => void;
+  safety: ReturnType<typeof setTimeout>;
+};
+const pendingByChannel = new Map<string, Pending>();
+
+async function flushChannel(channelId: string, reason: string): Promise<void> {
+  const p = pendingByChannel.get(channelId);
+  if (!p) return;
+  pendingByChannel.delete(channelId);
+  clearTimeout(p.safety);
+  const text = p.parts.join('\n\n').trim();
+  if (!text) return;
+  try {
+    await p.post(text);
+    p.log?.(`fabric: flushed ${p.parts.length} segment(s) channel=${channelId} (${reason})`);
+  } catch (e) {
+    p.log?.(`fabric: flush FAILED channel=${channelId} (${reason}): ${String(e)}`);
+  }
+}
+
+// Buffer one delivered segment (or send immediately when coalesce=false).
+// `post` performs the real Fabric postMessage with the caller's already
+// resolved guild/token; on flush it is called once with the merged text.
+export async function enqueueDelivery(params: {
+  channelId: string;
+  text: string;
+  coalesce: boolean;
+  post: (text: string) => Promise<void>;
+  log?: (m: string) => void;
+}): Promise<void> {
+  const cid = normChannelId(params.channelId);
+  const text = (params.text ?? '').trim();
+  if (!text) return;
+  if (!params.coalesce) {
+    await params.post(text);
+    return;
+  }
+  const existing = pendingByChannel.get(cid);
+  if (existing) {
+    existing.parts.push(text);
+    existing.post = params.post; // freshest guild/token closure
+    existing.log = params.log;
+  } else {
+    pendingByChannel.set(cid, {
+      parts: [text],
+      post: params.post,
+      log: params.log,
+      safety: setTimeout(
+        () => void flushChannel(cid, 'safety-timeout'),
+        SAFETY_FLUSH_MS,
+      ),
+    });
+  }
+}
+
+// Called by the agent_end hook with the hook ctx's channelId (bare or
+// fabric:-prefixed). Deterministic per-turn boundary.
+export async function flushFabricForChannel(
+  rawChannelId: string | null | undefined,
+): Promise<void> {
+  const cid = normChannelId(rawChannelId);
+  if (cid) await flushChannel(cid, 'dispatch-end');
+}
+
+// gateway_stop: flush anything still buffered.
+export async function flushAllFabric(): Promise<void> {
+  for (const cid of [...pendingByChannel.keys()]) {
+    await flushChannel(cid, 'gateway_stop');
+  }
+}

src/inbound.ts

@@ -5,6 +5,8 @@ import { io, type Socket } from 'socket.io-client';
 import { dispatchInboundReplyWithBase } from 'openclaw/plugin-sdk/inbound-reply-dispatch';
 import type { FabricClient, FabricSession } from './fabric-client.js';
 import type { IdentityRegistry } from './identity.js';
+import { resolveCoalesce } from './accounts.js';
+import { enqueueDelivery, flushFabricForChannel } from './coalesce.js';
 
 // COMPAT NOTE (openclaw v2026.5.7): the inbound path mirrors how bundled
 // channels (nextcloud-talk) drive the kernel:
@@ -42,11 +44,22 @@ type FabricMessage = {
   channelId?: string;
   attachments?: FabricAttachment[];
   wakeup?: boolean;
+  // x-type of the channel (sent on message.created). 'dm' bypasses the
+  // wakeup gate: any message that isn't the agent's own is delivered.
+  xType?: string;
 };
 
 export class FabricInbound {
   private sockets: Socket[] = [];
   private seen = new Set<string>();
+  // Timers that periodically re-sync channel membership per (agent, guild).
+  // Without this, the agent's socket.io subscriptions are a snapshot taken
+  // at connect time — any channel the agent joins later (e.g. a fresh DM
+  // created by another user) is unreachable until the gateway restarts.
+  private channelSyncTimers: NodeJS.Timeout[] = [];
+  // Resync cadence. Backend doesn't push a `channel.joined` event, so we
+  // poll. 60s keeps the lag bounded without hammering the backend.
+  private static readonly CHANNEL_SYNC_INTERVAL_MS = 60_000;
   // Guild access tokens are short-lived (~15 min). The socket survives via
   // socket.io reconnect, but the token captured at connect time goes stale,
   // so HTTP calls (attachment download, posting the reply) start 401ing.
@@ -114,6 +127,8 @@ export class FabricInbound {
   }
 
   stop(): void {
+    for (const t of this.channelSyncTimers) clearInterval(t);
+    this.channelSyncTimers = [];
     for (const s of this.sockets) s.disconnect();
     this.sockets = [];
   }
@@ -128,20 +143,87 @@ export class FabricInbound {
         auth: { token: tok },
         autoConnect: false,
       });
-      const joinAll = async () => {
+      // Tracked socket.io rooms for this (agent, guild). The initial fetch
+      // on `connect` seeds it; the periodic resync diffs against it so we
+      // only emit `join_channel` for genuinely new channels (and
+      // `leave_channel` for ones the agent is no longer in).
+      const joined = new Set<string>();
+      const syncChannels = async (kind: 'initial' | 'resync') => {
+        let freshTok: string | undefined;
+        try {
+          freshTok = await this.freshGuildToken(agentId, g.nodeId, session);
+        } catch {
+          freshTok = tok;
+        }
+        const authTok = freshTok ?? tok;
         try {
           const res = await fetch(
             `${g.endpoint}/api/channels?guildId=${encodeURIComponent(g.nodeId)}`,
-            { headers: { authorization: `Bearer ${tok}` } },
+            { headers: { authorization: `Bearer ${authTok}` } },
           );
-          const channels = res.ok ? ((await res.json()) as Array<{ id: string }>) : [];
-          for (const c of channels) socket.emit('join_channel', { channelId: c.id });
-          this.log.info(`fabric: agent ${agentId} joined ${channels.length} channel(s) on ${g.nodeId}`);
+          if (!res.ok) return;
+          const channels = (await res.json()) as Array<{ id: string }>;
+          const current = new Set(channels.map((c) => c.id));
+          let added = 0;
+          let removed = 0;
+          for (const id of current) {
+            if (!joined.has(id)) {
+              socket.emit('join_channel', { channelId: id });
+              joined.add(id);
+              added++;
+            }
+          }
+          for (const id of [...joined]) {
+            if (!current.has(id)) {
+              socket.emit('leave_channel', { channelId: id });
+              joined.delete(id);
+              removed++;
+            }
+          }
+          if (kind === 'initial') {
+            this.log.info(
+              `fabric: agent ${agentId} joined ${current.size} channel(s) on ${g.nodeId}`,
+            );
+          } else if (added > 0 || removed > 0) {
+            this.log.info(
+              `fabric: agent ${agentId} channel resync on ${g.nodeId}: +${added} -${removed} (now ${joined.size})`,
+            );
+          }
         } catch {
-          /* best effort */
+          /* best effort — next tick will retry */
         }
       };
-      socket.on('connect', () => void joinAll());
+      socket.on('connect', () => {
+        // On every (re)connect the server forgets prior subscriptions, so
+        // reset our local view and seed from a fresh fetch.
+        joined.clear();
+        void syncChannels('initial');
+      });
+      // Push-based membership events from the backend (companion to
+      // Fabric.Backend.Guild's RealtimeGateway.emitToUser). When the
+      // server tells us this user was added to / removed from a
+      // channel, we sub/unsub the socket.io room immediately — no
+      // 60s wait for the polling resync. Polling remains as a safety
+      // net for missed events.
+      socket.on('channel.joined', (evt: { channelId?: string }) => {
+        const id = evt?.channelId;
+        if (!id || joined.has(id)) return;
+        socket.emit('join_channel', { channelId: id });
+        joined.add(id);
+        this.log.info(`fabric: agent ${agentId} channel.joined push on ${g.nodeId}: ${id} (now ${joined.size})`);
+      });
+      socket.on('channel.left', (evt: { channelId?: string }) => {
+        const id = evt?.channelId;
+        if (!id || !joined.has(id)) return;
+        socket.emit('leave_channel', { channelId: id });
+        joined.delete(id);
+        this.log.info(`fabric: agent ${agentId} channel.left push on ${g.nodeId}: ${id} (now ${joined.size})`);
+      });
+      const syncTimer = setInterval(
+        () => void syncChannels('resync'),
+        FabricInbound.CHANNEL_SYNC_INTERVAL_MS,
+      );
+      this.channelSyncTimers.push(syncTimer);
       socket.on('message.created', (m: FabricMessage) => {
         const channelId = m.channelId ?? '';
         if (!channelId) return;
@@ -250,7 +332,11 @@ export class FabricInbound {
       // the woken speaker emits a normal message or /no-reply). We still
       // record the message into the agent's session so it has the full
       // channel conversation as context whenever it IS later woken.
-      if (m.wakeup !== true) {
+      //
+      // Exception: dm channels are 1:1 — there is no turn/wakeup gating;
+      // any message that isn't the agent's own (already filtered above) is
+      // always delivered to the model.
+      if (m.xType !== 'dm' && m.wakeup !== true) {
         const ctxPayload = core.channel.reply.finalizeInboundContext(baseCtx);
         await core.channel.session.recordInboundSession({
           storePath,
@@ -301,8 +387,17 @@ export class FabricInbound {
           const text = (payload?.text ?? '').trim();
           this.log.info(`fabric: deliver agent=${agentId} channel=${channelId} len=${text.length}`);
           if (!text || !gt) return;
-          await this.client.postMessage(guild.endpoint, gt, channelId, text, session.user.id);
-          this.log.info(`fabric: posted reply agent=${agentId} channel=${channelId}`);
+          // Buffer segments; the merged message is posted right after
+          // dispatch returns (the deterministic turn boundary, see the
+          // finally below). Disable per channel: channels.fabric.coalesce.
+          await enqueueDelivery({
+            channelId,
+            text,
+            coalesce: resolveCoalesce(this.cfg as never),
+            post: (t) =>
+              this.client.postMessage(guild.endpoint, gt, channelId, t, session.user.id) as Promise<void>,
+            log: (m) => this.log.info(m),
+          });
         },
         onRecordError: (err: unknown) =>
           this.log.warn(`fabric: session record failed agent=${agentId}: ${String(err)}`),
@@ -327,6 +422,12 @@ export class FabricInbound {
       this.log.info(`fabric: dispatch returned agent=${agentId} channel=${channelId}`);
     } catch (err) {
       this.log.warn(`fabric: dispatch failed agent=${agentId} channel=${channelId}: ${String(err)}`);
+    } finally {
+      // Deterministic per-turn boundary: dispatchInboundReplyWithBase only
+      // resolves AFTER every deliver() call of this turn has run, so the
+      // buffer now holds all segments — flush them as ONE Fabric message.
+      // No hooks, no timers, no idle guessing.
+      await flushFabricForChannel(channelId);
     }
   }
 }