feat(telegram): typing-indicator heartbeat while a turn is in flight

A chat turn can run up to ~10 min, during which the athlete previously
got zero feedback and would assume the bot was dead. Pulse Telegram's
native "typing" chat action every 4s for the generation phase of the
shared turn skeleton, stopping it in a finally around generation (never
delivery). Each pulse is best-effort — a rejected or throwing pulse is
logged at debug and never affects the turn or its reply — and the
interval is unref'd so a pending pulse can't hold the process open
during shutdown or an /update drain.

Closes #167

Author: yerzhansa

.changeset/typing-indicator-heartbeat.md

@@ -0,0 +1,7 @@
+---
+"cycling-coach": patch
+---
+
+User-facing: The bot now shows a typing indicator while it works on your message, so a long reply no longer looks like it went silent.
+
+Starts a best-effort heartbeat that re-emits Telegram's native "typing" chat action every 4s for the duration of the generation phase in the shared turn skeleton, then stops it in a `finally` around generation (never around delivery). Each pulse is isolated: a rejected or throwing pulse is logged at debug and can never affect the turn or its reply, and the interval is unref'd so a pending pulse cannot hold the process open during shutdown or an `/update` drain.

packages/core/src/channels/telegram.ts

@@ -47,13 +47,38 @@ const DELIVERY_FAILURE_HINT = `I generated the answer, but Telegram had trouble
 // must not be dressed in provider-specific vocabulary.
 const GENERIC_TRANSPORT_APOLOGY = "Sorry, something went wrong. Please try again.";
 
+// How often to re-emit Telegram's native "typing" indicator while a turn is in
+// flight. Telegram auto-clears the indicator ~5s after each sendChatAction, so we
+// refresh faster than that to keep it continuous without flicker.
+const TYPING_HEARTBEAT_MS = 4_000;
+
 function drainBounded(drain: () => Promise<void>, timeoutMs: number): Promise<void> {
   return Promise.race([
     drain(),
     new Promise<void>((resolve) => setTimeout(resolve, timeoutMs).unref?.()),
   ]);
 }
 
+// Pulse Telegram's native "typing" indicator on an interval so a long turn never
+// leaves the athlete staring at silence (a turn can now run up to ~10 min). Fires
+// once immediately, then every intervalMs. Each pulse is best-effort: a rejected
+// (or throwing) pulse is routed to onError and can never affect the turn or its
+// reply. The interval is unref'd so a pending pulse cannot hold the process open
+// during shutdown / an /update drain. Returns a stop() that clears the interval.
+export function startTypingHeartbeat(
+  pulse: () => Promise<unknown>,
+  intervalMs: number,
+  onError: (err: unknown) => void,
+): () => void {
+  const beat = () => {
+    void Promise.resolve().then(pulse).catch(onError);
+  };
+  beat();
+  const timer = setInterval(beat, intervalMs);
+  timer.unref?.();
+  return () => clearInterval(timer);
+}
+
 // ============================================================================
 // TELEGRAM BOT
 // ============================================================================
@@ -256,7 +281,10 @@ export function createTelegramBot(
   // vice versa. The only per-handler differences (genericReply text, log command
   // name, reply-to id) are passed in rather than re-templated.
   function runTurn(opts: {
-    ctx: { reply: (text: string, options?: Record<string, unknown>) => Promise<unknown> };
+    ctx: {
+      reply: (text: string, options?: Record<string, unknown>) => Promise<unknown>;
+      replyWithChatAction: (action: "typing") => Promise<unknown>;
+    };
     command: string;
     chatId: string;
     message: string;
@@ -265,6 +293,11 @@ export function createTelegramBot(
     replyToMessageId?: number;
   }): void {
     dispatch(async () => {
+      const stopHeartbeat = startTypingHeartbeat(
+        () => opts.ctx.replyWithChatAction("typing"),
+        TYPING_HEARTBEAT_MS,
+        () => log.debug("typing_heartbeat_failed", { command: opts.command, chatId: opts.chatId }),
+      );
       let response: string;
       try {
         response = await agent.chat(opts.chatId, opts.message, opts.deps);
@@ -273,6 +306,8 @@ export function createTelegramBot(
         const { kind, athleteMessage } = classifyAgentError(err);
         await opts.ctx.reply(kind === "unknown" ? opts.genericReply : athleteMessage);
         return;
+      } finally {
+        stopHeartbeat();
       }
       writeResend(opts.chatId, response);
       try {

packages/core/tests/telegram-dispatch.test.ts

@@ -4,6 +4,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { APICallError } from "@ai-sdk/provider";
 import { cyclingBinary } from "./helpers/cycling-binary-fixture.js";
+import { startTypingHeartbeat } from "../src/channels/telegram.js";
 
 let dataDir: string;
 
@@ -121,6 +122,7 @@ interface FakeCtx {
   message: { text: string; message_id?: number };
   reply: ReturnType<typeof vi.fn>;
   replyWithDocument: ReturnType<typeof vi.fn>;
+  replyWithChatAction: ReturnType<typeof vi.fn>;
 }
 
 function makeCtx(overrides?: Partial<FakeCtx>): FakeCtx {
@@ -130,6 +132,7 @@ function makeCtx(overrides?: Partial<FakeCtx>): FakeCtx {
     message: { text: "hi" },
     reply: vi.fn(async () => undefined),
     replyWithDocument: vi.fn(async () => undefined),
+    replyWithChatAction: vi.fn(async () => true),
     ...overrides,
   };
 }
@@ -446,6 +449,144 @@ describe("retry transformer / classified errors / delivery split", () => {
   });
 });
 
+describe("typing-indicator heartbeat", () => {
+  // Mirrors the un-exported TYPING_HEARTBEAT_MS in telegram.ts. The <= 5s
+  // assertion below keeps the heartbeat under Telegram's ~5s typing auto-clear window.
+  const HEARTBEAT_MS = 4_000;
+
+  describe("in-flight pulses (integration)", () => {
+    beforeEach(() => {
+      vi.useFakeTimers();
+    });
+    afterEach(() => {
+      vi.useRealTimers();
+    });
+
+    it("pulses 'typing' immediately and again each interval while agent.chat is in flight", async () => {
+      const { bot, agent, drainPending } = await buildBot();
+      agent.hasSession.mockReturnValue(true);
+      let resolveChat!: (v: string) => void;
+      agent.chat.mockReturnValue(
+        new Promise<string>((r) => {
+          resolveChat = r;
+        }),
+      );
+      const ctx = makeCtx({ message: { text: "how's my form?" } });
+
+      await getMessageText(bot)(ctx);
+
+      // Flush the immediate pulse microtask (fires before any interval elapses).
+      await vi.advanceTimersByTimeAsync(0);
+      expect(ctx.replyWithChatAction).toHaveBeenCalledWith("typing");
+      const afterImmediate = ctx.replyWithChatAction.mock.calls.length;
+      expect(afterImmediate).toBeGreaterThanOrEqual(1);
+
+      // One interval → at least one more pulse; and a second interval → another.
+      await vi.advanceTimersByTimeAsync(HEARTBEAT_MS);
+      await vi.advanceTimersByTimeAsync(HEARTBEAT_MS);
+      expect(ctx.replyWithChatAction.mock.calls.length).toBeGreaterThanOrEqual(afterImmediate + 2);
+      expect(HEARTBEAT_MS).toBeLessThanOrEqual(5_000);
+
+      // Let the turn finish so no fake interval leaks past the test.
+      resolveChat("done");
+      await drainPending();
+    });
+
+    it("stops the interval on normal completion — no further pulses", async () => {
+      const { bot, agent, drainPending } = await buildBot();
+      agent.hasSession.mockReturnValue(true);
+      let resolveChat!: (v: string) => void;
+      agent.chat.mockReturnValue(
+        new Promise<string>((r) => {
+          resolveChat = r;
+        }),
+      );
+      const ctx = makeCtx({ message: { text: "how's my form?" } });
+
+      await getMessageText(bot)(ctx);
+      await vi.advanceTimersByTimeAsync(HEARTBEAT_MS);
+
+      resolveChat("the answer body");
+      await drainPending();
+      expect(someReply(ctx, "the answer body")).toBe(true);
+
+      const frozen = ctx.replyWithChatAction.mock.calls.length;
+      await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 5);
+      expect(ctx.replyWithChatAction.mock.calls.length).toBe(frozen);
+    });
+
+    it("stops the interval on a generation error, and still fires the classified reply", async () => {
+      const { bot, agent, drainPending } = await buildBot();
+      agent.hasSession.mockReturnValue(true);
+      agent.chat.mockRejectedValue(rateLimitError(30));
+      const ctx = makeCtx({ message: { text: "plan my week" } });
+
+      await getMessageText(bot)(ctx);
+      await drainPending();
+      expect(someReply(ctx, "Rate limited — please try again in ~30 seconds.")).toBe(true);
+
+      const frozen = ctx.replyWithChatAction.mock.calls.length;
+      await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 5);
+      expect(ctx.replyWithChatAction.mock.calls.length).toBe(frozen);
+    });
+
+    it("a pulse that rejects on every call never affects the turn's reply path", async () => {
+      const { bot, agent, drainPending } = await buildBot();
+      agent.hasSession.mockReturnValue(true);
+      agent.chat.mockResolvedValue("the delivered answer");
+      const ctx = makeCtx({ message: { text: "how's my form?" } });
+      ctx.replyWithChatAction.mockRejectedValue(new Error("chat action failed"));
+
+      await getMessageText(bot)(ctx);
+      await drainPending();
+
+      // The reply still lands unchanged despite every pulse rejecting.
+      expect(someReply(ctx, "the delivered answer")).toBe(true);
+      const htmlReply = ctx.reply.mock.calls.find(
+        (c: unknown[]) => (c[1] as { parse_mode?: string } | undefined)?.parse_mode === "HTML",
+      );
+      expect(htmlReply).toBeDefined();
+    });
+  });
+
+  describe("startTypingHeartbeat (unit)", () => {
+    it("a rejecting pulse routes to onError and never throws to the caller", async () => {
+      const onError = vi.fn();
+      const stop = startTypingHeartbeat(
+        () => Promise.reject(new Error("pulse boom")),
+        10_000,
+        onError,
+      );
+      // Drain all pending microtasks (a rejected thenable adoption spans several
+      // ticks) so the immediate best-effort pulse settles into .catch(onError).
+      await new Promise((r) => setTimeout(r, 0));
+      expect(onError).toHaveBeenCalledTimes(1);
+      expect(() => stop()).not.toThrow();
+    });
+
+    it("unrefs the interval on start and clears exactly that timer on stop", () => {
+      const unref = vi.fn();
+      const fakeTimer = { unref };
+      const setIntervalSpy = vi.fn(() => fakeTimer);
+      const clearIntervalSpy = vi.fn();
+      vi.stubGlobal("setInterval", setIntervalSpy);
+      vi.stubGlobal("clearInterval", clearIntervalSpy);
+      try {
+        const stop = startTypingHeartbeat(async () => undefined, HEARTBEAT_MS, () => {});
+        expect(setIntervalSpy).toHaveBeenCalledTimes(1);
+        expect(unref).toHaveBeenCalledTimes(1);
+        expect(clearIntervalSpy).not.toHaveBeenCalled();
+
+        stop();
+        expect(clearIntervalSpy).toHaveBeenCalledTimes(1);
+        expect(clearIntervalSpy).toHaveBeenCalledWith(fakeTimer);
+      } finally {
+        vi.unstubAllGlobals();
+      }
+    });
+  });
+});
+
 describe("message:text — greet / re-greet logic", () => {
   it("first message from a newcomer (no session) → WELCOME then chat", async () => {
     const { bot, agent, drainPending } = await buildBot();