Debounce inbound messages so forward + comment reach the agent as one turn

Telegram sends "forward with a comment" as two separate messages. Buffer
inputs for ~600ms after each arrival and flush a single merged AgentInput;
label forwarded content with its origin so the merged text is intelligible.

src/telegram/bridge.test.ts

@@ -55,7 +55,7 @@ class FakeScheduler implements SchedulerLike {
   }
 }
 
-function harness() {
+function harness(overrides: { debounceMs?: number } = {}) {
   const botInfo: UserFromGetMe = {
     id: 1, is_bot: true, first_name: "sec", username: "sec_bot", can_join_groups: false,
     can_read_all_group_messages: false, supports_inline_queries: false, can_connect_to_business: false, has_main_web_app: false,
@@ -82,6 +82,7 @@ function harness() {
     scheduler,
     outbound,
     logger,
+    debounceMs: overrides.debounceMs ?? 5,
   });
   const texts = () => calls.filter((c) => c.method === "sendMessage").map((c) => c.payload.text as string);
   return { bot, calls, host, scheduler, texts };
@@ -142,15 +143,46 @@ describe("registerBridge", () => {
     expect(h.texts().slice(1)).toEqual(["Cancelled job abc12.", "No job with id abc12.", "Usage: /cancel <id>", "No scheduled jobs."]);
   });
 
-  it("forwards text to the host and reacts when busy", async () => {
+  it("forwards text to the host after the debounce window and reacts when busy", async () => {
     const h = harness();
     await h.bot.handleUpdate(textUpdate("what time is it"));
-    expect(h.host.sent).toEqual([{ text: "what time is it", images: undefined }]);
+    await vi.waitFor(() => expect(h.host.sent).toEqual([{ text: "what time is it", images: undefined }]));
     expect(h.calls.find((c) => c.method === "setMessageReaction")).toBeUndefined();
     h.host.busy = true;
     await h.bot.handleUpdate(textUpdate("also this"));
     expect(h.calls.find((c) => c.method === "setMessageReaction")).toBeDefined();
-    expect(h.host.sent).toHaveLength(2);
+    await vi.waitFor(() => expect(h.host.sent).toHaveLength(2));
+  });
+
+  it("merges a forward-with-comment pair into a single agent turn", async () => {
+    const h = harness({ debounceMs: 30 });
+    // Comment arrives first.
+    await h.bot.handleUpdate(textUpdate("look at this"));
+    // Then the forwarded message from Alice, within the window.
+    const fwd = textUpdate("hello everyone");
+    (fwd.message as { forward_origin?: unknown }).forward_origin = {
+      type: "user",
+      date: 0,
+      sender_user: { id: 77, is_bot: false, first_name: "Alice", username: "alice" },
+    };
+    await h.bot.handleUpdate(fwd);
+    // Nothing has been delivered yet.
+    expect(h.host.sent).toEqual([]);
+    await vi.waitFor(() =>
+      expect(h.host.sent).toEqual([
+        { text: "look at this\n\n[Forwarded from @alice]\nhello everyone", images: undefined },
+      ]),
+    );
+  });
+
+  it("drops buffered messages when /new starts a fresh session", async () => {
+    const h = harness({ debounceMs: 100 });
+    await h.bot.handleUpdate(textUpdate("hold this thought"));
+    await h.bot.handleUpdate(textUpdate("/new"));
+    // Wait past the debounce window; nothing should reach the host.
+    await new Promise((r) => setTimeout(r, 150));
+    expect(h.host.sent).toEqual([]);
+    expect(h.host.newSessions).toBe(1);
   });
 
   it("rejects unknown commands without bothering the agent", async () => {

src/telegram/bridge.ts

@@ -7,9 +7,13 @@ import type { Logger } from "../log.js";
 import type { Scheduler } from "../scheduler/scheduler.js";
 import type { Transcriber } from "../stt/openai.js";
 import { describeJob } from "../tools/schedule.js";
+import { InputDebouncer } from "./debouncer.js";
 import { normalizeMessage } from "./inbound.js";
 import type { Outbound } from "./outbound.js";
 
+/** How long we wait after the last message before delivering the merged batch to the agent. */
+const DEFAULT_DEBOUNCE_MS = 600;
+
 export type HostLike = Pick<AgentHost, "busy" | "send" | "abort" | "newSession" | "compact" | "status" | "subscribe">;
 export type SchedulerLike = Pick<Scheduler, "list" | "delete">;
 
@@ -22,6 +26,8 @@ export interface BridgeDeps {
   transcriber?: Transcriber;
   logger: Logger;
   fetchFn?: typeof fetch;
+  /** Debounce window for merging rapid-succession messages (e.g. forward + comment). Defaults to 600ms. */
+  debounceMs?: number;
 }
 
 export const BOT_COMMANDS = [
@@ -36,6 +42,15 @@ export const BOT_COMMANDS = [
 export function registerBridge(deps: BridgeDeps): void {
   const { bot, config, host, scheduler, outbound, logger } = deps;
   const warned = new Set<number>();
+  const debouncer = new InputDebouncer({
+    windowMs: deps.debounceMs ?? DEFAULT_DEBOUNCE_MS,
+    onFlush: (input) => {
+      void host.send(input).catch((err) => {
+        logger.error("agent send failed", err);
+        void outbound.sendPlain(`⚠️ ${errorText(err)}`).catch(() => undefined);
+      });
+    },
+  });
 
   bot.use(async (ctx, next) => {
     const from = ctx.from?.id;
@@ -58,6 +73,7 @@ export function registerBridge(deps: BridgeDeps): void {
   });
 
   bot.command("new", async (ctx) => {
+    debouncer.discard();
     await host.newSession();
     await ctx.reply("New session started.");
   });
@@ -109,10 +125,7 @@ export function registerBridge(deps: BridgeDeps): void {
       return;
     }
     if (host.busy) await ctx.react("👀").catch(() => undefined);
-    void host.send(result.input).catch((err) => {
-      logger.error("agent send failed", err);
-      void outbound.sendPlain(`⚠️ ${errorText(err)}`).catch(() => undefined);
-    });
+    debouncer.add(result.input);
   });
 
   let queue: Promise<void> = Promise.resolve();

src/telegram/debouncer.test.ts

@@ -0,0 +1,92 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { AgentInput } from "../agent/events.js";
+import { InputDebouncer, merge } from "./debouncer.js";
+
+describe("InputDebouncer", () => {
+  beforeEach(() => vi.useFakeTimers());
+  afterEach(() => vi.useRealTimers());
+
+  it("flushes a single buffered input after the window", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 200, onFlush: (i) => flushed.push(i) });
+    d.add({ text: "hi" });
+    vi.advanceTimersByTime(199);
+    expect(flushed).toEqual([]);
+    vi.advanceTimersByTime(1);
+    expect(flushed).toEqual([{ text: "hi", images: undefined }]);
+  });
+
+  it("merges inputs that arrive within the window and resets the timer on each add", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 200, onFlush: (i) => flushed.push(i) });
+    d.add({ text: "my comment" });
+    vi.advanceTimersByTime(150);
+    d.add({ text: "[Forwarded from Alice]: hello" });
+    vi.advanceTimersByTime(150);
+    expect(flushed).toEqual([]);
+    vi.advanceTimersByTime(50);
+    expect(flushed).toEqual([{ text: "my comment\n\n[Forwarded from Alice]: hello", images: undefined }]);
+  });
+
+  it("concatenates images across buffered inputs", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) });
+    d.add({ text: "one", images: [{ data: "a", mimeType: "image/jpeg" }] });
+    d.add({ text: "two", images: [{ data: "b", mimeType: "image/jpeg" }] });
+    vi.advanceTimersByTime(100);
+    expect(flushed).toEqual([
+      {
+        text: "one\n\ntwo",
+        images: [
+          { data: "a", mimeType: "image/jpeg" },
+          { data: "b", mimeType: "image/jpeg" },
+        ],
+      },
+    ]);
+  });
+
+  it("does not fire when the buffer is empty", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 50, onFlush: (i) => flushed.push(i) });
+    d.flushNow();
+    vi.advanceTimersByTime(1000);
+    expect(flushed).toEqual([]);
+  });
+
+  it("flushNow delivers immediately and cancels the pending timer", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 500, onFlush: (i) => flushed.push(i) });
+    d.add({ text: "urgent" });
+    d.flushNow();
+    expect(flushed).toEqual([{ text: "urgent", images: undefined }]);
+    vi.advanceTimersByTime(1000);
+    expect(flushed).toHaveLength(1);
+  });
+
+  it("discard drops buffered items without delivering", () => {
+    const flushed: AgentInput[] = [];
+    const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) });
+    d.add({ text: "gone" });
+    d.discard();
+    vi.advanceTimersByTime(1000);
+    expect(flushed).toEqual([]);
+  });
+});
+
+describe("merge", () => {
+  it("joins texts with a blank line and preserves image order", () => {
+    expect(
+      merge([
+        { text: "a", images: [{ data: "x", mimeType: "image/jpeg" }] },
+        { text: "b" },
+      ]),
+    ).toEqual({ text: "a\n\nb", images: [{ data: "x", mimeType: "image/jpeg" }] });
+  });
+
+  it("skips empty text parts", () => {
+    expect(merge([{ text: "", images: [{ data: "x", mimeType: "image/jpeg" }] }, { text: "b" }])).toEqual({
+      text: "b",
+      images: [{ data: "x", mimeType: "image/jpeg" }],
+    });
+  });
+});

