telegram-user: harden login and monitor lifecycle

extensions/telegram-user/src/login.ts

@@ -46,6 +46,10 @@ export async function loginTelegramUser(params: {
   let phoneEnv = process.env.TELEGRAM_USER_PHONE?.trim() || undefined;
   const codeEnv = process.env.TELEGRAM_USER_CODE?.trim() || undefined;
 
+  const passwordPrompt = passwordEnv
+    ? passwordEnv
+    : async () => await promptText("2FA password: ");
+
   try {
     if (!phoneEnv) {
       const mode = await promptLoginMode();
@@ -53,12 +57,12 @@ export async function loginTelegramUser(params: {
         phoneEnv = await promptText("Telegram phone number (E.164): ");
       }
     }
-    const user = await client.start(
+      const user = await client.start(
       phoneEnv
         ? {
             phone: phoneEnv,
             code: codeEnv ? codeEnv : async () => await promptText("Telegram code: "),
-            password: passwordEnv ? passwordEnv : async () => await promptText("2FA password: "),
+            password: passwordPrompt,
             codeSentCallback: (code) => {
               runtime.log(
                 `Telegram code sent via ${code.type}. Check your device and enter it here.`,
@@ -84,11 +88,13 @@ export async function loginTelegramUser(params: {
               runtime.log(`Scan this QR in Telegram (expires ${expires.toLocaleTimeString()}):`);
               qrcode.generate(url, { small: true });
             },
-            ...(passwordEnv ? { password: passwordEnv } : {}),
+            password: passwordPrompt,
             invalidCodeCallback: async (type) => {
               if (type === "password") {
                 runtime.error?.(
-                  "Telegram 2FA password rejected. Set TELEGRAM_USER_PASSWORD and rerun.",
+                  passwordEnv
+                    ? "Telegram 2FA password rejected. Update TELEGRAM_USER_PASSWORD and rerun."
+                    : "Telegram 2FA password rejected. Try again.",
                 );
               }
             },

extensions/telegram-user/src/monitor/handler.ts

@@ -361,8 +361,9 @@ export function createTelegramUserMessageHandler(params: TelegramUserHandlerPara
       const combinedAllowFrom = [...allowFrom, ...storeAllowFrom];
       const chatId = msg.chat.type === "chat" ? msg.chat.id : undefined;
       const isForum = msg.chat.type === "chat" && msg.chat.isForum === true;
+      const isTopicMessage = msg.isTopicMessage === true;
       const threadId =
-        isGroup && isForum ? msg.replyToMessage?.threadId ?? undefined : undefined;
+        isGroup && isForum && isTopicMessage ? msg.replyToMessage?.threadId ?? undefined : undefined;
       const { groupConfig, topicConfig } =
         isGroup && chatId != null
           ? resolveTelegramUserGroupConfig(accountConfig, chatId, threadId)
@@ -499,7 +500,8 @@ export function createTelegramUserMessageHandler(params: TelegramUserHandlerPara
         ...core.channel.mentions.buildMentionRegexes(cfg, route.agentId),
         ...buildTelegramUserSelfMentionRegexes({ username: self?.username, name: self?.name }),
       ];
-      const hasAnyMention = msg.entities.some(
+      const entities = msg.entities ?? [];
+      const hasAnyMention = entities.some(
         (ent) => ent.kind === "mention" || ent.kind === "text_mention",
       );
       const hasControlCommandInMessage = core.channel.text.hasControlCommand(text, cfg, {

extensions/telegram-user/src/monitor/index.ts

@@ -64,9 +64,11 @@ export async function monitorTelegramUserProvider(opts: MonitorTelegramUserOpts
     );
   }
   const client = await createTelegramUserClient({ apiId, apiHash, storagePath });
-  setActiveTelegramUserClient(account.accountId, client);
+  let stopped = false;
 
   const stop = async () => {
+    if (stopped) return;
+    stopped = true;
     shuttingDown = true;
     setActiveTelegramUserClient(account.accountId, null);
     await client.destroy().catch(() => undefined);
@@ -81,77 +83,80 @@ export async function monitorTelegramUserProvider(opts: MonitorTelegramUserOpts
     { once: true },
   );
 
-  await client.start();
+  try {
+    await client.start();
+    setActiveTelegramUserClient(account.accountId, client);
 
-  const { Dispatcher, filters } = await loadMtcuteDispatcher();
+    const { Dispatcher, filters } = await loadMtcuteDispatcher();
-  const dispatcher = Dispatcher.for(client);
+    const dispatcher = Dispatcher.for(client);
-  const self = await client.getMe().catch(() => undefined);
+    const self = await client.getMe().catch(() => undefined);
-  const selfName =
+    const selfName =
-    self && typeof (self as unknown as { displayName?: unknown }).displayName === "string"
+      self && typeof (self as unknown as { displayName?: unknown }).displayName === "string"
-      ? (self as unknown as { displayName: string }).displayName
+        ? (self as unknown as { displayName: string }).displayName
-      : self && typeof (self as unknown as { firstName?: unknown }).firstName === "string"
+        : self && typeof (self as unknown as { firstName?: unknown }).firstName === "string"
-        ? [
+          ? [
-            (self as unknown as { firstName?: string }).firstName,
+              (self as unknown as { firstName?: string }).firstName,
-            typeof (self as unknown as { lastName?: unknown }).lastName === "string"
+              typeof (self as unknown as { lastName?: unknown }).lastName === "string"
-              ? (self as unknown as { lastName: string }).lastName
+                ? (self as unknown as { lastName: string }).lastName
-              : undefined,
+                : undefined,
-          ]
+            ]
-            .filter((entry): entry is string => Boolean(entry && entry.trim()))
+              .filter((entry): entry is string => Boolean(entry && entry.trim()))
-            .join(" ")
+              .join(" ")
-        : undefined;
+          : undefined;
-  const handleMessage = createTelegramUserMessageHandler({
+    const handleMessage = createTelegramUserMessageHandler({
-    client,
+      client,
-    cfg,
+      cfg,
-    runtime,
+      runtime,
-    accountId: account.accountId,
+      accountId: account.accountId,
-    accountConfig: account.config,
+      accountConfig: account.config,
-    abortSignal: opts.abortSignal,
+      abortSignal: opts.abortSignal,
-    self: self
+      self: self
-      ? {
+        ? {
-          id: self.id,
+            id: self.id,
-          username: "username" in self ? self.username : undefined,
+            username: "username" in self ? self.username : undefined,
-          name: selfName,
+            name: selfName,
+          }
+        : undefined,
+    });
+
+    dispatcher.onNewMessage(
+      filters.or(
+        filters.chat("user"),
+        filters.chat("group"),
+        filters.chat("supergroup"),
+        filters.chat("gigagroup"),
+      ),
+      handleMessage,
+    );
+
+    await new Promise<void>((resolve, reject) => {
+      let settled = false;
+      const settleResolve = () => {
+        if (settled) return;
+        settled = true;
+        resolve();
+      };
+      const settleReject = (err: unknown) => {
+        if (settled) return;
+        settled = true;
+        reject(err);
+      };
+
+      client.onError.add((err) => {
+        if (shuttingDown || opts.abortSignal?.aborted || isDestroyedClientError(err)) {
+          settleResolve();
+          return;
         }
-      : undefined,
+        runtime.error?.(`telegram-user client error: ${String(err)}`);
-  });
+        settleReject(err);
-
+      });
-  dispatcher.onNewMessage(
+      if (opts.abortSignal?.aborted) {
-    filters.or(
-      filters.chat("user"),
-      filters.chat("group"),
-      filters.chat("supergroup"),
-      filters.chat("gigagroup"),
-    ),
-    handleMessage,
-  );
-
-  await new Promise<void>((resolve, reject) => {
-    let settled = false;
-    const settleResolve = () => {
-      if (settled) return;
-      settled = true;
-      resolve();
-    };
-    const settleReject = (err: unknown) => {
-      if (settled) return;
-      settled = true;
-      reject(err);
-    };
-
-    client.onError.add((err) => {
-      if (shuttingDown || opts.abortSignal?.aborted || isDestroyedClientError(err)) {
         settleResolve();
         return;
       }
-      runtime.error?.(`telegram-user client error: ${String(err)}`);
+      opts.abortSignal?.addEventListener("abort", () => settleResolve(), { once: true });
-      settleReject(err);
     });
-    if (opts.abortSignal?.aborted) {
+  } finally {
-      settleResolve();
+    await stop();
-      return;
+  }
-    }
-    opts.abortSignal?.addEventListener("abort", () => settleResolve(), { once: true });
-  });
-
-  await stop();
 }