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

* 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

* fix(telegram): guard heartbeat pulses and export the interval constant

- add an in-flight guard so a beat is skipped while the previous pulse is
  still unsettled, preventing pulse pile-up when a pulse is parked in the API
  retry layer; the flag is released in a finally so a rejected pulse can't
  wedge it
- export TYPING_HEARTBEAT_MS and assert the <=5s auto-clear invariant against
  the production constant instead of a test-local mirror

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,51 @@ 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.
+export 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 {
+  // Skip a beat while the previous pulse is still unsettled: under a Telegram
+  // flood a pulse can be parked inside the API retry layer, and firing a fresh
+  // one every interval regardless would pile parked pulses up and roughly double
+  // request volume against an already-throttled API. The flag is cleared in a
+  // finally so a rejected pulse cannot wedge the guard permanently.
+  let inFlight = false;
+  const beat = () => {
+    if (inFlight) return;
+    inFlight = true;
+    void Promise.resolve()
+      .then(pulse)
+      .catch(onError)
+      .finally(() => {
+        inFlight = false;
+      });
+  };
+  beat();
+  const timer = setInterval(beat, intervalMs);
+  timer.unref?.();
+  return () => clearInterval(timer);
+}
+
 // ============================================================================
 // TELEGRAM BOT
 // ============================================================================
@@ -256,7 +294,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 +306,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 +319,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, TYPING_HEARTBEAT_MS } 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,173 @@ describe("retry transformer / classified errors / delivery split", () => {
   });
 });
 
+describe("typing-indicator heartbeat", () => {
+  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(TYPING_HEARTBEAT_MS);
+      await vi.advanceTimersByTimeAsync(TYPING_HEARTBEAT_MS);
+      expect(ctx.replyWithChatAction.mock.calls.length).toBeGreaterThanOrEqual(afterImmediate + 2);
+      expect(TYPING_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(TYPING_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(TYPING_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(TYPING_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, TYPING_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();
+      }
+    });
+
+    it("skips beats while the previous pulse is still unsettled, then resumes once it settles", async () => {
+      vi.useFakeTimers();
+      try {
+        let settle!: () => void;
+        const pulse = vi.fn(
+          () =>
+            new Promise<void>((resolve) => {
+              settle = resolve;
+            }),
+        );
+        const stop = startTypingHeartbeat(pulse, 10_000, () => {});
+
+        // The immediate beat fires exactly one pulse, which stays pending.
+        await vi.advanceTimersByTimeAsync(0);
+        expect(pulse).toHaveBeenCalledTimes(1);
+
+        // Several intervals elapse while that pulse is parked — every tick is
+        // skipped by the in-flight guard, so still only the one pulse.
+        await vi.advanceTimersByTimeAsync(10_000 * 3);
+        expect(pulse).toHaveBeenCalledTimes(1);
+
+        // Settle the parked pulse; the guard releases and the next tick beats.
+        settle();
+        await vi.advanceTimersByTimeAsync(0);
+        await vi.advanceTimersByTimeAsync(10_000);
+        expect(pulse).toHaveBeenCalledTimes(2);
+
+        stop();
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
+});
+
 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();