use accountId

extensions/twitch/index.ts

@@ -4,6 +4,8 @@ import { emptyPluginConfigSchema } from "clawdbot/plugin-sdk";
 import { twitchPlugin } from "./src/plugin.js";
 import { setTwitchRuntime } from "./src/runtime.js";
 
+export { monitorTwitchProvider } from "./src/monitor.js";
+
 const plugin = {
   id: "twitch",
   name: "Twitch",
@@ -11,7 +13,7 @@ const plugin = {
   configSchema: emptyPluginConfigSchema(),
   register(api: ClawdbotPluginApi) {
     setTwitchRuntime(api.runtime);
-    api.registerChannel({ plugin: twitchPlugin });
+    api.registerChannel({ plugin: twitchPlugin as any });
   },
 };
 

extensions/twitch/src/monitor.ts

@@ -0,0 +1,251 @@
+/**
+ * Twitch message monitor - processes incoming messages and routes to agents.
+ *
+ * This monitor connects to the Twitch client manager, processes incoming messages,
+ * resolves agent routes, and handles replies.
+ */
+
+import type { ReplyPayload } from "clawdbot/plugin-sdk";
+import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js";
+import { checkTwitchAccessControl } from "./access-control.js";
+import { getTwitchRuntime } from "./runtime.js";
+import { getOrCreateClientManager } from "./client-manager-registry.js";
+
+export type TwitchRuntimeEnv = {
+  log?: (message: string) => void;
+  error?: (message: string) => void;
+};
+
+export type TwitchMonitorOptions = {
+  account: TwitchAccountConfig;
+  accountId: string;
+  config: unknown; // ClawdbotConfig
+  runtime: TwitchRuntimeEnv;
+  abortSignal: AbortSignal;
+  statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
+};
+
+export type TwitchMonitorResult = {
+  stop: () => void;
+};
+
+type TwitchCoreRuntime = ReturnType<typeof getTwitchRuntime>;
+
+/**
+ * Process an incoming Twitch message and dispatch to agent.
+ */
+async function processTwitchMessage(params: {
+  message: TwitchChatMessage;
+  account: TwitchAccountConfig;
+  accountId: string;
+  config: unknown;
+  runtime: TwitchRuntimeEnv;
+  core: TwitchCoreRuntime;
+  statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
+}): Promise<void> {
+  const { message, account, accountId, config, runtime, core, statusSink } = params;
+
+  // Resolve route for this message
+  const route = core.channel.routing.resolveAgentRoute({
+    cfg: config as Parameters<typeof core.channel.routing.resolveAgentRoute>[0]["cfg"],
+    channel: "twitch",
+    accountId,
+    peer: {
+      kind: "group", // Twitch chat is always group-like
+      id: message.channel,
+    },
+  });
+
+  // Build message body
+  const rawBody = message.message;
+  const body = core.channel.reply.formatAgentEnvelope({
+    channel: "Twitch",
+    from: message.displayName ?? message.username,
+    timestamp: message.timestamp?.getTime(),
+    envelope: core.channel.reply.resolveEnvelopeFormatOptions(config as Parameters<typeof core.channel.reply.resolveEnvelopeFormatOptions>[0]),
+    body: rawBody,
+  });
+
+  // Build context payload
+  const ctxPayload = core.channel.reply.finalizeInboundContext({
+    Body: body,
+    RawBody: rawBody,
+    CommandBody: rawBody,
+    From: `twitch:user:${message.userId}`,
+    To: `twitch:channel:${message.channel}`,
+    SessionKey: route.sessionKey,
+    AccountId: route.accountId,
+    ChatType: "group",
+    ConversationLabel: message.channel,
+    SenderName: message.displayName ?? message.username,
+    SenderId: message.userId,
+    SenderUsername: message.username,
+    Provider: "twitch",
+    Surface: "twitch",
+    MessageSid: message.id,
+    OriginatingChannel: "twitch",
+    OriginatingTo: `twitch:channel:${message.channel}`,
+  });
+
+  // Record session
+  const storePath = core.channel.session.resolveStorePath(
+    (config as Parameters<typeof core.channel.session.resolveStorePath>[0]["cfg"])?.session?.store,
+    { agentId: route.agentId },
+  );
+  await core.channel.session.recordInboundSession({
+    storePath,
+    sessionKey: ctxPayload.SessionKey ?? route.sessionKey,
+    ctx: ctxPayload,
+    onRecordError: (err) => {
+      runtime.error?.(`[twitch] Failed updating session meta: ${String(err)}`);
+    },
+  });
+
+  // Resolve markdown table mode
+  const tableMode = core.channel.text.resolveMarkdownTableMode({
+    cfg: config as Parameters<typeof core.channel.text.resolveMarkdownTableMode>[0]["cfg"],
+    channel: "twitch",
+    accountId,
+  });
+
+  // Dispatch reply
+  await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({
+    ctx: ctxPayload,
+    cfg: config as Parameters<typeof core.channel.reply.dispatchReplyWithBufferedBlockDispatcher>[0]["cfg"],
+    dispatcherOptions: {
+      deliver: async (payload) => {
+        await deliverTwitchReply({
+          payload,
+          channel: message.channel,
+          account,
+          runtime,
+          statusSink,
+        });
+      },
+    },
+    runtime,
+    tableMode,
+  });
+}
+
+/**
+ * Deliver a reply to Twitch chat.
+ */
+async function deliverTwitchReply(params: {
+  payload: ReplyPayload;
+  channel: string;
+  account: TwitchAccountConfig;
+  runtime: TwitchRuntimeEnv;
+  statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
+}): Promise<void> {
+  const { payload, channel, account, runtime, statusSink } = params;
+
+  try {
+    const clientManager = getOrCreateClientManager(account.accountId ?? "default", {
+      info: (msg) => runtime.log?.(`[twitch] ${msg}`),
+      warn: (msg) => runtime.log?.(`[twitch] ${msg}`),
+      error: (msg) => runtime.error?.(`[twitch] ${msg}`),
+      debug: (msg) => runtime.log?.(`[twitch] ${msg}`),
+    });
+
+    const client = await clientManager.getClient(account, null, account.accountId ?? "default");
+    if (!client) {
+      runtime.error?.(`[twitch] No client available for sending reply`);
+      return;
+    }
+
+    // Send the reply
+    await client.say(channel, payload.text);
+    statusSink?.({ lastOutboundAt: Date.now() });
+  } catch (err) {
+    runtime.error?.(`[twitch] Failed to send reply: ${String(err)}`);
+  }
+}
+
+/**
+ * Main monitor provider for Twitch.
+ *
+ * Sets up message handlers and processes incoming messages.
+ */
+export async function monitorTwitchProvider(
+  options: TwitchMonitorOptions,
+): Promise<TwitchMonitorResult> {
+  const { account, accountId, config, runtime, abortSignal, statusSink } = options;
+
+  const core = getTwitchRuntime();
+  let stopped = false;
+
+  // Create logger for client manager
+  const logger = {
+    info: (msg: string) => runtime.log?.(`[twitch] ${msg}`),
+    warn: (msg: string) => runtime.log?.(`[twitch] ${msg}`),
+    error: (msg: string) => runtime.error?.(`[twitch] ${msg}`),
+    debug: (msg: string) => runtime.log?.(`[twitch] ${msg}`),
+  };
+
+  // Get or create client manager
+  const clientManager = getOrCreateClientManager(accountId, logger);
+
+  // Establish connection
+  try {
+    await clientManager.getClient(account, config as Parameters<typeof clientManager.getClient>[1], accountId);
+    runtime.log?.(`[twitch] Connected to Twitch as ${account.username}`);
+  } catch (error) {
+    const errorMsg = error instanceof Error ? error.message : String(error);
+    runtime.error?.(`[twitch] Failed to connect: ${errorMsg}`);
+    throw error;
+  }
+
+  // Register message handler
+  const unregisterHandler = clientManager.onMessage(account, async (message) => {
+    if (stopped) return;
+
+    // Access control check
+    const botUsername = account.username.toLowerCase();
+    if (message.username.toLowerCase() === botUsername) {
+      return; // Ignore own messages
+    }
+
+    const access = checkTwitchAccessControl({
+      message,
+      account,
+      botUsername,
+    });
+
+    if (!access.allowed) {
+      runtime.log?.(
+        `[twitch] Ignored message from ${message.username}: ${access.reason ?? "blocked"}`,
+      );
+      return;
+    }
+
+    statusSink?.({ lastInboundAt: Date.now() });
+
+    try {
+      await processTwitchMessage({
+        message,
+        account,
+        accountId,
+        config,
+        runtime,
+        core,
+        statusSink,
+      });
+    } catch (err) {
+      runtime.error?.(`[twitch] Message processing failed: ${String(err)}`);
+    }
+  });
+
+  // Stop function
+  const stop = () => {
+    stopped = true;
+    unregisterHandler();
+  };
+
+  // Handle abort signal
+  abortSignal.addEventListener("abort", stop, { once: true });
+
+  runtime.log?.(`[twitch] Monitor started for account ${accountId}`);
+
+  return { stop };
+}

extensions/twitch/src/plugin.ts

@@ -5,8 +5,6 @@
  * This is the primary entry point for the Twitch channel integration.
  */
 
-import { checkTwitchAccessControl } from "./access-control.js";
-import { resolveAgentRoute } from "../../../src/routing/resolve-route.js";
 import type { ClawdbotConfig } from "clawdbot/plugin-sdk";
 import { buildChannelConfigSchema } from "clawdbot/plugin-sdk";
 import { twitchMessageActions } from "./actions.js";
@@ -21,10 +19,7 @@ import { twitchOutbound } from "./outbound.js";
 import { probeTwitch } from "./probe.js";
 import { resolveTwitchTargets } from "./resolver.js";
 import { collectTwitchStatusIssues } from "./status.js";
-import {
+import { removeClientManager } from "./client-manager-registry.js";
-  getOrCreateClientManager,
-  removeClientManager,
-} from "./client-manager-registry.js";
 import type {
   ChannelAccountSnapshot,
   ChannelCapabilities,
@@ -245,61 +240,6 @@ export const twitchPlugin: ChannelPlugin<TwitchAccountConfig> = {
       const account = ctx.account as TwitchAccountConfig;
       const accountId = ctx.accountId;
 
-      // Create logger for client manager
-      const logger: ChannelLogSink = {
-        info: (msg) => ctx.log?.info(`[twitch] ${msg}`),
-        warn: (msg) => ctx.log?.warn(`[twitch] ${msg}`),
-        error: (msg) => ctx.log?.error(`[twitch] ${msg}`),
-        debug: (msg) => ctx.log?.debug?.(`[twitch] ${msg}`),
-      };
-
-      // Get or create client manager from registry
-      const clientManager = getOrCreateClientManager(accountId, logger);
-
-      // Set up message handler
-      clientManager.onMessage(account, (message) => {
-        // Access control check
-        const botUsername = account.username.toLowerCase();
-        if (message.username.toLowerCase() === botUsername) {
-          return; // Ignore own messages
-        }
-
-        const access = checkTwitchAccessControl({
-          message,
-          account,
-          botUsername,
-        });
-
-        if (!access.allowed) {
-          ctx.log?.info(
-            `[twitch] Ignored message from ${message.username}: ${access.reason ?? "blocked"}`,
-          );
-          return;
-        }
-
-        // Resolve route for this message
-        resolveAgentRoute({
-          cfg: ctx.cfg,
-          channel: "twitch",
-          accountId,
-          peer: {
-            kind: "group", // Twitch chat is always group-like
-            id: message.channel,
-          },
-        });
-
-        // Build message preview
-        const preview = message.message.replace(/\s+/g, " ").slice(0, 160);
-
-        // Log message receipt
-        // Note: Message dispatch to agent system will be handled by a separate monitor/loop
-        // For now, we just log that we received and validated the message
-        ctx.log?.info(
-          `[twitch] Received message from ${message.displayName ?? message.username}: ${preview}`,
-        );
-      });
-
-      // Update status
       ctx.setStatus?.({
         accountId,
         running: true,
@@ -311,19 +251,15 @@ export const twitchPlugin: ChannelPlugin<TwitchAccountConfig> = {
         `[twitch] Starting Twitch connection for ${account.username}`,
       );
 
-      try {
+      // Lazy import: the monitor pulls the reply pipeline; avoid ESM init cycles.
-        await clientManager.getClient(account, ctx.cfg);
+      const { monitorTwitchProvider } = await import("./monitor.js");
-        ctx.log?.info(`[twitch] Connected to Twitch as ${account.username}`);
+      await monitorTwitchProvider({
-      } catch (error) {
+        account,
-        const errorMsg = error instanceof Error ? error.message : String(error);
+        accountId,
-        ctx.log?.error(`[twitch] Failed to connect: ${errorMsg}`);
+        config: ctx.cfg,
-        ctx.setStatus?.({
+        runtime: ctx.runtime,
-          accountId,
+        abortSignal: ctx.abortSignal,
-          running: false,
+      });
-          lastError: errorMsg,
-        });
-        throw error;
-      }
     },
 
     /** Stop an account connection */

extensions/twitch/src/send.test.ts

@@ -285,6 +285,7 @@ describe("send", () => {
 				"testchannel", // normalized account channel
 				"Hello!",
 				mockConfig,
+				"default",
 			);
 		});
 	});

extensions/twitch/src/send.ts

@@ -113,6 +113,7 @@ export async function sendMessageTwitchInternal(
 			normalizeTwitchChannel(normalizedChannel),
 			cleanedText,
 			cfg,
+			accountId,
 		);
 
 		if (!result.ok) {

extensions/twitch/src/twitch-client.ts

@@ -96,6 +96,7 @@ export class TwitchClientManager {
 	async getClient(
 		account: TwitchAccountConfig,
 		cfg?: ClawdbotConfig,
+		accountId?: string,
 	): Promise<ChatClient> {
 		const key = this.getAccountKey(account);
 
@@ -106,7 +107,7 @@ export class TwitchClientManager {
 
 		// Resolve token from config or environment
 		const tokenResolution = resolveTwitchToken(cfg, {
-			accountId: account.username,
+			accountId,
 		});
 
 		if (!tokenResolution.token) {
@@ -283,9 +284,10 @@ export class TwitchClientManager {
 		channel: string,
 		message: string,
 		cfg?: ClawdbotConfig,
+		accountId?: string,
 	): Promise<{ ok: boolean; error?: string; messageId?: string }> {
 		try {
-			const client = await this.getClient(account, cfg);
+			const client = await this.getClient(account, cfg, accountId);
 
 			// Generate a message ID (Twurple's say() doesn't return the message ID, so we generate one)
 			const messageId = `${Date.now()}-${Math.random().toString(36).substring(2, 15)}`;

tsconfig.json

@@ -5,7 +5,6 @@
     "moduleResolution": "NodeNext",
     "outDir": "dist",
     "rootDir": "src",
-    "declaration": true,
     "strict": true,
     "esModuleInterop": true,
     "forceConsistentCasingInFileNames": true,