src/telegram/debouncer.ts

@@ -0,0 +1,58 @@
+import type { AgentImage, AgentInput } from "../agent/events.js";
+
+export interface InputDebouncerOptions {
+  windowMs: number;
+  onFlush: (input: AgentInput) => void;
+}
+
+/**
+ * Buffers AgentInputs arriving in quick succession and flushes a single merged
+ * input once the window goes quiet. Telegram sends "forward with a comment" as
+ * two separate messages (the comment, then the forwarded content); this stitches
+ * them together so the agent sees one turn.
+ */
+export class InputDebouncer {
+  private buffer: AgentInput[] = [];
+  private timer: NodeJS.Timeout | undefined;
+
+  constructor(private readonly opts: InputDebouncerOptions) {}
+
+  add(input: AgentInput): void {
+    this.buffer.push(input);
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = setTimeout(() => this.flush(), this.opts.windowMs);
+    this.timer.unref?.();
+  }
+
+  /** Flushes immediately (e.g. on shutdown or when the caller wants to force delivery). */
+  flushNow(): void {
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = undefined;
+    this.flush();
+  }
+
+  /** Drops the buffered items without delivering them. */
+  discard(): void {
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = undefined;
+    this.buffer = [];
+  }
+
+  private flush(): void {
+    this.timer = undefined;
+    if (this.buffer.length === 0) return;
+    const merged = merge(this.buffer);
+    this.buffer = [];
+    this.opts.onFlush(merged);
+  }
+}
+
+export function merge(inputs: AgentInput[]): AgentInput {
+  const parts: string[] = [];
+  const images: AgentImage[] = [];
+  for (const input of inputs) {
+    if (input.text) parts.push(input.text);
+    if (input.images) images.push(...input.images);
+  }
+  return { text: parts.join("\n\n"), images: images.length > 0 ? images : undefined };
+}

