fix(message-handler): cleanup stale pendingReplies to prevent memory leak

README.md

@@ -403,6 +403,21 @@ CHAT_BUDDY_RESPONSE_DEBOUNCE_MS=1800 chat-buddy run
 
 Note: command messages such as `/time`, `/history`, `/reset`, and `/schedule` are handled immediately and are not debounced.
 
+### Pending reply cleanup
+
+Chat Buddy automatically prunes stale pending reply state for inactive users.
+
+- `CHAT_BUDDY_PENDING_REPLY_TTL_HOURS`: inactivity TTL in hours before a pending reply entry is removed. Default: `1`.
+- `CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS`: cleanup check interval in milliseconds. Default: `300000` (5 minutes).
+
+Example:
+
+```bash
+CHAT_BUDDY_PENDING_REPLY_TTL_HOURS=2 CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS=120000 chat-buddy run
+```
+
+This helps keep long-running bot instances stable by clearing stale user state after the configured inactivity period.
+
 ---
 
 ## Architecture

src/cli/run.ts

@@ -8,6 +8,7 @@ import dotenv from "dotenv";
 import { loadConfig, configExists, BotConfig } from "../storage/configStore.js";
 import { WhatsAppBot } from "../bot.js";
 import { resolveAuthContext } from "../auth/googleAuth.js";
+import { stopPendingReplyCleanup } from "../services/messageHandler.service.js";
 
 const envPath = path.join(process.cwd(), ".env");
 const envPathAlt = path.join(process.cwd(), "env");
@@ -72,4 +73,12 @@ export const runBot = async (): Promise<void> => {
 
   const bot = new WhatsAppBot(username, agentName);
   bot.start();
+
+  const shutdown = (): void => {
+    stopPendingReplyCleanup();
+    process.exit();
+  };
+
+  process.once("SIGINT", shutdown);
+  process.once("SIGTERM", shutdown);
 };

src/services/messageHandler.service.test.ts

@@ -0,0 +1,80 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("../agents/agent.servce.js", () => ({
+  runAgent: vi.fn(async () => "Test reply"),
+}));
+
+vi.mock("./memory.service.js", () => ({
+  storeMessage: vi.fn(),
+}));
+
+vi.mock("./command.service.js", () => ({
+  handleCommand: vi.fn(async () => {}),
+}));
+
+import {
+  cleanupPendingReplies,
+  getPendingReplyCount,
+  handleMessages,
+  stopPendingReplyCleanup,
+} from "./messageHandler.service.js";
+
+describe("messageHandler service", () => {
+  const futureTimestamp = Math.floor(new Date("2099-01-01T00:00:00Z").getTime() / 1000);
+
+  beforeEach(() => {
+    vi.useFakeTimers({ now: new Date(2025, 0, 1, 0, 0, 0) });
+    process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS = "86400000";
+    process.env.CHAT_BUDDY_PENDING_REPLY_TTL_HOURS = "1";
+    process.env.CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS = "300000";
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    stopPendingReplyCleanup();
+    delete process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS;
+    delete process.env.CHAT_BUDDY_PENDING_REPLY_TTL_HOURS;
+    delete process.env.CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS;
+    vi.restoreAllMocks();
+  });
+
+  it("removes stale pending replies after the configured TTL", async () => {
+    const message = {
+      fromMe: false,
+      timestamp: futureTimestamp,
+      body: "hello",
+      from: "user@c.us",
+      getContact: vi.fn(async () => ({ pushname: "Test User", number: "12345" })),
+      reply: vi.fn(async () => {}),
+    } as any;
+
+    await handleMessages(message, "Asad", "Luffy");
+    expect(getPendingReplyCount()).toBe(1);
+
+    vi.setSystemTime(Date.now() + 2 * 60 * 60 * 1000);
+    cleanupPendingReplies();
+
+    expect(getPendingReplyCount()).toBe(0);
+  });
+
+  it("processes buffered replies and clears the pending entry after completion", async () => {
+    process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS = "2200";
+
+    const message = {
+      fromMe: false,
+      timestamp: futureTimestamp,
+      body: "hi",
+      from: "user@c.us",
+      getContact: vi.fn(async () => ({ pushname: "Test User", number: "12345" })),
+      reply: vi.fn(async () => {}),
+    } as any;
+
+    await handleMessages(message, "Asad", "Luffy");
+    expect(getPendingReplyCount()).toBe(1);
+
+    await vi.advanceTimersByTimeAsync(2200);
+
+    expect(getPendingReplyCount()).toBe(0);
+    expect(message.reply).toHaveBeenCalled();
+  });
+});

src/services/messageHandler.service.ts

@@ -19,9 +19,11 @@ type PendingUserReply = {
   agentName: string;
   timer: ReturnType<typeof setTimeout> | null;
   processing: boolean;
+  lastActivity: number;
 };
 
 const pendingReplies = new Map<string, PendingUserReply>();
