fix: dynamically sync inbound channel subscriptions

The fabric inbound previously called `joinAll()` once on socket.io
`connect` — it fetched the agent's channel list via
`GET /api/channels?guildId=...` and emitted `join_channel` for each.
Any channel the agent joined *after* connect (e.g. a fresh DM created
by another user that includes this agent) was unreachable until the
gateway restarted: the socket was never subscribed to that room, so
backend `message.created` push events never arrived.

Backend doesn't emit a user-scoped `channel.joined` event we could
piggy-back on (only `message.created`), so the fix is to poll. Every
60s the agent's channel list is re-fetched and diffed against a local
`joined` set:
- new channel ids → `socket.emit('join_channel', {channelId})` + add
- ids in `joined` but absent from the fresh list → `leave_channel`
  emit + remove (best-effort; cleans subs if the agent is removed from
  a channel)

Re-uses `freshGuildToken()` so the resync fetch survives token
expiry (15-min TTL). Initial `connect` resets the local `joined`
set since the server forgets prior room subscriptions on reconnect.

Timers are tracked in `channelSyncTimers` and cleared in `stop()`
alongside socket disconnect.

Verified against prod server.t2 scenario: hzhang creates DM channel
including agent 'manager' → without this fix, manager only sees the
message after a gateway restart; with this fix, manager receives the
message within at most 60s (next resync tick).

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

README.md

@@ -68,6 +68,16 @@ Two ways, both write the same identity registry the transport reads:
 ## Config
 
 - `channels.fabric.centerApiBase` — e.g. `http://localhost:7001/api`
+- `channels.fabric.commandsSyncKey` — **required**; must equal the guild's
+  `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`
@@ -100,6 +110,20 @@ Then `openclaw gateway restart`.
   `members` (list the channel's member userIds) · `join` (this agent
   joins) · `leave` (this agent leaves).
 