src/telegram/inbound.test.ts

@@ -91,6 +91,49 @@ describe("normalizeMessage", () => {
     expect(r.input).toEqual({ text: "see pic", images: undefined });
     expect(r.notes[0]).toMatch(/Could not download photo: unknown file missing/);
   });
+
+  it("labels forwarded text with the origin", async () => {
+    const r = await normalizeMessage(
+      { text: "look at this", forward_origin: { type: "user", sender_user: { first_name: "Alice", username: "alice" } } },
+      deps(),
+    );
+    expect(r.input).toEqual({ text: "[Forwarded from @alice]\nlook at this", images: undefined });
+  });
+
+  it("labels forwards from users without a username by name", async () => {
+    const r = await normalizeMessage(
+      { text: "hi", forward_origin: { type: "user", sender_user: { first_name: "Alice", last_name: "Smith" } } },
+      deps(),
+    );
+    expect(r.input?.text).toBe("[Forwarded from Alice Smith]\nhi");
+  });
+
+  it("labels channel forwards by the channel title", async () => {
+    const r = await normalizeMessage(
+      { text: "news", forward_origin: { type: "channel", chat: { title: "Daily News" } } },
+      deps(),
+    );
+    expect(r.input?.text).toBe("[Forwarded from Daily News]\nnews");
+  });
+
+  it("labels hidden-user forwards by their exposed name", async () => {
+    const r = await normalizeMessage(
+      { text: "secret", forward_origin: { type: "hidden_user", sender_user_name: "Someone" } },
+      deps(),
+    );
+    expect(r.input?.text).toBe("[Forwarded from Someone]\nsecret");
+  });
+
+  it("labels forwarded photos even when there is no caption", async () => {
+    const r = await normalizeMessage(
+      {
+        photo: [{ file_id: "p2", width: 1, height: 1 }],
+        forward_origin: { type: "user", sender_user: { first_name: "Alice" } },
+      },
+      deps(),
+    );
+    expect(r.input?.text).toBe("[Forwarded from Alice]\n(photo attached)");
+  });
 });
 
 describe("saveToInbox", () => {

src/telegram/inbound.ts

@@ -3,6 +3,12 @@ import path from "node:path";
 import type { AgentImage, AgentInput } from "../agent/events.js";
 import type { Transcriber } from "../stt/openai.js";
 
+export type ForwardOrigin =
+  | { type: "user"; sender_user: { first_name: string; last_name?: string; username?: string } }
+  | { type: "hidden_user"; sender_user_name: string }
+  | { type: "chat"; sender_chat: { title?: string; username?: string }; author_signature?: string }
+  | { type: "channel"; chat: { title?: string; username?: string }; author_signature?: string };
+
 export interface InboundMessage {
   text?: string;
   caption?: string;
@@ -10,6 +16,7 @@ export interface InboundMessage {
   voice?: { file_id: string; mime_type?: string; duration?: number };
   audio?: { file_id: string; mime_type?: string; file_name?: string };
   document?: { file_id: string; file_name?: string; mime_type?: string };
+  forward_origin?: ForwardOrigin;
 }
 
 export interface InboundDeps {
@@ -75,7 +82,11 @@ export async function normalizeMessage(msg: InboundMessage, deps: InboundDeps):
   }
 
   if (parts.length === 0 && images.length > 0) parts.push("(photo attached)");
-  const text = parts.join("\n\n");
+  let text = parts.join("\n\n");
+  if (msg.forward_origin && (text || images.length > 0)) {
+    const label = describeForwardOrigin(msg.forward_origin);
+    text = text ? `[Forwarded from ${label}]\n${text}` : `[Forwarded from ${label}]`;
+  }
   return {
     input: text ? { text, images: images.length > 0 ? images : undefined } : null,
     transcript,
@@ -83,6 +94,27 @@ export async function normalizeMessage(msg: InboundMessage, deps: InboundDeps):
   };
 }
 
+export function describeForwardOrigin(origin: ForwardOrigin): string {
+  switch (origin.type) {
+    case "user": {
+      const u = origin.sender_user;
+      if (u.username) return `@${u.username}`;
+      const name = [u.first_name, u.last_name].filter(Boolean).join(" ");
+      return name || "unknown user";
+    }
+    case "hidden_user":
+      return origin.sender_user_name || "hidden user";
+    case "chat": {
+      const c = origin.sender_chat;
+      return c.title || (c.username ? `@${c.username}` : "chat");
+    }
+    case "channel": {
+      const c = origin.chat;
+      return c.title || (c.username ? `@${c.username}` : "channel");
+    }
+  }
+}
+
 const MAX_STEM = 80;
 const MAX_EXT = 16;