+let cleanupTimer: ReturnType<typeof setInterval> | null = null;
 
 export const getDebounceMs = (): number => {
   const value = Number(process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS ?? "2200");
@@ -31,6 +33,65 @@ export const getDebounceMs = (): number => {
   return Math.floor(value);
 };
 
+const getPendingReplyTTLHours = (): number => {
+  const value = Number(process.env.CHAT_BUDDY_PENDING_REPLY_TTL_HOURS ?? "1");
+  if (!Number.isFinite(value)) return 1;
+  if (value < 0.1) return 0.1;
+  if (value > 168) return 168;
+  return value;
+};
+
+const getPendingReplyCleanupIntervalMs = (): number => {
+  const value = Number(
+    process.env.CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS ?? `${5 * 60 * 1000}`,
+  );
+  if (!Number.isFinite(value)) return 5 * 60 * 1000;
+  if (value < 60000) return 60000;
+  if (value > 3600000) return 3600000;
+  return Math.floor(value);
+};
+
+export const getPendingReplyCount = (): number => pendingReplies.size;
+
+const getPendingReplyTTLMs = (): number => Math.floor(getPendingReplyTTLHours() * 60 * 60 * 1000);
+
+export const cleanupPendingReplies = (): void => {
+  const now = Date.now();
+  const ttlMs = getPendingReplyTTLMs();
+
+  for (const [userId, pending] of pendingReplies) {
+    if (now - pending.lastActivity < ttlMs) {
+      continue;
+    }
+
+    if (pending.timer) {
+      clearTimeout(pending.timer);
+      pending.timer = null;
+    }
+
+    if (pending.processing) {
+      continue;
+    }
+
+    pendingReplies.delete(userId);
+  }
+};
+
+export const stopPendingReplyCleanup = (): void => {
+  if (cleanupTimer) {
+    clearInterval(cleanupTimer);
+    cleanupTimer = null;
+  }
+};
+
+const startPendingReplyCleanup = (): void => {
+  if (cleanupTimer) return;
+
+  cleanupTimer = setInterval(cleanupPendingReplies, getPendingReplyCleanupIntervalMs());
+};
+
+startPendingReplyCleanup();
+
 const scheduleBufferedReply = (userId: string): void => {
   const pending = pendingReplies.get(userId);
   if (!pending) return;
@@ -129,13 +190,15 @@ export const handleMessages = async (
       agentName,
       timer: null,
       processing: false,
+      lastActivity: Date.now(),
     });
   } else {
     existing.messages.push(text);
     existing.latestMessage = message;
     existing.contactName = contactName;
     existing.username = username;
     existing.agentName = agentName;
+    existing.lastActivity = Date.now();
   }
 
   scheduleBufferedReply(userId);

tests/messageHandlerService.test.ts

@@ -1,12 +1,14 @@
-import { describe, test, expect, afterEach, vi } from "vitest";
+import { describe, test, expect, afterEach, beforeEach, vi } from "vitest";
 
 vi.mock("../src/bot.js", () => ({ botRebootTime: Date.now() }));
-vi.mock("../src/agents/agent.servce.js", () => ({ runAgent: vi.fn() }));
-vi.mock("../src/config/agent.protocol.js", () => ({ createProtocols: vi.fn() }));
+vi.mock("../src/agents/agent.servce.js", () => ({ runAgent: vi.fn(async () => "Test reply") }));
+vi.mock("../src/config/agent.protocol.js", () => ({ createProtocols: vi.fn(() => ({ allowGroupReplies: true })) }));
 vi.mock("../src/services/memory.service.js", () => ({ storeMessage: vi.fn() }));
-vi.mock("../src/services/command.service.js", () => ({ handleCommand: vi.fn() }));
+vi.mock("../src/services/command.service.js", () => ({ handleCommand: vi.fn(async () => {}) }));
 
-import { getDebounceMs } from "../src/services/messageHandler.service.js";
+import { getDebounceMs, cleanupPendingReplies, getPendingReplyCount, handleMessages, stopPendingReplyCleanup } from "../src/services/messageHandler.service.js";
+
+const futureTimestamp = Math.floor(new Date("2099-01-01T00:00:00Z").getTime() / 1000);
 
 describe("getDebounceMs()", () => {
   afterEach(() => {
@@ -48,3 +50,61 @@ describe("getDebounceMs()", () => {
     expect(getDebounceMs()).toBe(2500);
   });
 });
+
+describe("messageHandler service", () => {
+  beforeEach(() => {
+    vi.useFakeTimers({ now: new Date(2025, 0, 1, 0, 0, 0) });
+    process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS = "86400000";
+    process.env.CHAT_BUDDY_PENDING_REPLY_TTL_HOURS = "1";
+    process.env.CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS = "300000";
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    stopPendingReplyCleanup();
+    delete process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS;
+    delete process.env.CHAT_BUDDY_PENDING_REPLY_TTL_HOURS;
+    delete process.env.CHAT_BUDDY_PENDING_REPLY_CLEANUP_INTERVAL_MS;
+    vi.restoreAllMocks();
+  });
+
+  test("removes stale pending replies after the configured TTL", async () => {
+    const message = {
+      fromMe: false,
+      timestamp: futureTimestamp,
+      body: "hello",
+      from: "user@c.us",
+      getContact: vi.fn(async () => ({ pushname: "Test User", number: "12345" })),
+      reply: vi.fn(async () => {}),
+    } as any;
+
+    await handleMessages(message, "Asad", "Luffy");
+    expect(getPendingReplyCount()).toBe(1);
+
+    vi.setSystemTime(Date.now() + 2 * 60 * 60 * 1000);
+    cleanupPendingReplies();
+
+    expect(getPendingReplyCount()).toBe(0);
+  });
+
+  test("processes buffered replies and clears the pending entry after completion", async () => {
+    process.env.CHAT_BUDDY_RESPONSE_DEBOUNCE_MS = "2200";
+
+    const message = {
+      fromMe: false,
+      timestamp: futureTimestamp,
+      body: "hi",
+      from: "user@c.us",
+      getContact: vi.fn(async () => ({ pushname: "Test User", number: "12345" })),
+      reply: vi.fn(async () => {}),
+    } as any;
+
+    await handleMessages(message, "Asad", "Luffy");
+    expect(getPendingReplyCount()).toBe(1);
+
+    await vi.advanceTimersByTimeAsync(2200);
+
+    expect(getPendingReplyCount()).toBe(0);
+    expect(message.reply).toHaveBeenCalled();
+  });
+});