Fabric.OpenclawPlugin/index.ts

// Fabric channel plugin entry.
// COMPAT NOTE (openclaw v2026.5.7): defineChannelPluginEntry signature
// { id, name, description, plugin, setRuntime?, registerFull? }. setRuntime
// receives the PluginRuntime (has channel.turn kernel); registerFull receives
// the OpenClawPluginApi for runtime startup (transport + tools).
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 { getChannelType, flushChannelMeta } from './src/channel-meta.js';
import { FabricInbound } from './src/inbound.js';
import { listEnabledFabricAccounts } from './src/accounts.js';
import { registerFabricTools } from './src/tools.js';
import { FabricClient } from './src/fabric-client.js';
import { IdentityRegistry } from './src/identity.js';
import { syncFabricCommands } from './src/command-sync.js';
import { PresenceSync } from './src/presence-sync.js';
import { SubDiscussionStore } from './src/sub-discussion-store.js';
import { registerSubDiscussionHook } from './src/sub-discussion-hook.js';
import path from 'node:path';
import os from 'node:os';

let runtimeRef: unknown = null;
let inbound: FabricInbound | null = null;
let presence: PresenceSync | null = null;
// Periodic re-harvest of presence accounts so newly-connected agents
// (registered through tool-based identity flow AFTER initial start)
// get picked up. Cleared on gateway_stop.
let presenceRefreshTimer: ReturnType<typeof setInterval> | null = null;

export { fabricChannelPlugin } from './src/channel.js';

export default defineChannelPluginEntry({
  id: 'fabric',
  name: 'Fabric',
  description: 'Fabric channel plugin — OpenClaw agents speak in Fabric guilds',
  plugin: fabricChannelPlugin,

  setRuntime(runtime: unknown) {
    runtimeRef = runtime;
  },

  registerFull(apiRaw: OpenClawPluginApi) {
    // COMPAT: access the subset we use through a loose view so SDK type
    // drift in unrelated api members doesn't break the build.
    const api = apiRaw as unknown as {
      config?: unknown;
      pluginConfig?: { identityFilePath?: string };
      logger: { info: (m: string) => void; warn: (m: string) => void };
      on: (ev: string, fn: (...args: unknown[]) => unknown) => void;
      registerTool: (d: unknown) => void;
    };
    const cfg = (api.config ?? {}) as {
      channels?: { fabric?: { centerApiBase?: string; commandsSyncKey?: string } };
    };
    const centerApiBase = cfg.channels?.fabric?.centerApiBase ?? 'http://localhost:7001/api';
    const idFile =
      api.pluginConfig?.identityFilePath ??
      path.join(os.homedir(), '.openclaw', 'fabric-identity.json');
    const subDiscussionFile = path.join(
      os.homedir(),
      '.openclaw',
      'fabric-sub-discussion.json',
    );

    // tools operate against a default Center; per-account keys come from config
    const client = new FabricClient(centerApiBase);
    const identity = new IdentityRegistry(idFile);
    const subDiscussion = new SubDiscussionStore(subDiscussionFile);
    registerFabricTools(
      { registerTool: (d) => api.registerTool(d), logger: api.logger },
      client,
      identity,
      subDiscussion,
      cfg,
    );
    // Per-(agent, channel) prompt injection for sub-discussion channels.
    // Runs as a sibling to PrismFacet's before_prompt_build hook (and
    // ClawPrompts' fabric-chat-injector); openclaw composes
    // appendSystemContext from all registered handlers.
    registerSubDiscussionHook(
      { on: api.on, logger: api.logger },
      subDiscussion,
      identity,
    );

    // Cross-plugin API: globalThis.__fabric
    // Consumed by ClawPrompts' fabric-chat-injector to narrow its prompt
    // injection to DM-typed channels only. The channel-meta cache is
    // populated lazily from inbound (message.created carries xType) and
    // persisted to ~/.openclaw/fabric-channel-meta.json — so even the
    // very first DM after a fresh gateway start hits cache from the
    // previous run rather than firing the injector on the wrong type.
    //
    // null return = channel never seen (cache cold). Callers MUST NOT
    // fall back to "assume DM" — fail closed on unknown.
    {
      const _G = globalThis as Record<string, unknown>;
      _G['__fabric'] = {
        getChannelType,
        // Dynamic-subscription bridges: tools (notably `fabric-register`)
        // call these to add/remove an account's inbound socket without
        // a gateway restart. Both delegate to the live FabricInbound
        // instance via the module-level `inbound` closure variable; the
        // closures stay valid across gateway_start / gateway_stop
        // because we re-assign the variable, not the property.
        addAccount: async (entry: { agentId: string; fabricApiKey: string }) => {
          if (!inbound) throw new Error('fabric inbound not ready yet (gateway not started?)');
          await inbound.addAccount(entry);
        },
        removeAccount: (agentId: string) => {
          if (!inbound) return;
          inbound.removeAccount(agentId);
        },
      };
      // Flush channel-meta cache when the gateway shuts down so
      // recently-recorded xType entries don't get lost.
      api.on('gateway_stop', () => {
        try { flushChannelMeta(); } catch { /* ignore */ }
      });
      api.logger.info('fabric: __fabric cross-plugin API installed (getChannelType + addAccount + removeAccount)');
    }

    api.on('gateway_start', () => {
      const _G = globalThis as Record<string, unknown>;
      if (_G._fabricInboundStarted) return;
      _G._fabricInboundStarted = true;
      const accounts = listEnabledFabricAccounts(cfg as never).map((a) => ({
        agentId: a.accountId,
        fabricApiKey: a.fabricApiKey,
      }));
      // also include any tool-registered identities
      for (const e of identity.list()) {
        if (!accounts.some((x) => x.agentId === e.agentId)) {
          accounts.push({ agentId: e.agentId, fabricApiKey: e.fabricApiKey });
        }
      }
      if (!runtimeRef) {
        api.logger.warn('fabric: runtime not set; inbound disabled');
        return;
      }
      inbound = new FabricInbound(
        runtimeRef,
        api.config,
        client,
        identity,
        api.logger,
        accounts,
      );
      // start() resolves once all accounts have attempted login; per-
      // agent failures are logged but don't reject. Once it resolves we
      // can harvest the presence accounts (those that DID log in have
      // their fabricUserId + first guild endpoint populated).
      void inbound.start().then(() => {
        if (!inbound) return;
        presence = new PresenceSync(api.logger, client);
        presence.setAccounts(inbound.getPresenceAccounts());
        presence.start();
        api.logger.info(`fabric: presence-sync started for ${inbound.getPresenceAccounts().length} account(s)`);

        // Re-harvest every 5 min: catches agents added via tool-based
        // identity provisioning after gateway_start (recruitment flow).
        // setAccounts is idempotent — duplicates collapse on agentId.
        presenceRefreshTimer = setInterval(() => {
          if (inbound && presence) presence.setAccounts(inbound.getPresenceAccounts());
        }, 5 * 60_000);
      });
      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();
      if (presenceRefreshTimer) { clearInterval(presenceRefreshTimer); presenceRefreshTimer = null; }
      presence?.stop();
      presence = null;
      inbound?.stop();
      inbound = null;
    });
  },
});