Files
Fabric.OpenclawPlugin/index.ts
hzhang f8c8c21727 feat(gateway): register fabric.register gateway method for live no-restart registration
Recruitment's register-agent step is a plain shell step (no LLM turn), so it
cannot invoke the `fabric-register` TOOL (tool only fires inside an agent
turn) and there is no `openclaw tools call` CLI. It previously fell back to
the standalone bootstrap binary, which writes fabric-identity.json but cannot
notify the running plugin -> the new agent's inbound socket only came up after
a gateway restart.

This adds an in-process gateway method `fabric.register` (scope:
operator.write) whose handler runs inbound.addAccount: validates the key,
persists identity, and brings the inbound socket up immediately. The script
now calls `openclaw gateway call fabric.register --params ...` and only falls
back to the bootstrap binary if the method is unavailable.

Also resyncs committed dist/ to source (sub-discussion-hook/store + tools/
inbound were source-committed but their dist artifacts were stale).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-07 21:04:38 +01:00

225 lines
9.7 KiB
TypeScript

// 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;
registerGatewayMethod: (
method: string,
handler: (req: {
params?: unknown;
respond: (
ok: boolean,
data?: unknown,
error?: { code: string; message: string },
) => void;
}) => void | Promise<void>,
opts?: { scope?: string },
) => 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)');
}
// CLI-invocable live registration, callable from a shell script via
// openclaw gateway call fabric.register --params '{"agentId":"…","apiKey":"fak_…"}'
// The `fabric-register` TOOL only fires inside an agent turn, and there is
// no `openclaw tools call` CLI — so recruitment's `register-agent` script
// (a plain shell step, no LLM turn) had to fall back to the standalone
// binary, which can't notify the running plugin → needed a gateway
// restart. This gateway method runs in-process: inbound.addAccount
// validates the key, persists identity, and brings the socket up live —
// no restart.
api.registerGatewayMethod('fabric.register', async ({ params, respond }) => {
const p = (params ?? {}) as { agentId?: string; apiKey?: string };
if (!p.agentId || !p.apiKey) {
respond(false, { ok: false }, { code: 'INVALID_REQUEST', message: 'agentId and apiKey required' });
return;
}
if (!inbound) {
respond(false, { ok: false }, { code: 'UNAVAILABLE', message: 'fabric inbound not ready (gateway still starting?)' });
return;
}
try {
await inbound.addAccount({ agentId: p.agentId, fabricApiKey: p.apiKey });
respond(true, { ok: true, agentId: p.agentId });
} catch (err) {
respond(false, { ok: false }, { code: 'UNAVAILABLE', message: `fabric-register failed: ${String(err)}` });
}
}, { scope: 'operator.write' });
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;
});
},
});