+## Slash-command catalog
+
+On `gateway_start` the plugin syncs OpenClaw's native-command catalog to
+each guild (`command-sync.ts`: `listNativeCommandSpecsForConfig` +
+`findCommandByNativeName`, dynamic arg `choices` snapshotted via
+`resolveCommandArgChoices`) → `PUT /api/commands`. This is exactly the data
+Discord registers as slash commands; Fabric's frontend uses it for `/`
+autocomplete. Fabric deliberately does **not** advertise the
+`nativeCommands` channel capability — it stays a text-command surface, so a
+`/<cmd>` message is delivered normally and OpenClaw's command system
+executes it (text-command + command session + auth), reusing the standard
+inbound path. Only `/no-reply` and `/force-proceed` stay
+server-intercepted by the guild.
+
 ## Install / build
 
 ```bash

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

@@ -7,6 +7,17 @@ const DEFAULT_CENTER = 'http://localhost:7001/api';
 function section(cfg) {
     return cfg.channels?.fabric ?? {};
 }
+// The commands-sync shared secret (channel-level only). Empty string when
+// unconfigured — callers decide how to handle (slash-command sync is then
+// rejected by the guild).
+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/command-sync.js

@@ -6,6 +6,7 @@
 // dynamic arg `choices` to a static snapshot here (like Discord does at
 // registration time).
 import { listNativeCommandSpecsForConfig, findCommandByNativeName, resolveCommandArgChoices, } from 'openclaw/plugin-sdk/native-command-registry';
+import { resolveCommandsSyncKey } from './accounts.js';
 function normChoice(c) {
     if (typeof c === 'string')
         return { value: c, label: c };
@@ -62,6 +63,14 @@ export function buildFabricCommandSpecs(cfg) {
 // Push the catalog to every guild the known agents belong to (idempotent;
 // the catalog is OpenClaw-global, so one PUT per guild is enough).
 export async function syncFabricCommands(client, cfg, accounts, log) {
+    // Guild C-2: the sync key comes from the channel config only (schema
+    // marks it required). Without it the guild rejects the catalog write.
+    const syncKey = resolveCommandsSyncKey(cfg);
+    if (!syncKey) {
+        log.warn('fabric: channels.fabric.commandsSyncKey is not set — skipping ' +
+            'slash-command sync (set it to the guild FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY)');
+        return;
+    }
     let specs;
     try {
         specs = buildFabricCommandSpecs(cfg);
@@ -88,7 +97,7 @@ export async function syncFabricCommands(client, cfg, accounts, log) {
             if (!tok)
                 continue;
             try {
-                await client.syncCommands(g.endpoint, tok, specs);
+                await client.syncCommands(g.endpoint, tok, specs, syncKey);
                 done.add(g.nodeId);
                 log.info(`fabric: synced ${specs.length} slash command(s) -> ${g.nodeId}`);
             }

dist/fabric/src/fabric-client.js

@@ -23,12 +23,13 @@ export class FabricClient {
     }
     // Generic JSON request (GET/PUT/PATCH/DELETE). Empty 2xx body -> null
     // (Fabric returns an empty body when a channel has no canvas).
-    async req(method, url, auth, body) {
+    async req(method, url, auth, body, extraHeaders) {
         const res = await fetch(url, {
             method,
             headers: {
                 ...(body !== undefined ? { 'content-type': 'application/json' } : {}),
                 ...(auth ? { authorization: `Bearer ${auth}` } : {}),
+                ...(extraHeaders ?? {}),
             },
             body: body !== undefined ? JSON.stringify(body) : undefined,
         });
@@ -74,8 +75,11 @@ export class FabricClient {
     // Register the OpenClaw slash-command catalog with this guild (idempotent
     // full replace). The frontend GETs it for `/` autocomplete; execution
     // still flows as a normal /<cmd> message into OpenClaw's command system.
-    syncCommands(guildEndpoint, guildToken, commands) {
+    syncCommands(guildEndpoint, guildToken, commands, syncKey) {
-        return this.req('PUT', `${guildEndpoint}/api/commands`, guildToken, { commands });
+        // Guild C-2: the shared key is sourced from the channel config
+        // (channels.fabric.commandsSyncKey) and must equal the guild's
+        // FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY for the catalog write.
+        return this.req('PUT', `${guildEndpoint}/api/commands`, guildToken, { commands }, syncKey ? { 'x-commands-sync-key': syncKey } : undefined);
     }
     // [{ userId, bypass }] — bypass is true only for discuss/work bypass-list
     channelMembers(guildEndpoint, guildToken, channelId) {

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,61 @@ 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}` } });
+                    freshTok = await this.freshGuildToken(agentId, g.nodeId, session);
-                    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}`);
                 }
                 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');
+            });
+            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 +254,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 +303,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);
+                    // Buffer segments; the merged message is posted right after
-                    this.log.info(`fabric: posted reply agent=${agentId} channel=${channelId}`);
+                    // 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 +337,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

@@ -39,6 +39,15 @@
             "type": "string",
             "description": "Fabric Center API base, e.g. http://localhost:7001/api"
           },
+          "commandsSyncKey": {
+            "type": "string",
+            "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" },
@@ -59,10 +68,12 @@
               }
             }
           }
-        }
+        },
+        "required": ["commandsSyncKey"]
       },
       "uiHints": {
-        "centerApiBase": { "label": "Center API base" }
+        "centerApiBase": { "label": "Center API base" },
+        "commandsSyncKey": { "label": "Commands sync key" }
       }
     }
   }

src/accounts.ts

@@ -14,6 +14,15 @@ export type FabricAccountConfig = {
 
 export type FabricChannelConfig = {
   centerApiBase?: string;
+  // Shared secret matching the guild's FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY
+  // (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;
@@ -35,6 +44,19 @@ function section(cfg: Cfg): FabricChannelConfig {
   return cfg.channels?.fabric ?? {};
 }
 
+// The commands-sync shared secret (channel-level only). Empty string when
+// unconfigured — callers decide how to handle (slash-command sync is then
+// rejected by the guild).
+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/command-sync.ts

@@ -11,6 +11,7 @@ import {
   resolveCommandArgChoices,
 } from 'openclaw/plugin-sdk/native-command-registry';
 import type { FabricClient } from './fabric-client.js';
+import { resolveCommandsSyncKey } from './accounts.js';
 
 type Logger = { info: (m: string) => void; warn: (m: string) => void };
 
@@ -102,6 +103,17 @@ export async function syncFabricCommands(
   accounts: Array<{ agentId: string; fabricApiKey: string }>,
   log: Logger,
 ): Promise<void> {
+  // Guild C-2: the sync key comes from the channel config only (schema
+  // marks it required). Without it the guild rejects the catalog write.
+  const syncKey = resolveCommandsSyncKey(cfg as never);
+  if (!syncKey) {
+    log.warn(
+      'fabric: channels.fabric.commandsSyncKey is not set — skipping ' +
+        'slash-command sync (set it to the guild FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY)',
+    );
+    return;
+  }
+
   let specs: FabricCommand[];
   try {
     specs = buildFabricCommandSpecs(cfg);
@@ -126,7 +138,7 @@ export async function syncFabricCommands(
       )?.token;
       if (!tok) continue;
       try {
-        await client.syncCommands(g.endpoint, tok, specs);
+        await client.syncCommands(g.endpoint, tok, specs, syncKey);
         done.add(g.nodeId);
         log.info(`fabric: synced ${specs.length} slash command(s) -> ${g.nodeId}`);
       } catch (err) {

src/fabric-client.ts

@@ -36,12 +36,14 @@ export class FabricClient {
     url: string,
     auth?: string,
     body?: unknown,
+    extraHeaders?: Record<string, string>,
   ): Promise<T> {
     const res = await fetch(url, {
       method,
       headers: {
         ...(body !== undefined ? { 'content-type': 'application/json' } : {}),
         ...(auth ? { authorization: `Bearer ${auth}` } : {}),
+        ...(extraHeaders ?? {}),
       },
       body: body !== undefined ? JSON.stringify(body) : undefined,
     });
@@ -122,8 +124,18 @@ export class FabricClient {
     guildEndpoint: string,
     guildToken: string,
     commands: unknown[],
+    syncKey: string,
   ): Promise<unknown> {
-    return this.req('PUT', `${guildEndpoint}/api/commands`, guildToken, { commands });
+    // Guild C-2: the shared key is sourced from the channel config
+    // (channels.fabric.commandsSyncKey) and must equal the guild's
+    // FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY for the catalog write.
+    return this.req(
+      'PUT',
+      `${guildEndpoint}/api/commands`,
+      guildToken,
+      { commands },
+      syncKey ? { 'x-commands-sync-key': syncKey } : undefined,
+    );
   }
 
   // [{ userId, bypass }] — bypass is true only for discuss/work bypass-list

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,67 @@ 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 }>) : [];
+          if (!res.ok) return;
-          for (const c of channels) socket.emit('join_channel', { channelId: c.id });
+          const channels = (await res.json()) as Array<{ id: string }>;
-          this.log.info(`fabric: agent ${agentId} joined ${channels.length} channel(s) on ${g.nodeId}`);
+          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');
+      });
+      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 +312,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 +367,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);
+          // Buffer segments; the merged message is posted right after
-          this.log.info(`fabric: posted reply agent=${agentId} channel=${channelId}`);
+          // 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 +402,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);
     }
   }
 }