fix(telegram): close evicted client-options cache transports after in-flight sends (#101783)

Long-running Telegram senders leaked HTTP transports: the client-options cache in extensions/telegram/src/send.ts created an owned undici dispatcher per account/network entry but evicted entries without closing it. Cache entries now own their transport; eviction retires the entry and closes the transport once the operation-level lease releases, so cache pressure never aborts an in-flight, retrying, or still-preparing send. Includes a real-socket e2e regression against a local Bot API server.

Co-authored-by: zhangguiping-xydt <zhangguiping-xydt@users.noreply.github.com>
Co-authored-by: Ayaan Zaidi <hi@obviy.us>

extensions/telegram/src/send.proxy.test.ts

@@ -14,7 +14,9 @@ const { botApi, botCtorSpy } = vi.hoisted(() => ({
     };
     return {
       config: { use: vi.fn() },
+      getChat: vi.fn(),
       sendMessage,
+      sendDocument: vi.fn(),
       setMessageReaction: vi.fn(),
       deleteMessage: vi.fn(),
       raw: {
@@ -45,6 +47,14 @@ const { resolveTelegramTransport } = vi.hoisted(() => ({
   resolveTelegramTransport: vi.fn(),
 }));
 
+const { loadWebMedia } = vi.hoisted(() => ({
+  loadWebMedia: vi.fn(),
+}));
+
+const { maybePersistResolvedTelegramTarget } = vi.hoisted(() => ({
+  maybePersistResolvedTelegramTarget: vi.fn(),
+}));
+
 const resolveTelegramApiBase = vi.hoisted(
   () => (apiRoot?: string) => apiRoot?.trim()?.replace(/\/+$/, "") || "https://api.telegram.org",
 );
@@ -68,6 +78,18 @@ vi.mock("./fetch.js", () => ({
   resolveTelegramApiBase,
 }));
 
+vi.mock("./send.runtime.js", async () => {
+  const actual = await vi.importActual<typeof import("./send.runtime.js")>("./send.runtime.js");
+  return {
+    ...actual,
+    loadWebMedia,
+  };
+});
+
+vi.mock("./target-writeback.js", () => ({
+  maybePersistResolvedTelegramTarget,
+}));
+
 vi.mock("grammy", () => ({
   API_CONSTANTS: {
     DEFAULT_UPDATE_TYPES: ["message"],
@@ -134,9 +156,14 @@ describe("telegram proxy client", () => {
   beforeEach(() => {
     resetTelegramClientOptionsCacheForTests();
     vi.unstubAllEnvs();
+    botApi.getChat.mockResolvedValue({ id: "123" });
     botApi.sendMessage.mockResolvedValue({ message_id: 1, chat: { id: "123" } });
+    botApi.sendDocument.mockResolvedValue({ message_id: 1, chat: { id: "123" } });
     botApi.setMessageReaction.mockResolvedValue(undefined);
     botApi.deleteMessage.mockResolvedValue(true);
+    loadWebMedia.mockReset();
+    maybePersistResolvedTelegramTarget.mockReset();
+    maybePersistResolvedTelegramTarget.mockResolvedValue(undefined);
     botApi.config.use.mockClear();
     botCtorSpy.mockClear();
     loadConfig.mockReturnValue(TELEGRAM_PROXY_CFG);
@@ -169,6 +196,301 @@ describe("telegram proxy client", () => {
     expect(botCtorSpy).toHaveBeenNthCalledWith(2, "tok", firstOptions);
   });
 
+  it("closes the evicted Telegram transport when the client options cache exceeds its limit", async () => {
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+    const closeSpies: Array<ReturnType<typeof vi.fn>> = [];
+    makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch);
+    resolveTelegramTransport.mockImplementation(() => {
+      const fetchImpl = vi.fn() as unknown as typeof fetch;
+      const close = vi.fn(async () => undefined);
+      closeSpies.push(close);
+      return {
+        fetch: fetchImpl,
+        sourceFetch: fetchImpl,
+        close,
+      };
+    });
+    const cfg = {
+      channels: {
+        telegram: {
+          accounts: Object.fromEntries(
+            Array.from({ length: 65 }, (_unused, index) => [
+              `acct-${index}`,
+              { proxy: `http://proxy-${index}.test:8080` },
+            ]),
+          ),
+        },
+      },
+    };
+
+    for (let index = 0; index < 65; index += 1) {
+      await sendMessageTelegram("123", `message ${index}`, {
+        cfg,
+        token: "tok",
+        accountId: `acct-${index}`,
+      });
+    }
+
+    expect(resolveTelegramTransport).toHaveBeenCalledTimes(65);
+    expect(closeSpies[0]).toHaveBeenCalledTimes(1);
+    expect(closeSpies.slice(1).every((close) => close.mock.calls.length === 0)).toBe(true);
+  });
+
+  it.each([
+    {
+      name: "an active send finishes",
+      setupFirstSend: () => {
+        let resolveFirstSend: (value: {
+          message_id: number;
+          chat: { id: string };
+        }) => void = () => {};
+        botApi.sendMessage.mockImplementationOnce(
+          () =>
+            new Promise((resolve) => {
+              resolveFirstSend = resolve;
+            }),
+        );
+        return {
+          waitUntilActive: async () => {
+            await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(1));
+          },
+          finish: async () => {
+            resolveFirstSend({ message_id: 1, chat: { id: "123" } });
+          },
+        };
+      },
+    },
+    {
+      name: "a retryable send is waiting between attempts",
+      setupFirstSend: () => {
+        vi.useFakeTimers();
+        botApi.sendMessage
+          .mockRejectedValueOnce({
+            error_code: 429,
+            description: "Too Many Requests: retry after 1",
+            parameters: { retry_after: 1 },
+          })
+          .mockResolvedValueOnce({ message_id: 1, chat: { id: "123" } });
+        return {
+          waitUntilActive: async () => {
+            await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(1));
+          },
+          finish: async () => {
+            await vi.advanceTimersByTimeAsync(1000);
+            await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(66));
+            vi.useRealTimers();
+          },
+        };
+      },
+    },
+  ])("defers closing an evicted Telegram transport until $name", async ({ setupFirstSend }) => {
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+    const closeSpies: Array<ReturnType<typeof vi.fn>> = [];
+    makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch);
+    resolveTelegramTransport.mockImplementation(() => {
+      const fetchImpl = vi.fn() as unknown as typeof fetch;
+      const close = vi.fn(async () => undefined);
+      closeSpies.push(close);
+      return {
+        fetch: fetchImpl,
+        sourceFetch: fetchImpl,
+        close,
+      };
+    });
+    const cfg = {
+      channels: {
+        telegram: {
+          accounts: Object.fromEntries(
+            Array.from({ length: 65 }, (_unused, index) => [
+              `acct-${index}`,
+              { proxy: `http://proxy-${index}.test:8080` },
+            ]),
+          ),
+        },
+      },
+    };
+    botApi.sendMessage.mockClear();
+    const firstSendControl = setupFirstSend();
+
+    const firstSend = sendMessageTelegram("123", "first", {
+      cfg,
+      token: "tok",
+      accountId: "acct-0",
+      retry: { attempts: 2, minDelayMs: 1000, maxDelayMs: 1000, jitter: 0 },
+    });
+    await firstSendControl.waitUntilActive();
+
+    await Promise.all(
+      Array.from({ length: 64 }, (_unused, index) =>
+        sendMessageTelegram("123", `message ${index + 1}`, {
+          cfg,
+          token: "tok",
+          accountId: `acct-${index + 1}`,
+        }),
+      ),
+    );
+
+    expect(resolveTelegramTransport).toHaveBeenCalledTimes(65);
+    expect(closeSpies[0]).not.toHaveBeenCalled();
+
+    await firstSendControl.finish();
+    await firstSend;
+
+    expect(closeSpies[0]).toHaveBeenCalledTimes(1);
+    vi.useRealTimers();
+  });
+
+  it("defers closing an evicted Telegram transport while media loads before the first API request", async () => {
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+    const closeSpies: Array<ReturnType<typeof vi.fn>> = [];
+    makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch);
+    resolveTelegramTransport.mockImplementation(() => {
+      const fetchImpl = vi.fn() as unknown as typeof fetch;
+      const close = vi.fn(async () => undefined);
+      closeSpies.push(close);
+      return {
+        fetch: fetchImpl,
+        sourceFetch: fetchImpl,
+        close,
+      };
+    });
+    const cfg = {
+      channels: {
+        telegram: {
+          accounts: Object.fromEntries(
+            Array.from({ length: 65 }, (_unused, index) => [
+              `acct-${index}`,
+              { proxy: `http://proxy-${index}.test:8080` },
+            ]),
+          ),
+        },
+      },
+    };
+    let resolveMedia: (value: {
+      buffer: Buffer;
+      contentType: string;
+      fileName: string;
+    }) => void = () => {};
+    loadWebMedia.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolveMedia = resolve;
+        }),
+    );
+
+    const firstSend = sendMessageTelegram("123", "caption", {
+      cfg,
+      token: "tok",
+      accountId: "acct-0",
+      mediaUrl: "file:///tmp/example.bin",
+      forceDocument: true,
+    });
+    await vi.waitFor(() => expect(loadWebMedia).toHaveBeenCalledTimes(1));
+
+    await Promise.all(
+      Array.from({ length: 64 }, (_unused, index) =>
+        sendMessageTelegram("123", `message ${index + 1}`, {
+          cfg,
+          token: "tok",
+          accountId: `acct-${index + 1}`,
+        }),
+      ),
+    );
+
+    expect(resolveTelegramTransport).toHaveBeenCalledTimes(65);
+    expect(closeSpies[0]).not.toHaveBeenCalled();
+
+    resolveMedia({
+      buffer: Buffer.from("file"),
+      contentType: "application/octet-stream",
+      fileName: "example.bin",
+    });
+    await firstSend;
+
+    expect(botApi.sendDocument).toHaveBeenCalledTimes(1);
+    expect(closeSpies[0]).toHaveBeenCalledTimes(1);
+  });
+
+  it("defers closing an evicted Telegram transport while reactions persist a resolved target before the first action request", async () => {
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+    const closeSpies: Array<ReturnType<typeof vi.fn>> = [];
+    makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch);
+    resolveTelegramTransport.mockImplementation(() => {
+      const fetchImpl = vi.fn() as unknown as typeof fetch;
+      const close = vi.fn(async () => undefined);
+      closeSpies.push(close);
+      return {
+        fetch: fetchImpl,
+        sourceFetch: fetchImpl,
+        close,
+      };
+    });
+    const cfg = {
+      channels: {
+        telegram: {
+          accounts: Object.fromEntries(
+            Array.from({ length: 65 }, (_unused, index) => [
+              `acct-${index}`,
+              { proxy: `http://proxy-${index}.test:8080` },
+            ]),
+          ),
+        },
+      },
+    };
+    let resolvePersist: () => void = () => {};
+    maybePersistResolvedTelegramTarget.mockImplementationOnce(
+      () =>
+        new Promise<void>((resolve) => {
+          resolvePersist = resolve;
+        }),
+    );
+
+    const firstReaction = reactMessageTelegram("@target", "456", "✅", {
+      cfg,
+      token: "tok",
+      accountId: "acct-0",
+    });
+    await vi.waitFor(() => expect(maybePersistResolvedTelegramTarget).toHaveBeenCalledTimes(1));
+
+    await Promise.all(
+      Array.from({ length: 64 }, (_unused, index) =>
+        reactMessageTelegram("123", "456", "✅", {
+          cfg,
+          token: "tok",
+          accountId: `acct-${index + 1}`,
+        }),
+      ),
+    );
+
+    expect(resolveTelegramTransport).toHaveBeenCalledTimes(65);
+    expect(closeSpies[0]).not.toHaveBeenCalled();
+
+    resolvePersist();
+    await firstReaction;
+
+    expect(botApi.setMessageReaction).toHaveBeenCalledTimes(65);
+    expect(closeSpies[0]).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not allocate cached client transport when a Telegram API override is provided", async () => {
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+
+    await reactMessageTelegram("123", "456", "✅", {
+      cfg: TELEGRAM_PROXY_CFG,
+      token: "tok",
+      accountId: "foo",
+      api: botApi as unknown as Parameters<typeof reactMessageTelegram>[3]["api"],
+    });
+
+    expect(makeProxyFetch).not.toHaveBeenCalled();
+    expect(resolveTelegramTransport).not.toHaveBeenCalled();
+  });
+
   it.each([
     {
       name: "sendMessage",

extensions/telegram/src/send.test.ts

@@ -54,6 +54,7 @@ const {
   createForumTopicTelegram,
   deleteMessageTelegram,
   editForumTopicTelegram,
+  editMessageReplyMarkupTelegram,
   editMessageTelegram,
   pinMessageTelegram,
   reactMessageTelegram,
@@ -183,6 +184,69 @@ function installTelegramStateRuntimeForTest(): void {
   } as TelegramRuntime);
 }
 
+describe("Telegram send Promise contract", () => {
+  const contextFailureCalls: ReadonlyArray<readonly [string, () => Promise<unknown>]> = [
+    [
+      "sendMessageTelegram",
+      () => sendMessageTelegramImported("123", "hello", { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    ["sendTypingTelegram", () => sendTypingTelegram("123", { cfg: TELEGRAM_TEST_CFG })],
+    [
+      "reactMessageTelegram",
+      () => reactMessageTelegram("123", 1, "👍", { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    ["deleteMessageTelegram", () => deleteMessageTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG })],
+    ["pinMessageTelegram", () => pinMessageTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG })],
+    [
+      "unpinMessageTelegram",
+      () => unpinMessageTelegram("123", undefined, { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    [
+      "editForumTopicTelegram",
+      () => editForumTopicTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG, name: "topic" }),
+    ],
+    [
+      "editMessageReplyMarkupTelegram",
+      () => editMessageReplyMarkupTelegram("123", 1, [], { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    [
+      "editMessageTelegram",
+      () => editMessageTelegram("123", 1, "hello", { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    [
+      "sendStickerTelegram",
+      () => sendStickerTelegram("123", "file-id", { cfg: TELEGRAM_TEST_CFG }),
+    ],
+    [
+      "sendPollTelegram",
+      () =>
+        sendPollTelegram(
+          "123",
+          { question: "Question?", options: ["A", "B"] },
+          { cfg: TELEGRAM_TEST_CFG },
+        ),
+    ],
+    [
+      "createForumTopicTelegram",
+      () => createForumTopicTelegram("123", "topic", { cfg: TELEGRAM_TEST_CFG }),
+    ],
+  ];
+
+  it.each(contextFailureCalls)(
+    "%s reports context failures as Promise rejections",
+    async (_name, invoke) => {
+      let operation: Promise<unknown> | undefined;
+      expect(() => {
+        operation = invoke();
+      }).not.toThrow();
+      if (!operation) {
+        throw new Error("expected Telegram operation promise");
+      }
+      await expect(operation).rejects.toThrow(/Telegram bot token missing/i);
+    },
+  );
+});
+
 async function expectChatNotFoundWithChatId(
   action: Promise<unknown>,
   expectedChatId: string,
@@ -784,20 +848,25 @@ describe("sendMessageTelegram", () => {
     });
   });
 
-  it("rejects empty topic edits", async () => {
+  it("rejects empty topic edits before creating a Telegram client", async () => {
+    botCtorSpy.mockClear();
+
     await expect(
       editForumTopicTelegram("-1001234567890", 271, {
         cfg: TELEGRAM_TEST_CFG,
+        token: "tok",
         accountId: "default",
       }),
     ).rejects.toThrow("Telegram forum topic update requires a name or iconCustomEmojiId");
     await expect(
       editForumTopicTelegram("-1001234567890", 271, {
         cfg: TELEGRAM_TEST_CFG,
+        token: "tok",
         accountId: "default",
         iconCustomEmojiId: "   ",
       }),
     ).rejects.toThrow("Telegram forum topic icon custom emoji ID is required");
+    expect(botCtorSpy).not.toHaveBeenCalled();
   });
 
   it("applies timeoutSeconds config precedence", async () => {
@@ -3724,12 +3793,15 @@ describe("sendStickerTelegram", () => {
     });
   }
 
-  it("throws error when fileId is blank", async () => {
+  it("rejects a blank fileId before creating a Telegram client", async () => {
+    botCtorSpy.mockClear();
+
     for (const fileId of ["", "   "]) {
       await expect(
         sendStickerTelegram("123", fileId, { cfg: TELEGRAM_TEST_CFG, token: "tok" }),
       ).rejects.toThrow(/file_id is required/i);
     }
+    expect(botCtorSpy).not.toHaveBeenCalled();
   });
 
   it("fails sticker sends instead of retrying without message_thread_id", async () => {
@@ -4507,4 +4579,16 @@ describe("createForumTopicTelegram", () => {
       expect(result).toEqual(testCase.expectedResult);
     });
   }
+
+  it("rejects an invalid topic name before creating a Telegram client", async () => {
+    botCtorSpy.mockClear();
+
+    await expect(
+      createForumTopicTelegram("-1001234567890", "   ", {
+        cfg: TELEGRAM_TEST_CFG,
+        token: "tok",
+      }),
+    ).rejects.toThrow("Forum topic name is required");
+    expect(botCtorSpy).not.toHaveBeenCalled();
+  });
 });

extensions/telegram/src/send.transport-close-proof.test.ts

@@ -0,0 +1,139 @@
+// E2E proof for the transport cache-eviction lifecycle: no module mocks — real
+// grammY Bot, real undici agents, production-mode cache, against a local HTTP
+// server standing in for the Telegram Bot API. Observes actual TCP sockets.
+import { createServer, type Server } from "node:http";
+import type { AddressInfo } from "node:net";
+import type { Socket } from "node:net";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+let sendMessageTelegram: typeof import("./send.js").sendMessageTelegram;
+let resetTelegramClientOptionsCacheForTests: typeof import("./send.js").resetTelegramClientOptionsCacheForTests;
+
+describe("telegram transport cache eviction over real sockets", () => {
+  let server: Server;
+  let apiRoot: string;
+  const liveSockets = new Set<Socket>();
+  const sockets = { opened: 0, closed: 0 };
+  let sendMessageCalls = 0;
+  let slowMode = false;
+  let slowRequestReceived: () => void = () => {};
+
+  beforeAll(async () => {
+    server = createServer((req, res) => {
+      let body = "";
+      req.on("data", (chunk: Buffer) => {
+        body += chunk;
+      });
+      req.on("end", () => {
+        const url = req.url ?? "";
+        const respond = (result: unknown) => {
+          res.setHeader("content-type", "application/json");
+          res.end(JSON.stringify({ ok: true, result }));
+        };
+        if (url.includes("/sendMessage")) {
+          sendMessageCalls += 1;
+          if (slowMode) {
+            slowRequestReceived();
+            setTimeout(() => respond({ message_id: sendMessageCalls, chat: { id: 123 } }), 800);
+            return;
+          }
+          respond({ message_id: sendMessageCalls, chat: { id: 123 } });
+          return;
+        }
+        if (url.includes("/getChat")) {
+          respond({ id: 123, type: "private" });
+          return;
+        }
+        respond(true);
+      });
+    });
+    server.on("connection", (socket) => {
+      sockets.opened += 1;
+      liveSockets.add(socket);
+      socket.on("close", () => {
+        sockets.closed += 1;
+        liveSockets.delete(socket);
+      });
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    apiRoot = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+    ({ sendMessageTelegram, resetTelegramClientOptionsCacheForTests } = await import("./send.js"));
+  });
+
+  afterAll(async () => {
+    resetTelegramClientOptionsCacheForTests();
+    vi.unstubAllEnvs();
+    for (const socket of liveSockets) {
+      socket.destroy();
+    }
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  });
+
+  it("closes evicted transports, deferring close for an in-flight send", async () => {
+    // The cache is disabled under test env; force the production path.
+    vi.stubEnv("VITEST", "");
+    vi.stubEnv("NODE_ENV", "production");
+    resetTelegramClientOptionsCacheForTests();
+
+    const ACCOUNTS = 70;
+    const cfg = {
+      channels: {
+        telegram: {
+          accounts: Object.fromEntries(
+            Array.from({ length: ACCOUNTS }, (_, i) => [
+              `acct-${i}`,
+              { botToken: `10${i}:e2e-token-${i}`, apiRoot },
+            ]),
+          ),
+        },
+      },
+    };
+
+    // Fill the cache to its 64-entry cap: one real agent + socket per account.
+    for (let i = 0; i < 64; i += 1) {
+      const result = await sendMessageTelegram("123", `hello ${i}`, {
+        cfg,
+        accountId: `acct-${i}`,
+      });
+      expect(result.messageId).toBeTruthy();
+    }
+    expect(sockets.opened).toBe(64);
+    // Keep-alive is 30s; nothing may have closed yet.
+    expect(sockets.closed).toBe(0);
+
+    // Put acct-0 (the oldest cache entry) mid-flight, then evict it.
+    slowMode = true;
+    const inFlight = new Promise<void>((resolve) => {
+      slowRequestReceived = resolve;
+    });
+    const slowSend = sendMessageTelegram("123", "slow", { cfg, accountId: "acct-0" });
+    await inFlight;
+    slowMode = false;
+
+    // New cache key -> evicts acct-0 while its send holds the lease.
+    const evictor = await sendMessageTelegram("123", "evictor", { cfg, accountId: "acct-64" });
+    expect(evictor.messageId).toBeTruthy();
+    // Deferred close: the evicted transport must NOT be closed mid-request.
+    expect(sockets.closed).toBe(0);
+
+    const slow = await slowSend;
+    expect(slow.messageId).toBeTruthy();
+    // Lease released -> the retired acct-0 transport closes its real socket.
+    await vi.waitFor(() => expect(sockets.closed).toBeGreaterThanOrEqual(1), { timeout: 3000 });
+
+    // Five more evictions against idle entries (acct-1..acct-5) close immediately.
+    for (let i = 65; i < ACCOUNTS; i += 1) {
+      const result = await sendMessageTelegram("123", `hello ${i}`, {
+        cfg,
+        accountId: `acct-${i}`,
+      });
+      expect(result.messageId).toBeTruthy();
+    }
+    await vi.waitFor(() => expect(sockets.closed).toBe(6), { timeout: 3000 });
+
+    // All sends succeeded; retained cache entries keep their sockets open.
+    expect(sendMessageCalls).toBe(ACCOUNTS + 1);
+    expect(sockets.opened).toBe(ACCOUNTS);
+    expect(liveSockets.size).toBe(ACCOUNTS - 6);
+  });
+});

extensions/telegram/src/send.ts

@@ -26,7 +26,7 @@ import { buildTypingThreadParams } from "./bot/helpers.js";
 import type { TelegramInlineButtons } from "./button-types.js";
 import { splitTelegramCaption } from "./caption.js";
 import { asTelegramClientFetch, createTelegramClientFetch } from "./client-fetch.js";
-import { resolveTelegramTransport } from "./fetch.js";
+import { resolveTelegramTransport, type TelegramTransport } from "./fetch.js";
 import {
   renderTelegramHtmlText,
   splitTelegramHtmlChunks,
@@ -306,7 +306,21 @@ const MESSAGE_DELETE_NOOP_RE =
 const CHAT_NOT_FOUND_RE = /400: Bad Request: chat not found/i;
 const sendLogger = createSubsystemLogger("telegram/send");
 const diagLogger = createSubsystemLogger("telegram/diagnostic");
-const telegramClientOptionsCache = new Map<string, ApiClientOptions | undefined>();
+type CachedTelegramClientOptions = {
+  activeLeases: number;
+  clientOptions: ApiClientOptions | undefined;
+  closeStarted: boolean;
+  retired: boolean;
+  transport: TelegramTransport;
+};
+type TelegramClientOptionsLease = {
+  release: () => void;
+};
+type ResolvedTelegramClientOptions = {
+  clientOptions: ApiClientOptions | undefined;
+  lease?: () => TelegramClientOptionsLease;
+};
+const telegramClientOptionsCache = new Map<string, CachedTelegramClientOptions>();
 const MAX_TELEGRAM_CLIENT_OPTIONS_CACHE_SIZE = 64;
 
 export function resetTelegramClientOptionsCacheForTests(): void {
@@ -346,23 +360,66 @@ function buildTelegramClientOptionsCacheKey(params: {
   return `${params.account.accountId}::${proxyKey}::${autoSelectFamilyKey}::${dnsResultOrderKey}::${apiRootKey}::${timeoutSecondsKey}`;
 }
 
+function closeCachedTelegramClientOptions(entry: CachedTelegramClientOptions): void {
+  // Eviction may retire a cache entry while a send still holds a lease; defer
+  // transport.close until the last op-level lease releases so mid-request sockets stay open.
+  entry.retired = true;
+  if (entry.activeLeases > 0 || entry.closeStarted) {
+    return;
+  }
+  entry.closeStarted = true;
+  void entry.transport.close().catch((err: unknown) => {
+    diagLogger.warn(
+      `telegram client options cache transport close failed: ${redactSensitiveText(
+        formatUncaughtError(err),
+      )}`,
+    );
+  });
+}
+
+function leaseCachedTelegramClientOptions(
+  entry: CachedTelegramClientOptions,
+): TelegramClientOptionsLease {
+  entry.activeLeases += 1;
+  let released = false;
+  return {
+    release: () => {
+      if (released) {
+        return;
+      }
+      released = true;
+      entry.activeLeases = Math.max(0, entry.activeLeases - 1);
+      if (entry.retired) {
+        closeCachedTelegramClientOptions(entry);
+      }
+    },
+  };
+}
+
 function setCachedTelegramClientOptions(
   cacheKey: string,
-  clientOptions: ApiClientOptions | undefined,
-): ApiClientOptions | undefined {
-  telegramClientOptionsCache.set(cacheKey, clientOptions);
+  entry: CachedTelegramClientOptions,
+): ResolvedTelegramClientOptions {
+  telegramClientOptionsCache.set(cacheKey, entry);
   if (telegramClientOptionsCache.size > MAX_TELEGRAM_CLIENT_OPTIONS_CACHE_SIZE) {
     const oldestKey = telegramClientOptionsCache.keys().next().value;
     if (oldestKey !== undefined) {
+      const evictedEntry = telegramClientOptionsCache.get(oldestKey);
       telegramClientOptionsCache.delete(oldestKey);
+      if (evictedEntry) {
+        closeCachedTelegramClientOptions(evictedEntry);
+      }
     }
   }
-  return clientOptions;
+  return {
+    clientOptions: entry.clientOptions,
+    lease: () => leaseCachedTelegramClientOptions(entry),
+  };
 }
 
 function resolveTelegramClientOptions(
   account: ResolvedTelegramAccount,
-): ApiClientOptions | undefined {
+): ResolvedTelegramClientOptions {
   const timeoutSeconds =
     typeof account.config.timeoutSeconds === "number" &&
     Number.isFinite(account.config.timeoutSeconds)
@@ -377,7 +434,13 @@ function resolveTelegramClientOptions(
       })
     : null;
   if (cacheKey && telegramClientOptionsCache.has(cacheKey)) {
-    return telegramClientOptionsCache.get(cacheKey);
+    const entry = telegramClientOptionsCache.get(cacheKey);
+    if (entry) {
+      return {
+        clientOptions: entry.clientOptions,
+        lease: () => leaseCachedTelegramClientOptions(entry),
+      };
+    }
   }
 
   const proxyUrl = normalizeOptionalString(account.config.proxy);
@@ -401,9 +464,15 @@ function resolveTelegramClientOptions(
         }
       : undefined;
   if (cacheKey) {
-    return setCachedTelegramClientOptions(cacheKey, clientOptions);
+    return setCachedTelegramClientOptions(cacheKey, {
+      activeLeases: 0,
+      clientOptions,
+      closeStarted: false,
+      retired: false,
+      transport,
+    });
   }
-  return clientOptions;
+  return { clientOptions };
 }
 
 function resolveToken(explicit: string | undefined, params: { accountId: string; token: string }) {
@@ -564,6 +633,7 @@ type TelegramApiContext = {
   cfg: OpenClawConfig;
   account: ResolvedTelegramAccount;
   api: TelegramApi;
+  clientOptionsLease?: TelegramClientOptionsLease | undefined;
 };
 
 function resolveTelegramApiContext(opts: {
@@ -578,16 +648,32 @@ function resolveTelegramApiContext(opts: {
     accountId: opts.accountId,
   });
   const token = resolveToken(opts.token, account);
-  const client = resolveTelegramClientOptions(account);
   let api: TelegramApi;
+  let clientOptionsLease: TelegramClientOptionsLease | undefined;
   if (opts.api) {
     api = opts.api as TelegramApi;
   } else {
-    const bot = new Bot(token, client ? { client } : undefined);
+    const client = resolveTelegramClientOptions(account);
+    // One op-level lease covers the full send/action (including pre-request work
+    // and retries) so eviction cannot close the transport mid-operation.
+    clientOptionsLease = client.lease?.();
+    const bot = new Bot(token, client.clientOptions ? { client: client.clientOptions } : undefined);
     bot.api.config.use(getOrCreateAccountThrottler(token));
     api = bot.api;
   }
-  return { cfg, account, api };
+  return {
+    cfg,
+    account,
+    api,
+    ...(clientOptionsLease ? { clientOptionsLease } : {}),
+  };
+}
+
+function withTelegramApiContextLease<T>(
+  context: TelegramApiContext,
+  operation: Promise<T>,
+): Promise<T> {
+  return operation.finally(() => context.clientOptionsLease?.release());
 }
 
 type TelegramRequestWithDiag = <T>(
@@ -704,7 +790,20 @@ export async function sendMessageTelegram(
   text: string,
   opts: TelegramSendOpts,
 ): Promise<TelegramSendResult> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    sendMessageTelegramWithContext(to, text, opts, context),
+  );
+}
+
+async function sendMessageTelegramWithContext(
+  to: string,
+  text: string,
+  opts: TelegramSendOpts,
+  apiContext: TelegramApiContext,
+): Promise<TelegramSendResult> {
+  const { cfg, account, api } = apiContext;
   const botUserId = resolveTelegramBotUserIdFromToken(opts.token || account.token);
   const target = parseTelegramTarget(to);
   const chatId = await resolveAndPersistChatId({
@@ -1469,7 +1568,16 @@ export async function sendTypingTelegram(
   to: string,
   opts: TelegramTypingOpts,
 ): Promise<{ ok: true }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(context, sendTypingTelegramWithContext(to, opts, context));
+}
+
+async function sendTypingTelegramWithContext(
+  to: string,
+  opts: TelegramTypingOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true }> {
+  const { cfg, account, api } = context;
   const target = parseTelegramTarget(to);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1504,7 +1612,21 @@ export async function reactMessageTelegram(
   emoji: string,
   opts: TelegramReactionOpts,
 ): Promise<{ ok: true } | { ok: false; warning: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    reactMessageTelegramWithContext(chatIdInput, messageIdInput, emoji, opts, context),
+  );
+}
+
+async function reactMessageTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number,
+  emoji: string,
+  opts: TelegramReactionOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true } | { ok: false; warning: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1561,7 +1683,20 @@ export async function deleteMessageTelegram(
   messageIdInput: string | number,
   opts: TelegramDeleteOpts,
 ): Promise<{ ok: true } | { ok: false; warning: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    deleteMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context),
+  );
+}
+
+async function deleteMessageTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number,
+  opts: TelegramDeleteOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true } | { ok: false; warning: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1603,7 +1738,20 @@ export async function pinMessageTelegram(
   messageIdInput: string | number,
   opts: TelegramDeleteOpts,
 ): Promise<{ ok: true; messageId: string; chatId: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    pinMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context),
+  );
+}
+
+async function pinMessageTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number,
+  opts: TelegramDeleteOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true; messageId: string; chatId: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1636,7 +1784,20 @@ export async function unpinMessageTelegram(
   messageIdInput: string | number | undefined,
   opts: TelegramDeleteOpts,
 ): Promise<{ ok: true; chatId: string; messageId?: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    unpinMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context),
+  );
+}
+
+async function unpinMessageTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number | undefined,
+  opts: TelegramDeleteOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true; chatId: string; messageId?: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1697,7 +1858,28 @@ export async function editForumTopicTelegram(
     throw new Error("Telegram forum topic update requires a name or iconCustomEmojiId");
   }
 
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    editForumTopicTelegramWithContext(chatIdInput, messageThreadIdInput, opts, context),
+  );
+}
+
+async function editForumTopicTelegramWithContext(
+  chatIdInput: string | number,
+  messageThreadIdInput: string | number,
+  opts: TelegramEditForumTopicOpts,
+  context: TelegramApiContext,
+): Promise<{
+  ok: true;
+  chatId: string;
+  messageThreadId: number;
+  name?: string;
+  iconCustomEmojiId?: string;
+}> {
+  const trimmedName = opts.name?.trim();
+  const trimmedIconCustomEmojiId = opts.iconCustomEmojiId?.trim();
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const target = parseTelegramTarget(rawTarget);
   const chatId = await resolveAndPersistChatId({
@@ -1788,10 +1970,21 @@ export async function editMessageReplyMarkupTelegram(
   buttons: TelegramInlineButtons,
   opts: TelegramEditReplyMarkupOpts,
 ): Promise<{ ok: true; messageId: string; chatId: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext({
-    ...opts,
-    cfg: opts.cfg,
-  });
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    editMessageReplyMarkupTelegramWithContext(chatIdInput, messageIdInput, buttons, opts, context),
+  );
+}
+
+async function editMessageReplyMarkupTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number,
+  buttons: TelegramInlineButtons,
+  opts: TelegramEditReplyMarkupOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true; messageId: string; chatId: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -1832,10 +2025,21 @@ export async function editMessageTelegram(
   text: string,
   opts: TelegramEditOpts,
 ): Promise<{ ok: true; messageId: string; chatId: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext({
-    ...opts,
-    cfg: opts.cfg,
-  });
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    editMessageTelegramWithContext(chatIdInput, messageIdInput, text, opts, context),
+  );
+}
+
+async function editMessageTelegramWithContext(
+  chatIdInput: string | number,
+  messageIdInput: string | number,
+  text: string,
+  opts: TelegramEditOpts,
+  context: TelegramApiContext,
+): Promise<{ ok: true; messageId: string; chatId: string }> {
+  const { cfg, account, api } = context;
   const rawTarget = String(chatIdInput);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -2063,7 +2267,20 @@ export async function sendStickerTelegram(
     throw new Error("Telegram sticker file_id is required");
   }
 
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    sendStickerTelegramWithContext(to, fileId, opts, context),
+  );
+}
+
+async function sendStickerTelegramWithContext(
+  to: string,
+  fileId: string,
+  opts: TelegramStickerOpts,
+  context: TelegramApiContext,
+): Promise<TelegramSendResult> {
+  const { cfg, account, api } = context;
   const target = parseTelegramTarget(to);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -2145,7 +2362,17 @@ export async function sendPollTelegram(
   poll: PollInput,
   opts: TelegramPollOpts,
 ): Promise<{ messageId: string; chatId: string; pollId?: string }> {
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(context, sendPollTelegramWithContext(to, poll, opts, context));
+}
+
+async function sendPollTelegramWithContext(
+  to: string,
+  poll: PollInput,
+  opts: TelegramPollOpts,
+  context: TelegramApiContext,
+): Promise<{ messageId: string; chatId: string; pollId?: string }> {
+  const { cfg, account, api } = context;
   const target = parseTelegramTarget(to);
   const chatId = await resolveAndPersistChatId({
     cfg,
@@ -2267,7 +2494,21 @@ export async function createForumTopicTelegram(
     throw new Error("Forum topic name must be 128 characters or fewer");
   }
 
-  const { cfg, account, api } = resolveTelegramApiContext(opts);
+  const context = resolveTelegramApiContext(opts);
+  return withTelegramApiContextLease(
+    context,
+    createForumTopicTelegramWithContext(chatId, name, opts, context),
+  );
+}
+
+async function createForumTopicTelegramWithContext(
+  chatId: string,
+  name: string,
+  opts: TelegramCreateForumTopicOpts,
+  context: TelegramApiContext,
+): Promise<TelegramCreateForumTopicResult> {
+  const trimmedName = name.trim();
+  const { cfg, account, api } = context;
   // Accept topic-qualified targets (e.g. telegram:group:<id>:topic:<thread>)
   // but createForumTopic must always target the base supergroup chat id.
   const target = parseTelegramTarget(chatId);