Merge branch 'main' into feat/opencode-integration

AGENTS.md

@@ -36,3 +36,19 @@
 ## Agent-Specific Notes
 - If the relay is running in tmux (`warelay-relay`), restart it after code changes: kill pane/session and run `pnpm warelay relay --verbose` inside tmux. Check tmux before editing; keep the watcher healthy if you start it.
 - Also read the shared guardrails at `~/Projects/oracle/AGENTS.md` and `~/Projects/agent-scripts/AGENTS.MD` before making changes; align with any cross-repo rules noted there.
+
+## Exclamation Mark Escaping Workaround
+The Claude Code Bash tool escapes `!` to `\!` in command arguments. When using `warelay send` with messages containing exclamation marks, use heredoc syntax:
+
+```bash
+# WRONG - will send "Hello\!" with backslash
+warelay send --provider web --to "+1234" --message 'Hello!'
+
+# CORRECT - use heredoc to avoid escaping
+warelay send --provider web --to "+1234" --message "$(cat <<'EOF'
+Hello!
+EOF
+)"
+```
+
+This is a Claude Code quirk, not a warelay bug.

CHANGELOG.md

@@ -1,10 +1,36 @@
 # Changelog
 
+## 1.3.0 — Unreleased
+
+### Bug Fixes
+- **Empty result field handling:** Fixed bug where Claude CLI returning `result: ""` (empty string) would cause raw JSON to be sent to WhatsApp instead of being treated as valid empty output. Changed truthy check to explicit type check in `command-reply.ts`.
+- **Response prefix on heartbeat replies:** Fixed `responsePrefix` (e.g., `🦞`) not being applied to heartbeat alert messages. The prefix was only applied in the regular message handler, not in `runReplyHeartbeat`.
+- **User-visible error messages:** Command failures (non-zero exit, killed processes, exceptions) now return user-friendly error messages to WhatsApp instead of silently failing with empty responses.
+- **Test session isolation:** Fixed tests corrupting production `sessions.json` by mocking session persistence in all test files.
+- **Signal session corruption prevention:** Added IPC mechanism so `warelay send` and `warelay heartbeat` reuse the running relay's WhatsApp connection instead of creating new Baileys sockets. Previously, using these commands while the relay was running could corrupt the Signal session ratchet (both connections wrote to the same auth state), causing the relay's subsequent sends to fail silently.
+
+### Changes
+- **IPC server for relay:** The web relay now starts a Unix socket server at `~/.warelay/relay.sock`. Commands like `warelay send --provider web` automatically connect via IPC when the relay is running, falling back to direct connection otherwise.
+- **Typing indicator after IPC send:** After sending a message via IPC (e.g., `warelay send`), the relay now automatically shows the typing indicator ("composing") to signal that more messages may be coming.
+- **Auto-recovery from stuck WhatsApp sessions:** Added watchdog timer that detects when WhatsApp event emitter stops firing (e.g., after Bad MAC decryption errors) and automatically restarts the connection after 30 minutes of no message activity. Heartbeat logging now includes `minutesSinceLastMessage` and warns when >30 minutes without messages. The 30-minute timeout is intentionally longer than typical `heartbeatMinutes` configs to avoid false positives.
+- **Early allowFrom filtering:** Unauthorized senders are now blocked in `inbound.ts` BEFORE encryption/decryption attempts, preventing Bad MAC errors from corrupting session state. Previously, messages from unauthorized senders would trigger decryption failures that could silently kill the event emitter.
+- **Test isolation improvements:** Mock `loadConfig()` in all test files to prevent loading real user config (with emojis/prefixes) during tests. Default test config now has no prefixes/timestamps for cleaner assertions.
+- **Same-phone mode (self-messaging):** warelay now supports running on the same phone number you message from. This enables setups where you chat with yourself to control an AI assistant. Same-phone mode (`from === to`) is always allowed, even without configuring `allowFrom`. Echo detection prevents infinite loops by tracking recently sent message text and skipping auto-replies when incoming messages match.
+- **Echo detection:** The `fromMe` filter in `inbound.ts` is deliberately removed for same-phone setups; instead, text-based echo detection in `auto-reply.ts` tracks sent messages in a bounded Set (max 100 entries) and skips processing when a match is found.
+- **Same-phone detection logging:** Verbose mode now logs `📱 Same-phone mode detected` when `from === to`.
+- **Configurable same-phone marker:** New `inbound.samePhoneMarker` config option to customize the prefix added to messages in same-phone mode (default: `[same-phone]`). Set it to something cute like `[🦞 same-phone]` to help distinguish bot replies.
+
 ## 1.2.2 — 2025-11-28
 
 ### Changes
 - **Manual heartbeat sends:** `warelay heartbeat` accepts `--message/--body` with `--provider web|twilio` to push real outbound messages through the same plumbing; `--dry-run` previews payloads without sending.
 
+## Unreleased
+
+### Changes
+- **Heartbeat backpressure:** Web reply heartbeats now check the shared command queue and skip while any command/Claude runs are in flight, preventing concurrent prompts during long-running requests.
+- **Isolated session fixtures in web tests:** Heartbeat/auto-reply tests now create temporary session stores instead of using the default `~/.warelay/sessions.json`, preventing local config pollution during test runs.
+
 ## 1.2.1 — 2025-11-28
 
 ### Changes

CLAUDE.md

@@ -0,0 +1 @@
+AGENTS.md

README.md

@@ -96,6 +96,16 @@ Install from npm (global): `npm install -g warelay` (Node 22+). Then choose **on
 
 Best practice: use a dedicated WhatsApp account (separate SIM/eSIM or business account) for automation instead of your primary personal account to avoid unexpected logouts or rate limits.
 
+### Same-phone mode (self-messaging)
+warelay supports running on the same phone number you message from—you chat with yourself and an AI assistant replies in the same bubble. This requires:
+- Adding your own number to `allowFrom` in `warelay.json`
+- The `fromMe` filter is disabled; echo detection in `auto-reply.ts` prevents loops
+
+**Gotchas:**
+- Messages appear in the same chat bubble (WhatsApp "Note to self")
+- Echo detection relies on exact text matching; if the reply is identical to your input, it may be skipped
+- Works best with a dedicated WhatsApp account
+
 ## Configuration
 
 ### Environment (.env)
@@ -160,6 +170,9 @@ Best practice: use a dedicated WhatsApp account (separate SIM/eSIM or business a
 | Key | Type & default | Notes |
 | --- | --- | --- |
 | `inbound.allowFrom` | `string[]` (default: empty) | E.164 numbers allowed to trigger auto-reply (no `whatsapp:`); `"*"` allows any sender. |
+| `inbound.messagePrefix` | `string` (default: `"[warelay]"` if no allowFrom, else `""`) | Prefix added to all inbound messages before passing to command. |
+| `inbound.responsePrefix` | `string` (default: —) | Prefix auto-added to all outbound replies (e.g., `"🦞"`). |
+| `inbound.timestampPrefix` | `boolean \| string` (default: `true`) | Timestamp prefix: `true` (UTC), `false` (disabled), or IANA timezone like `"Europe/Vienna"`. |
 | `inbound.reply.mode` | `"text"` \| `"command"` (default: —) | Reply style. |
 | `inbound.reply.text` | `string` (default: —) | Used when `mode=text`; templating supported. |
 | `inbound.reply.command` | `string[]` (default: —) | Argv for `mode=command`; each element templated. Stdout (trimmed) is sent. |

package.json

@@ -1,6 +1,6 @@
 {
   "name": "warelay",
-  "version": "1.2.2",
+  "version": "1.3.0",
   "description": "WhatsApp relay CLI (send, monitor, webhook, auto-reply) using Twilio",
   "type": "module",
   "main": "dist/index.js",

src/auto-reply/command-reply.test.ts

@@ -292,4 +292,79 @@ describe("runCommandReply", () => {
     expect(meta.queuedMs).toBe(25);
     expect(meta.queuedAhead).toBe(2);
   });
+
+  it("handles empty result string without dumping raw JSON", async () => {
+    // Bug fix: Claude CLI returning {"result": ""} should not send raw JSON to WhatsApp
+    // The fix changed from truthy check to explicit typeof check
+    const runner = makeRunner({
+      stdout: '{"result":"","duration_ms":50,"total_cost_usd":0.001}',
+    });
+    const { payload } = await runCommandReply({
+      reply: {
+        mode: "command",
+        command: ["claude", "{{Body}}"],
+        claudeOutputFormat: "json",
+      },
+      templatingCtx: noopTemplateCtx,
+      sendSystemOnce: false,
+      isNewSession: true,
+      isFirstTurnInSession: true,
+      systemSent: false,
+      timeoutMs: 1000,
+      timeoutSeconds: 1,
+      commandRunner: runner,
+      enqueue: enqueueImmediate,
+    });
+    // Should NOT contain raw JSON - empty result should produce fallback message
+    expect(payload?.text).not.toContain('{"result"');
+    expect(payload?.text).toContain("command produced no output");
+  });
+
+  it("handles empty text string in Claude JSON", async () => {
+    const runner = makeRunner({
+      stdout: '{"text":"","duration_ms":50}',
+    });
+    const { payload } = await runCommandReply({
+      reply: {
+        mode: "command",
+        command: ["claude", "{{Body}}"],
+        claudeOutputFormat: "json",
+      },
+      templatingCtx: noopTemplateCtx,
+      sendSystemOnce: false,
+      isNewSession: true,
+      isFirstTurnInSession: true,
+      systemSent: false,
+      timeoutMs: 1000,
+      timeoutSeconds: 1,
+      commandRunner: runner,
+      enqueue: enqueueImmediate,
+    });
+    // Empty text should produce fallback message, not raw JSON
+    expect(payload?.text).not.toContain('{"text"');
+    expect(payload?.text).toContain("command produced no output");
+  });
+
+  it("returns actual text when result is non-empty", async () => {
+    const runner = makeRunner({
+      stdout: '{"result":"hello world","duration_ms":50}',
+    });
+    const { payload } = await runCommandReply({
+      reply: {
+        mode: "command",
+        command: ["claude", "{{Body}}"],
+        claudeOutputFormat: "json",
+      },
+      templatingCtx: noopTemplateCtx,
+      sendSystemOnce: false,
+      isNewSession: true,
+      isFirstTurnInSession: true,
+      systemSent: false,
+      timeoutMs: 1000,
+      timeoutSeconds: 1,
+      commandRunner: runner,
+      enqueue: enqueueImmediate,
+    });
+    expect(payload?.text).toBe("hello world");
+  });
 });

src/auto-reply/command-reply.ts

@@ -269,7 +269,7 @@ export async function runCommandReply(
           `Claude JSON raw: ${JSON.stringify(parsed.parsed, null, 2)}`,
         );
       }
-      if (parsed?.text) {
+      if (typeof parsed?.text === "string") {
         logVerbose(
           `Claude JSON parsed -> ${parsed.text.slice(0, 120)}${parsed.text.length > 120 ? "…" : ""}`,
         );
@@ -306,8 +306,11 @@ export async function runCommandReply(
       console.error(
         `Command auto-reply exited with code ${code ?? "unknown"} (signal: ${signal ?? "none"})`,
       );
+      // Include any partial output or stderr in error message
+      const partialOut = trimmed ? `\n\nOutput: ${trimmed.slice(0, 500)}${trimmed.length > 500 ? "..." : ""}` : "";
+      const errorText = `⚠️ Command exited with code ${code ?? "unknown"}${signal ? ` (${signal})` : ""}${partialOut}`;
       return {
-        payload: undefined,
+        payload: { text: errorText },
         meta: {
           durationMs: Date.now() - started,
           queuedMs,
@@ -325,8 +328,9 @@ export async function runCommandReply(
       console.error(
         `Command auto-reply process killed before completion (exit code ${code ?? "unknown"})`,
       );
+      const errorText = `⚠️ Command was killed before completion (exit code ${code ?? "unknown"})`;
       return {
-        payload: undefined,
+        payload: { text: errorText },
         meta: {
           durationMs: Date.now() - started,
           queuedMs,
@@ -426,8 +430,11 @@ export async function runCommandReply(
       };
     }
     logError(`Command auto-reply failed after ${elapsed}ms: ${String(err)}`);
+    // Send error message to user so they know the command failed
+    const errMsg = err instanceof Error ? err.message : String(err);
+    const errorText = `⚠️ Command failed: ${errMsg}`;
     return {
-      payload: undefined,
+      payload: { text: errorText },
       meta: {
         durationMs: elapsed,
         queuedMs,

src/auto-reply/reply.ts

@@ -146,8 +146,14 @@ export async function getReplyFromConfig(
 
   // Optional allowlist by origin number (E.164 without whatsapp: prefix)
   const allowFrom = cfg.inbound?.allowFrom;
-  if (Array.isArray(allowFrom) && allowFrom.length > 0) {
+  const from = (ctx.From ?? "").replace(/^whatsapp:/, "");
-    const from = (ctx.From ?? "").replace(/^whatsapp:/, "");
+  const to = (ctx.To ?? "").replace(/^whatsapp:/, "");
+  const isSamePhone = from && to && from === to;
+
+  // Same-phone mode (self-messaging) is always allowed
+  if (isSamePhone) {
+    logVerbose(`Allowing same-phone mode: from === to (${from})`);
+  } else if (Array.isArray(allowFrom) && allowFrom.length > 0) {
     // Support "*" as wildcard to allow all senders
     if (!allowFrom.includes("*") && !allowFrom.includes(from)) {
       logVerbose(

src/commands/send.ts

@@ -1,7 +1,8 @@
 import type { CliDeps } from "../cli/deps.js";
-import { info } from "../globals.js";
+import { info, success } from "../globals.js";
 import type { RuntimeEnv } from "../runtime.js";
 import type { Provider } from "../utils.js";
+import { sendViaIpc } from "../web/ipc.js";
 
 export async function sendCommand(
   opts: {
@@ -39,6 +40,34 @@ export async function sendCommand(
     if (waitSeconds !== 0) {
       runtime.log(info("Wait/poll are Twilio-only; ignored for provider=web."));
     }
+
+    // Try to send via IPC to running relay first (avoids Signal session corruption)
+    const ipcResult = await sendViaIpc(opts.to, opts.message, opts.media);
+    if (ipcResult) {
+      if (ipcResult.success) {
+        runtime.log(success(`✅ Sent via relay IPC. Message ID: ${ipcResult.messageId}`));
+        if (opts.json) {
+          runtime.log(
+            JSON.stringify(
+              {
+                provider: "web",
+                via: "ipc",
+                to: opts.to,
+                messageId: ipcResult.messageId,
+                mediaUrl: opts.media ?? null,
+              },
+              null,
+              2,
+            ),
+          );
+        }
+        return;
+      }
+      // IPC failed but relay is running - warn and fall back
+      runtime.log(info(`IPC send failed (${ipcResult.error}), falling back to direct connection`));
+    }
+
+    // Fall back to direct connection (creates new Baileys socket)
     const res = await deps
       .sendMessageWeb(opts.to, opts.message, {
         verbose: false,
@@ -53,6 +82,7 @@ export async function sendCommand(
         JSON.stringify(
           {
             provider: "web",
+            via: "direct",
             to: opts.to,
             messageId: res.messageId,
             mediaUrl: opts.media ?? null,

src/config/config.ts

@@ -46,6 +46,9 @@ export type WarelayConfig = {
   logging?: LoggingConfig;
   inbound?: {
     allowFrom?: string[]; // E.164 numbers allowed to trigger auto-reply (without whatsapp:)
+    messagePrefix?: string; // Prefix added to all inbound messages (default: "[warelay]" if no allowFrom, else "")
+    responsePrefix?: string; // Prefix auto-added to all outbound replies (e.g., "🦞")
+    timestampPrefix?: boolean | string; // true/false or IANA timezone string (default: true with UTC)
     transcribeAudio?: {
       // Optional CLI to turn inbound audio into text; templated args, must output transcript to stdout.
       command: string[];
@@ -139,6 +142,9 @@ const WarelaySchema = z.object({
   inbound: z
     .object({
       allowFrom: z.array(z.string()).optional(),
+      messagePrefix: z.string().optional(),
+      responsePrefix: z.string().optional(),
+      timestampPrefix: z.union([z.boolean(), z.string()]).optional(),
       transcribeAudio: z
         .object({
           command: z.array(z.string()),

src/index.core.test.ts

@@ -9,6 +9,18 @@ import { createMockTwilio } from "../test/mocks/twilio.js";
 import * as exec from "./process/exec.js";
 import { withWhatsAppPrefix } from "./utils.js";
 
+// Mock config to avoid loading real user config
+vi.mock("../src/config/config.js", () => ({
+  loadConfig: vi.fn().mockReturnValue({
+    inbound: {
+      allowFrom: ["*"],
+      messagePrefix: undefined,
+      responsePrefix: undefined,
+      timestampPrefix: false,
+    },
+  }),
+}));
+
 // Twilio mock factory shared across tests
 vi.mock("twilio", () => {
   const { factory } = createMockTwilio();
@@ -93,6 +105,64 @@ describe("config and templating", () => {
     expect(onReplyStart).toHaveBeenCalled();
   });
 
+  it("getReplyFromConfig allows same-phone mode (from === to) without allowFrom", async () => {
+    const cfg = {
+      inbound: {
+        // No allowFrom configured
+        reply: {
+          mode: "text" as const,
+          text: "Echo: {{Body}}",
+        },
+      },
+    };
+
+    const result = await index.getReplyFromConfig(
+      { Body: "hello", From: "+1555", To: "+1555" },
+      undefined,
+      cfg,
+    );
+    expect(result?.text).toBe("Echo: hello");
+  });
+
+  it("getReplyFromConfig allows same-phone mode even when not in allowFrom list", async () => {
+    const cfg = {
+      inbound: {
+        allowFrom: ["+9999"], // Different number
+        reply: {
+          mode: "text" as const,
+          text: "Reply: {{Body}}",
+        },
+      },
+    };
+
+    // Same-phone mode should bypass allowFrom check
+    const result = await index.getReplyFromConfig(
+      { Body: "test", From: "+1555", To: "+1555" },
+      undefined,
+      cfg,
+    );
+    expect(result?.text).toBe("Reply: test");
+  });
+
+  it("getReplyFromConfig rejects non-same-phone when not in allowFrom", async () => {
+    const cfg = {
+      inbound: {
+        allowFrom: ["+9999"],
+        reply: {
+          mode: "text" as const,
+          text: "Should not see this",
+        },
+      },
+    };
+
+    const result = await index.getReplyFromConfig(
+      { Body: "test", From: "+1555", To: "+2666" },
+      undefined,
+      cfg,
+    );
+    expect(result).toBeUndefined();
+  });
+
   it("getReplyFromConfig templating includes media fields", async () => {
     const cfg = {
       inbound: {

src/web/auto-reply.test.ts

@@ -1,3 +1,10 @@
+// Import test-helpers FIRST to set up mocks before other imports
+import {
+  resetBaileysMocks,
+  resetLoadConfigMock,
+  setLoadConfigMock,
+} from "./test-helpers.js";
+
 import crypto from "node:crypto";
 import fs from "node:fs/promises";
 import os from "node:os";
@@ -6,7 +13,7 @@ import sharp from "sharp";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import type { WarelayConfig } from "../config/config.js";
-import { resolveStorePath } from "../config/sessions.js";
+import * as commandQueue from "../process/command-queue.js";
 import { resetLogger, setLoggerOverride } from "../logging.js";
 import {
   HEARTBEAT_PROMPT,
@@ -18,11 +25,20 @@ import {
   stripHeartbeatToken,
 } from "./auto-reply.js";
 import type { sendMessageWeb } from "./outbound.js";
-import {
+import * as commandQueue from "../process/command-queue.js";
-  resetBaileysMocks,
+import { getQueueSize } from "../process/command-queue.js";
-  resetLoadConfigMock,
+
-  setLoadConfigMock,
+const makeSessionStore = async (
-} from "./test-helpers.js";
+  entries: Record<string, unknown> = {},
+): Promise<{ storePath: string; cleanup: () => Promise<void> }> => {
+  const dir = await fs.mkdtemp(path.join(os.tmpdir(), "warelay-session-"));
+  const storePath = path.join(dir, "sessions.json");
+  await fs.writeFile(storePath, JSON.stringify(entries));
+  return {
+    storePath,
+    cleanup: () => fs.rm(dir, { recursive: true, force: true }),
+  };
+};
 
 describe("heartbeat helpers", () => {
   it("strips heartbeat token and skips when only token", () => {
@@ -78,19 +94,9 @@ describe("heartbeat helpers", () => {
 });
 
 describe("resolveHeartbeatRecipients", () => {
-  const makeStore = async (entries: Record<string, { updatedAt: number }>) => {
-    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "warelay-heartbeat-"));
-    const storePath = path.join(dir, "sessions.json");
-    await fs.writeFile(storePath, JSON.stringify(entries));
-    return {
-      storePath,
-      cleanup: async () => fs.rm(dir, { recursive: true, force: true }),
-    };
-  };
-
   it("returns the sole session recipient", async () => {
     const now = Date.now();
-    const store = await makeStore({ "+1000": { updatedAt: now } });
+    const store = await makeSessionStore({ "+1000": { updatedAt: now } });
     const cfg: WarelayConfig = {
       inbound: {
         allowFrom: ["+1999"],
@@ -105,7 +111,7 @@ describe("resolveHeartbeatRecipients", () => {
 
   it("surfaces ambiguity when multiple sessions exist", async () => {
     const now = Date.now();
-    const store = await makeStore({
+    const store = await makeSessionStore({
       "+1000": { updatedAt: now },
       "+2000": { updatedAt: now - 10 },
     });
@@ -122,7 +128,7 @@ describe("resolveHeartbeatRecipients", () => {
   });
 
   it("filters wildcard allowFrom when no sessions exist", async () => {
-    const store = await makeStore({});
+    const store = await makeSessionStore({});
     const cfg: WarelayConfig = {
       inbound: {
         allowFrom: ["*"],
@@ -137,7 +143,7 @@ describe("resolveHeartbeatRecipients", () => {
 
   it("merges sessions and allowFrom when --all is set", async () => {
     const now = Date.now();
-    const store = await makeStore({ "+1000": { updatedAt: now } });
+    const store = await makeSessionStore({ "+1000": { updatedAt: now } });
     const cfg: WarelayConfig = {
       inbound: {
         allowFrom: ["+1999"],
@@ -153,12 +159,16 @@ describe("resolveHeartbeatRecipients", () => {
 
 describe("runWebHeartbeatOnce", () => {
   it("skips when heartbeat token returned", async () => {
+    const store = await makeSessionStore();
     const sender: typeof sendMessageWeb = vi.fn();
     const resolver = vi.fn(async () => ({ text: HEARTBEAT_TOKEN }));
-    setLoadConfigMock({
-      inbound: { allowFrom: ["+1555"], reply: { mode: "command" } },
-    });
     await runWebHeartbeatOnce({
+      cfg: {
+        inbound: {
+          allowFrom: ["+1555"],
+          reply: { mode: "command", session: { store: store.storePath } },
+        },
+      },
       to: "+1555",
       verbose: false,
       sender,
@@ -166,54 +176,58 @@ describe("runWebHeartbeatOnce", () => {
     });
     expect(resolver).toHaveBeenCalled();
     expect(sender).not.toHaveBeenCalled();
+    await store.cleanup();
   });
 
   it("sends when alert text present", async () => {
+    const store = await makeSessionStore();
     const sender: typeof sendMessageWeb = vi
       .fn()
       .mockResolvedValue({ messageId: "m1", toJid: "jid" });
     const resolver = vi.fn(async () => ({ text: "ALERT" }));
-    setLoadConfigMock({
-      inbound: { allowFrom: ["+1555"], reply: { mode: "command" } },
-    });
     await runWebHeartbeatOnce({
+      cfg: {
+        inbound: {
+          allowFrom: ["+1555"],
+          reply: { mode: "command", session: { store: store.storePath } },
+        },
+      },
       to: "+1555",
       verbose: false,
       sender,
       replyResolver: resolver,
     });
     expect(sender).toHaveBeenCalledWith("+1555", "ALERT", { verbose: false });
+    await store.cleanup();
   });
 
   it("falls back to most recent session when no to is provided", async () => {
+    const store = await makeSessionStore();
+    const storePath = store.storePath;
     const sender: typeof sendMessageWeb = vi
       .fn()
       .mockResolvedValue({ messageId: "m1", toJid: "jid" });
     const resolver = vi.fn(async () => ({ text: "ALERT" }));
-    // Seed session store
     const now = Date.now();
-    const store = {
+    const sessionEntries = {
       "+1222": { sessionId: "s1", updatedAt: now - 1000 },
       "+1333": { sessionId: "s2", updatedAt: now },
     };
-    const storePath = resolveStorePath();
+    await fs.writeFile(storePath, JSON.stringify(sessionEntries));
-    await fs.mkdir(resolveStorePath().replace("sessions.json", ""), {
-      recursive: true,
-    });
-    await fs.writeFile(storePath, JSON.stringify(store));
-    setLoadConfigMock({
-      inbound: {
-        allowFrom: ["+1999"],
-        reply: { mode: "command", session: {} },
-      },
-    });
     await runWebHeartbeatOnce({
+      cfg: {
+        inbound: {
+          allowFrom: ["+1999"],
+          reply: { mode: "command", session: { store: storePath } },
+        },
+      },
       to: "+1999",
       verbose: false,
       sender,
       replyResolver: resolver,
     });
     expect(sender).toHaveBeenCalledWith("+1999", "ALERT", { verbose: false });
+    await store.cleanup();
   });
 
   it("does not refresh updatedAt when heartbeat is skipped", async () => {
@@ -353,14 +367,18 @@ describe("runWebHeartbeatOnce", () => {
   });
 
   it("sends overrideBody directly and skips resolver", async () => {
+    const store = await makeSessionStore();
     const sender: typeof sendMessageWeb = vi
       .fn()
       .mockResolvedValue({ messageId: "m1", toJid: "jid" });
     const resolver = vi.fn();
-    setLoadConfigMock({
-      inbound: { allowFrom: ["+1555"], reply: { mode: "command" } },
-    });
     await runWebHeartbeatOnce({
+      cfg: {
+        inbound: {
+          allowFrom: ["+1555"],
+          reply: { mode: "command", session: { store: store.storePath } },
+        },
+      },
       to: "+1555",
       verbose: false,
       sender,
@@ -371,15 +389,20 @@ describe("runWebHeartbeatOnce", () => {
       verbose: false,
     });
     expect(resolver).not.toHaveBeenCalled();
+    await store.cleanup();
   });
 
   it("dry-run overrideBody prints and skips send", async () => {
+    const store = await makeSessionStore();
     const sender: typeof sendMessageWeb = vi.fn();
     const resolver = vi.fn();
-    setLoadConfigMock({
-      inbound: { allowFrom: ["+1555"], reply: { mode: "command" } },
-    });
     await runWebHeartbeatOnce({
+      cfg: {
+        inbound: {
+          allowFrom: ["+1555"],
+          reply: { mode: "command", session: { store: store.storePath } },
+        },
+      },
       to: "+1555",
       verbose: false,
       sender,
@@ -389,6 +412,7 @@ describe("runWebHeartbeatOnce", () => {
     });
     expect(sender).not.toHaveBeenCalled();
     expect(resolver).not.toHaveBeenCalled();
+    await store.cleanup();
   });
 });
 
@@ -504,6 +528,125 @@ describe("web auto-reply", () => {
     );
   });
 
+  it("skips reply heartbeat when requests are running", async () => {
+    const tmpDir = await fs.mkdtemp(
+      path.join(os.tmpdir(), "warelay-heartbeat-queue-"),
+    );
+    const storePath = path.join(tmpDir, "sessions.json");
+    await fs.writeFile(storePath, JSON.stringify({}));
+
+    const queueSpy = vi
+      .spyOn(commandQueue, "getQueueSize")
+      .mockReturnValue(2);
+    const replyResolver = vi.fn();
+    const listenerFactory = vi.fn(async () => {
+      const onClose = new Promise<void>(() => {
+        // stay open until aborted
+      });
+      return { close: vi.fn(), onClose };
+    });
+    const runtime = { log: vi.fn(), error: vi.fn(), exit: vi.fn() } as never;
+
+    setLoadConfigMock(() => ({
+      inbound: {
+        allowFrom: ["+1555"],
+        reply: { mode: "command", session: { store: storePath } },
+      },
+    }));
+
+    const controller = new AbortController();
+    const run = monitorWebProvider(
+      false,
+      listenerFactory,
+      true,
+      replyResolver,
+      runtime,
+      controller.signal,
+      { replyHeartbeatMinutes: 1, replyHeartbeatNow: true },
+    );
+
+    try {
+      await Promise.resolve();
+      controller.abort();
+      await run;
+      expect(replyResolver).not.toHaveBeenCalled();
+    } finally {
+      queueSpy.mockRestore();
+    }
+  });
+
+  it("batches inbound messages while queue is busy and preserves timestamps", async () => {
+    vi.useFakeTimers();
+    const originalMax = process.getMaxListeners();
+    process.setMaxListeners?.(1); // force low to confirm bump
+
+    const sendMedia = vi.fn();
+    const reply = vi.fn().mockResolvedValue(undefined);
+    const sendComposing = vi.fn();
+    const resolver = vi.fn().mockResolvedValue({ text: "batched" });
+
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    // Queue starts busy, then frees after one polling tick.
+    let queueBusy = true;
+    const queueSpy = vi
+      .spyOn(commandQueue, "getQueueSize")
+      .mockImplementation(() => (queueBusy ? 1 : 0));
+
+    setLoadConfigMock(() => ({ inbound: { timestampPrefix: "UTC" } }));
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    // Two messages from the same sender with fixed timestamps
+    await capturedOnMessage?.({
+      body: "first",
+      from: "+1",
+      to: "+2",
+      id: "m1",
+      timestamp: 1735689600000, // Jan 1 2025 00:00:00 UTC
+      sendComposing,
+      reply,
+      sendMedia,
+    });
+    await capturedOnMessage?.({
+      body: "second",
+      from: "+1",
+      to: "+2",
+      id: "m2",
+      timestamp: 1735693200000, // Jan 1 2025 01:00:00 UTC
+      sendComposing,
+      reply,
+      sendMedia,
+    });
+
+    // Let the queued batch flush once the queue is free
+    queueBusy = false;
+    vi.advanceTimersByTime(200);
+
+    expect(resolver).toHaveBeenCalledTimes(1);
+    const args = resolver.mock.calls[0][0];
+    expect(args.Body).toContain("[Jan 1 00:00] [warelay] first");
+    expect(args.Body).toContain("[Jan 1 01:00] [warelay] second");
+
+    // Max listeners bumped to avoid warnings in multi-instance test runs
+    expect(process.getMaxListeners?.()).toBeGreaterThanOrEqual(50);
+
+    queueSpy.mockRestore();
+    process.setMaxListeners?.(originalMax);
+    vi.useRealTimers();
+  });
+
   it("falls back to text when media send fails", async () => {
     const sendMedia = vi.fn().mockRejectedValue(new Error("boom"));
     const reply = vi.fn().mockResolvedValue(undefined);
@@ -945,4 +1088,213 @@ describe("web auto-reply", () => {
     expect(content).toContain('"module":"web-auto-reply"');
     expect(content).toContain('"text":"auto"');
   });
+
+  it("prefixes body with same-phone marker when from === to", async () => {
+    // Enable messagePrefix for same-phone mode testing
+    setLoadConfigMock(() => ({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: "[same-phone]",
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    }));
+
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    const resolver = vi.fn().mockResolvedValue({ text: "reply" });
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    await capturedOnMessage?.({
+      body: "hello",
+      from: "+1555",
+      to: "+1555", // Same phone!
+      id: "msg1",
+      sendComposing: vi.fn(),
+      reply: vi.fn(),
+      sendMedia: vi.fn(),
+    });
+
+    // The resolver should receive a prefixed body with the configured marker
+    const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
+    expect(callArg?.Body).toBeDefined();
+    expect(callArg?.Body).toBe("[same-phone] hello");
+    resetLoadConfigMock();
+  });
+
+  it("does not prefix body when from !== to", async () => {
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    const resolver = vi.fn().mockResolvedValue({ text: "reply" });
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    await capturedOnMessage?.({
+      body: "hello",
+      from: "+1555",
+      to: "+2666", // Different phones
+      id: "msg1",
+      sendComposing: vi.fn(),
+      reply: vi.fn(),
+      sendMedia: vi.fn(),
+    });
+
+    // Body should NOT be prefixed
+    const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
+    expect(callArg?.Body).toBe("hello");
+  });
+
+  it("applies responsePrefix to regular replies", async () => {
+    setLoadConfigMock(() => ({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: "🦞",
+        timestampPrefix: false,
+      },
+    }));
+
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const reply = vi.fn();
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    const resolver = vi.fn().mockResolvedValue({ text: "hello there" });
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    await capturedOnMessage?.({
+      body: "hi",
+      from: "+1555",
+      to: "+2666",
+      id: "msg1",
+      sendComposing: vi.fn(),
+      reply,
+      sendMedia: vi.fn(),
+    });
+
+    // Reply should have responsePrefix prepended
+    expect(reply).toHaveBeenCalledWith("🦞 hello there");
+    resetLoadConfigMock();
+  });
+
+  it("skips responsePrefix for HEARTBEAT_OK responses", async () => {
+    setLoadConfigMock(() => ({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: "🦞",
+        timestampPrefix: false,
+      },
+    }));
+
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const reply = vi.fn();
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    // Resolver returns exact HEARTBEAT_OK
+    const resolver = vi.fn().mockResolvedValue({ text: HEARTBEAT_TOKEN });
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    await capturedOnMessage?.({
+      body: "test",
+      from: "+1555",
+      to: "+2666",
+      id: "msg1",
+      sendComposing: vi.fn(),
+      reply,
+      sendMedia: vi.fn(),
+    });
+
+    // HEARTBEAT_OK should NOT have prefix - warelay needs exact match
+    expect(reply).toHaveBeenCalledWith(HEARTBEAT_TOKEN);
+    resetLoadConfigMock();
+  });
+
+  it("does not double-prefix if responsePrefix already present", async () => {
+    setLoadConfigMock(() => ({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: "🦞",
+        timestampPrefix: false,
+      },
+    }));
+
+    let capturedOnMessage:
+      | ((msg: import("./inbound.js").WebInboundMessage) => Promise<void>)
+      | undefined;
+    const reply = vi.fn();
+    const listenerFactory = async (opts: {
+      onMessage: (
+        msg: import("./inbound.js").WebInboundMessage,
+      ) => Promise<void>;
+    }) => {
+      capturedOnMessage = opts.onMessage;
+      return { close: vi.fn() };
+    };
+
+    // Resolver returns text that already has prefix
+    const resolver = vi.fn().mockResolvedValue({ text: "🦞 already prefixed" });
+
+    await monitorWebProvider(false, listenerFactory, false, resolver);
+    expect(capturedOnMessage).toBeDefined();
+
+    await capturedOnMessage?.({
+      body: "test",
+      from: "+1555",
+      to: "+2666",
+      id: "msg1",
+      sendComposing: vi.fn(),
+      reply,
+      sendMedia: vi.fn(),
+    });
+
+    // Should not double-prefix
+    expect(reply).toHaveBeenCalledWith("🦞 already prefixed");
+    resetLoadConfigMock();
+  });
 });

src/web/auto-reply.ts

@@ -9,14 +9,16 @@ import {
   resolveStorePath,
   saveSessionStore,
 } from "../config/sessions.js";
-import { danger, isVerbose, logVerbose, success } from "../globals.js";
+import { danger, info, isVerbose, logVerbose, success } from "../globals.js";
 import { logInfo } from "../logger.js";
 import { getChildLogger } from "../logging.js";
 import { defaultRuntime, type RuntimeEnv } from "../runtime.js";
 import { normalizeE164 } from "../utils.js";
 import { monitorWebInbox } from "./inbound.js";
+import { sendViaIpc, startIpcServer, stopIpcServer } from "./ipc.js";
 import { loadWebMedia } from "./media.js";
 import { sendMessageWeb } from "./outbound.js";
+import { getQueueSize } from "../process/command-queue.js";
 import {
   computeBackoff,
   newConnectionId,
@@ -27,6 +29,26 @@ import {
 } from "./reconnect.js";
 import { getWebAuthAgeMs } from "./session.js";
 
+/**
+ * Send a message via IPC if relay is running, otherwise fall back to direct.
+ * This avoids Signal session corruption from multiple Baileys connections.
+ */
+async function sendWithIpcFallback(
+  to: string,
+  message: string,
+  opts: { verbose: boolean; mediaUrl?: string },
+): Promise<{ messageId: string; toJid: string }> {
+  const ipcResult = await sendViaIpc(to, message, opts.mediaUrl);
+  if (ipcResult?.success && ipcResult.messageId) {
+    if (opts.verbose) {
+      console.log(info(`Sent via relay IPC (avoiding session corruption)`));
+    }
+    return { messageId: ipcResult.messageId, toJid: `${to}@s.whatsapp.net` };
+  }
+  // Fall back to direct send
+  return sendMessageWeb(to, message, opts);
+}
+
 const DEFAULT_WEB_MEDIA_BYTES = 5 * 1024 * 1024;
 type WebInboundMsg = Parameters<
   typeof monitorWebInbox
@@ -94,7 +116,7 @@ export async function runWebHeartbeatOnce(opts: {
   } = opts;
   const _runtime = opts.runtime ?? defaultRuntime;
   const replyResolver = opts.replyResolver ?? getReplyFromConfig;
-  const sender = opts.sender ?? sendMessageWeb;
+  const sender = opts.sender ?? sendWithIpcFallback;
   const runId = newConnectionId();
   const heartbeatLogger = getChildLogger({
     module: "web-heartbeat",
@@ -493,6 +515,13 @@ export async function monitorWebProvider(
       }),
     );
 
+  // Avoid noisy MaxListenersExceeded warnings in test environments where
+  // multiple relay instances may be constructed.
+  const currentMaxListeners = process.getMaxListeners?.() ?? 10;
+  if (process.setMaxListeners && currentMaxListeners < 50) {
+    process.setMaxListeners(50);
+  }
+
   let sigintStop = false;
   const handleSigint = () => {
     sigintStop = true;
@@ -501,6 +530,10 @@ export async function monitorWebProvider(
 
   let reconnectAttempts = 0;
 
+  // Track recently sent messages to prevent echo loops
+  const recentlySent = new Set<string>();
+  const MAX_RECENT_MESSAGES = 100;
+
   while (true) {
     if (stopRequested()) break;
 
@@ -508,96 +541,242 @@ export async function monitorWebProvider(
     const startedAt = Date.now();
     let heartbeat: NodeJS.Timeout | null = null;
     let replyHeartbeatTimer: NodeJS.Timeout | null = null;
+    let watchdogTimer: NodeJS.Timeout | null = null;
     let lastMessageAt: number | null = null;
     let handledMessages = 0;
     let lastInboundMsg: WebInboundMsg | null = null;
 
+    // Watchdog to detect stuck message processing (e.g., event emitter died)
+    // Should be significantly longer than heartbeatMinutes to avoid false positives
+    const MESSAGE_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes without any messages
+    const WATCHDOG_CHECK_MS = 60 * 1000; // Check every minute
+
+    // Batch inbound messages while command queue is busy, then send one
+    // combined prompt with per-message timestamps (inbound-only behavior).
+    type PendingBatch = { messages: WebInboundMsg[]; timer?: NodeJS.Timeout };
+    const pendingBatches = new Map<string, PendingBatch>();
+
+    const formatTimestamp = (ts?: number) => {
+      const tsCfg = cfg.inbound?.timestampPrefix;
+      const tsEnabled = tsCfg !== false; // default true
+      if (!tsEnabled) return "";
+      const tz = typeof tsCfg === "string" ? tsCfg : "UTC";
+      const date = ts ? new Date(ts) : new Date();
+      try {
+        return `[${date.toLocaleDateString("en-US", { month: "short", day: "numeric", timeZone: tz })} ${date.toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit", hour12: false, timeZone: tz })}] `;
+      } catch {
+        return `[${date.toISOString().slice(5, 16).replace("T", " ")}] `;
+      }
+    };
+
+    const buildLine = (msg: WebInboundMsg) => {
+      // Build message prefix: explicit config > default based on allowFrom
+      let messagePrefix = cfg.inbound?.messagePrefix;
+      if (messagePrefix === undefined) {
+        const hasAllowFrom = (cfg.inbound?.allowFrom?.length ?? 0) > 0;
+        messagePrefix = hasAllowFrom ? "" : "[warelay]";
+      }
+      const prefixStr = messagePrefix ? `${messagePrefix} ` : "";
+      return `${formatTimestamp(msg.timestamp)}${prefixStr}${msg.body}`;
+    };
+
+    const processBatch = async (from: string) => {
+      const batch = pendingBatches.get(from);
+      if (!batch || batch.messages.length === 0) return;
+      if (getQueueSize() > 0) {
+        // Wait until command queue is free to run the combined prompt.
+        batch.timer = setTimeout(() => void processBatch(from), 150);
+        return;
+      }
+      pendingBatches.delete(from);
+
+      const messages = batch.messages;
+      const latest = messages[messages.length - 1];
+      const combinedBody = messages.map(buildLine).join("\n");
+
+      // Echo detection uses combined body so we don't respond twice.
+      if (recentlySent.has(combinedBody)) {
+        logVerbose(`Skipping auto-reply: detected echo for combined batch`);
+        recentlySent.delete(combinedBody);
+        return;
+      }
+
+      const correlationId = latest.id ?? newConnectionId();
+      replyLogger.info(
+        {
+          connectionId,
+          correlationId,
+          from,
+          to: latest.to,
+          body: combinedBody,
+          mediaType: latest.mediaType ?? null,
+          mediaPath: latest.mediaPath ?? null,
+          batchSize: messages.length,
+        },
+        "inbound web message (batched)",
+      );
+
+      const tsDisplay = latest.timestamp
+        ? new Date(latest.timestamp).toISOString()
+        : new Date().toISOString();
+      console.log(`\n[${tsDisplay}] ${from} -> ${latest.to}: ${combinedBody}`);
+
+      const replyResult = await (replyResolver ?? getReplyFromConfig)(
+        {
+          Body: combinedBody,
+          From: latest.from,
+          To: latest.to,
+          MessageSid: latest.id,
+          MediaPath: latest.mediaPath,
+          MediaUrl: latest.mediaUrl,
+          MediaType: latest.mediaType,
+        },
+        {
+          onReplyStart: latest.sendComposing,
+        },
+      );
+
+      if (
+        !replyResult ||
+        (!replyResult.text &&
+          !replyResult.mediaUrl &&
+          !replyResult.mediaUrls?.length)
+      ) {
+        logVerbose("Skipping auto-reply: no text/media returned from resolver");
+        return;
+      }
+
+      // Apply response prefix if configured (skip for HEARTBEAT_OK to preserve exact match)
+      const responsePrefix = cfg.inbound?.responsePrefix;
+      if (
+        responsePrefix &&
+        replyResult.text &&
+        replyResult.text.trim() !== HEARTBEAT_TOKEN
+      ) {
+        if (!replyResult.text.startsWith(responsePrefix)) {
+          replyResult.text = `${responsePrefix} ${replyResult.text}`;
+        }
+      }
+
+      try {
+        await deliverWebReply({
+          replyResult,
+          msg: latest,
+          maxMediaBytes,
+          replyLogger,
+          runtime,
+          connectionId,
+        });
+
+        if (replyResult.text) {
+          recentlySent.add(replyResult.text);
+          recentlySent.add(combinedBody); // Prevent echo on the batch text itself
+          logVerbose(
+            `Added to echo detection set (size now: ${recentlySent.size}): ${replyResult.text.substring(0, 50)}...`,
+          );
+          if (recentlySent.size > MAX_RECENT_MESSAGES) {
+            const firstKey = recentlySent.values().next().value;
+            if (firstKey) recentlySent.delete(firstKey);
+          }
+        }
+
+        if (isVerbose()) {
+          console.log(
+            success(
+              `↩️  Auto-replied to ${from} (web${replyResult.mediaUrl || replyResult.mediaUrls?.length ? ", media" : ""}; batched ${messages.length})`,
+            ),
+          );
+        } else {
+          console.log(
+            success(
+              `↩️  ${replyResult.text ?? "<media>"}${replyResult.mediaUrl || replyResult.mediaUrls?.length ? " (media)" : ""}`,
+            ),
+          );
+        }
+      } catch (err) {
+        console.error(
+          danger(
+            `Failed sending web auto-reply to ${from}: ${String(err)}`,
+          ),
+        );
+      }
+    };
+
+    const enqueueBatch = async (msg: WebInboundMsg) => {
+      const bucket = pendingBatches.get(msg.from) ?? { messages: [] };
+      bucket.messages.push(msg);
+      pendingBatches.set(msg.from, bucket);
+
+      // Process immediately when queue is free; otherwise wait until it drains.
+      if (getQueueSize() === 0) {
+        await processBatch(msg.from);
+      } else {
+        bucket.timer = bucket.timer ?? setTimeout(() => void processBatch(msg.from), 150);
+      }
+    };
+
     const listener = await (listenerFactory ?? monitorWebInbox)({
       verbose,
       onMessage: async (msg) => {
         handledMessages += 1;
         lastMessageAt = Date.now();
-        const ts = msg.timestamp
-          ? new Date(msg.timestamp).toISOString()
-          : new Date().toISOString();
-        const correlationId = msg.id ?? newConnectionId();
-        replyLogger.info(
-          {
-            connectionId,
-            correlationId,
-            from: msg.from,
-            to: msg.to,
-            body: msg.body,
-            mediaType: msg.mediaType ?? null,
-            mediaPath: msg.mediaPath ?? null,
-          },
-          "inbound web message",
-        );
-
-        console.log(`\n[${ts}] ${msg.from} -> ${msg.to}: ${msg.body}`);
-
         lastInboundMsg = msg;
 
-        const replyResult = await (replyResolver ?? getReplyFromConfig)(
+        // Same-phone mode logging retained
-          {
+        if (msg.from === msg.to) {
-            Body: msg.body,
+          logVerbose(`📱 Same-phone mode detected (from === to: ${msg.from})`);
-            From: msg.from,
+        }
-            To: msg.to,
+
-            MessageSid: msg.id,
+        // Skip if this is a message we just sent (echo detection)
-            MediaPath: msg.mediaPath,
+        if (recentlySent.has(msg.body)) {
-            MediaUrl: msg.mediaUrl,
+          console.log(`⏭️  Skipping echo: detected recently sent message`);
-            MediaType: msg.mediaType,
-          },
-          {
-            onReplyStart: msg.sendComposing,
-          },
-        );
-        if (
-          !replyResult ||
-          (!replyResult.text &&
-            !replyResult.mediaUrl &&
-            !replyResult.mediaUrls?.length)
-        ) {
           logVerbose(
-            "Skipping auto-reply: no text/media returned from resolver",
+            `Skipping auto-reply: detected echo (message matches recently sent text)`,
           );
+          recentlySent.delete(msg.body);
           return;
         }
-        try {
+
-          await deliverWebReply({
+        return enqueueBatch(msg);
-            replyResult,
-            msg,
-            maxMediaBytes,
-            replyLogger,
-            runtime,
-            connectionId,
-          });
-          if (isVerbose()) {
-            console.log(
-              success(
-                `↩️  Auto-replied to ${msg.from} (web${replyResult.mediaUrl || replyResult.mediaUrls?.length ? ", media" : ""})`,
-              ),
-            );
-          } else {
-            console.log(
-              success(
-                `↩️  ${replyResult.text ?? "<media>"}${replyResult.mediaUrl || replyResult.mediaUrls?.length ? " (media)" : ""}`,
-              ),
-            );
-          }
-        } catch (err) {
-          console.error(
-            danger(
-              `Failed sending web auto-reply to ${msg.from}: ${String(err)}`,
-            ),
-          );
-        }
       },
     });
 
+    // Start IPC server so `warelay send` can use this connection
+    // instead of creating a new one (which would corrupt Signal session)
+    if ("sendMessage" in listener && "sendComposingTo" in listener) {
+      startIpcServer(async (to, message, mediaUrl) => {
+        let mediaBuffer: Buffer | undefined;
+        let mediaType: string | undefined;
+        if (mediaUrl) {
+          const media = await loadWebMedia(mediaUrl);
+          mediaBuffer = media.buffer;
+          mediaType = media.contentType;
+        }
+        const result = await listener.sendMessage(to, message, mediaBuffer, mediaType);
+        // Add to echo detection so we don't process our own message
+        if (message) {
+          recentlySent.add(message);
+          if (recentlySent.size > MAX_RECENT_MESSAGES) {
+            const firstKey = recentlySent.values().next().value;
+            if (firstKey) recentlySent.delete(firstKey);
+          }
+        }
+        logInfo(`📤 IPC send to ${to}: ${message.substring(0, 50)}...`, runtime);
+        // Show typing indicator after send so user knows more may be coming
+        try {
+          await listener.sendComposingTo(to);
+        } catch {
+          // Ignore typing indicator errors - not critical
+        }
+        return result;
+      });
+    }
+
     const closeListener = async () => {
+      stopIpcServer();
       if (heartbeat) clearInterval(heartbeat);
       if (replyHeartbeatTimer) clearInterval(replyHeartbeatTimer);
+      if (watchdogTimer) clearInterval(watchdogTimer);
       try {
         await listener.close();
       } catch (err) {
@@ -608,21 +787,64 @@ export async function monitorWebProvider(
     if (keepAlive) {
       heartbeat = setInterval(() => {
         const authAgeMs = getWebAuthAgeMs();
-        heartbeatLogger.info(
+        const minutesSinceLastMessage = lastMessageAt
-          {
+          ? Math.floor((Date.now() - lastMessageAt) / 60000)
-            connectionId,
+          : null;
-            reconnectAttempts,
+
-            messagesHandled: handledMessages,
+        const logData = {
-            lastMessageAt,
+          connectionId,
-            authAgeMs,
+          reconnectAttempts,
-            uptimeMs: Date.now() - startedAt,
+          messagesHandled: handledMessages,
-          },
+          lastMessageAt,
-          "web relay heartbeat",
+          authAgeMs,
-        );
+          uptimeMs: Date.now() - startedAt,
+          ...(minutesSinceLastMessage !== null && minutesSinceLastMessage > 30
+            ? { minutesSinceLastMessage }
+            : {}),
+        };
+
+        // Warn if no messages in 30+ minutes
+        if (minutesSinceLastMessage && minutesSinceLastMessage > 30) {
+          heartbeatLogger.warn(logData, "⚠️ web relay heartbeat - no messages in 30+ minutes");
+        } else {
+          heartbeatLogger.info(logData, "web relay heartbeat");
+        }
       }, heartbeatSeconds * 1000);
+
+      // Watchdog: Auto-restart if no messages received for MESSAGE_TIMEOUT_MS
+      watchdogTimer = setInterval(() => {
+        if (lastMessageAt) {
+          const timeSinceLastMessage = Date.now() - lastMessageAt;
+          if (timeSinceLastMessage > MESSAGE_TIMEOUT_MS) {
+            const minutesSinceLastMessage = Math.floor(timeSinceLastMessage / 60000);
+            heartbeatLogger.warn(
+              {
+                connectionId,
+                minutesSinceLastMessage,
+                lastMessageAt: new Date(lastMessageAt),
+                messagesHandled: handledMessages,
+              },
+              "Message timeout detected - forcing reconnect",
+            );
+            console.error(
+              `⚠️  No messages received in ${minutesSinceLastMessage}m - restarting connection`,
+            );
+            closeListener(); // Trigger reconnect
+          }
+        }
+      }, WATCHDOG_CHECK_MS);
     }
 
     const runReplyHeartbeat = async () => {
+      const queued = getQueueSize();
+      if (queued > 0) {
+        heartbeatLogger.info(
+          { connectionId, reason: "requests-in-flight", queued },
+          "reply heartbeat skipped",
+        );
+        console.log(success("heartbeat: skipped (requests in flight)"));
+        return;
+      }
       if (!replyHeartbeatMinutes) return;
       const tickStart = Date.now();
       if (!lastInboundMsg) {
@@ -746,9 +968,16 @@ export async function monitorWebProvider(
           return;
         }
 
+        // Apply response prefix if configured (same as regular messages)
+        let finalText = stripped.text;
+        const responsePrefix = cfg.inbound?.responsePrefix;
+        if (responsePrefix && finalText && !finalText.startsWith(responsePrefix)) {
+          finalText = `${responsePrefix} ${finalText}`;
+        }
+
         const cleanedReply: ReplyPayload = {
           ...replyResult,
-          text: stripped.text,
+          text: finalText,
         };
 
         await deliverWebReply({

src/web/inbound.media.test.ts

@@ -5,6 +5,17 @@ import path from "node:path";
 
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 
+vi.mock("../config/config.js", () => ({
+  loadConfig: vi.fn().mockReturnValue({
+    inbound: {
+      allowFrom: ["*"], // Allow all in tests
+      messagePrefix: undefined,
+      responsePrefix: undefined,
+      timestampPrefix: false,
+    },
+  }),
+}));
+
 const HOME = path.join(
   os.tmpdir(),
   `warelay-inbound-media-${crypto.randomUUID()}`,

src/web/inbound.ts

@@ -8,10 +8,11 @@ import {
   downloadMediaMessage,
 } from "@whiskeysockets/baileys";
 
+import { loadConfig } from "../config/config.js";
 import { isVerbose, logVerbose } from "../globals.js";
 import { getChildLogger } from "../logging.js";
 import { saveMediaBuffer } from "../media/store.js";
-import { jidToE164 } from "../utils.js";
+import { jidToE164, normalizeE164 } from "../utils.js";
 import {
   createWaSocket,
   getStatusCode,
@@ -70,7 +71,7 @@ export async function monitorWebInbox(options: {
       // De-dupe on message id; Baileys can emit retries.
       if (id && seen.has(id)) continue;
       if (id) seen.add(id);
-      if (msg.key?.fromMe) continue;
+      // Note: not filtering fromMe here - echo detection happens in auto-reply layer
       const remoteJid = msg.key?.remoteJid;
       if (!remoteJid) continue;
       // Ignore status/broadcast traffic; we only care about direct chats.
@@ -94,6 +95,20 @@ export async function monitorWebInbox(options: {
       }
       const from = jidToE164(remoteJid);
       if (!from) continue;
+
+      // Filter unauthorized senders early to prevent wasted processing
+      // and potential session corruption from Bad MAC errors
+      const cfg = loadConfig();
+      const allowFrom = cfg.inbound?.allowFrom;
+      const isSamePhone = from === selfE164;
+
+      if (!isSamePhone && Array.isArray(allowFrom) && allowFrom.length > 0) {
+        if (!allowFrom.includes("*") && !allowFrom.map(normalizeE164).includes(from)) {
+          logVerbose(`Blocked unauthorized sender ${from} (not in allowFrom list)`);
+          continue; // Skip processing entirely
+        }
+      }
+
       let body = extractText(msg.message ?? undefined);
       if (!body) {
         body = extractMediaPlaceholder(msg.message ?? undefined);
@@ -197,6 +212,59 @@ export async function monitorWebInbox(options: {
       }
     },
     onClose,
+    /**
+     * Send a message through this connection's socket.
+     * Used by IPC to avoid creating new connections.
+     */
+    sendMessage: async (
+      to: string,
+      text: string,
+      mediaBuffer?: Buffer,
+      mediaType?: string,
+    ): Promise<{ messageId: string }> => {
+      const jid = `${to.replace(/^\+/, "")}@s.whatsapp.net`;
+      let payload: AnyMessageContent;
+      if (mediaBuffer && mediaType) {
+        if (mediaType.startsWith("image/")) {
+          payload = {
+            image: mediaBuffer,
+            caption: text || undefined,
+            mimetype: mediaType,
+          };
+        } else if (mediaType.startsWith("audio/")) {
+          payload = {
+            audio: mediaBuffer,
+            ptt: true,
+            mimetype: mediaType,
+          };
+        } else if (mediaType.startsWith("video/")) {
+          payload = {
+            video: mediaBuffer,
+            caption: text || undefined,
+            mimetype: mediaType,
+          };
+        } else {
+          payload = {
+            document: mediaBuffer,
+            fileName: "file",
+            caption: text || undefined,
+            mimetype: mediaType,
+          };
+        }
+      } else {
+        payload = { text };
+      }
+      const result = await sock.sendMessage(jid, payload);
+      return { messageId: result?.key?.id ?? "unknown" };
+    },
+    /**
+     * Send typing indicator ("composing") to a chat.
+     * Used after IPC send to show more messages are coming.
+     */
+    sendComposingTo: async (to: string): Promise<void> => {
+      const jid = `${to.replace(/^\+/, "")}@s.whatsapp.net`;
+      await sock.sendPresenceUpdate("composing", jid);
+    },
   } as const;
 }
 

src/web/ipc.ts

@@ -0,0 +1,225 @@
+/**
+ * IPC server for warelay relay.
+ *
+ * When the relay is running, it starts a Unix socket server that allows
+ * `warelay send` and `warelay heartbeat` to send messages through the
+ * existing WhatsApp connection instead of creating new ones.
+ *
+ * This prevents Signal session ratchet corruption from multiple connections.
+ */
+
+import fs from "node:fs";
+import net from "node:net";
+import os from "node:os";
+import path from "node:path";
+
+import { getChildLogger } from "../logging.js";
+
+const SOCKET_PATH = path.join(os.homedir(), ".warelay", "relay.sock");
+
+export interface IpcSendRequest {
+  type: "send";
+  to: string;
+  message: string;
+  mediaUrl?: string;
+}
+
+export interface IpcSendResponse {
+  success: boolean;
+  messageId?: string;
+  error?: string;
+}
+
+type SendHandler = (
+  to: string,
+  message: string,
+  mediaUrl?: string,
+) => Promise<{ messageId: string }>;
+
+let server: net.Server | null = null;
+
+/**
+ * Start the IPC server. Called by the relay when it starts.
+ */
+export function startIpcServer(sendHandler: SendHandler): void {
+  const logger = getChildLogger({ module: "ipc-server" });
+
+  // Clean up stale socket file
+  try {
+    fs.unlinkSync(SOCKET_PATH);
+  } catch {
+    // Ignore if doesn't exist
+  }
+
+  server = net.createServer((conn) => {
+    let buffer = "";
+
+    conn.on("data", async (data) => {
+      buffer += data.toString();
+
+      // Try to parse complete JSON messages (newline-delimited)
+      const lines = buffer.split("\n");
+      buffer = lines.pop() ?? ""; // Keep incomplete line in buffer
+
+      for (const line of lines) {
+        if (!line.trim()) continue;
+
+        try {
+          const request = JSON.parse(line) as IpcSendRequest;
+
+          if (request.type === "send") {
+            try {
+              const result = await sendHandler(
+                request.to,
+                request.message,
+                request.mediaUrl,
+              );
+              const response: IpcSendResponse = {
+                success: true,
+                messageId: result.messageId,
+              };
+              conn.write(JSON.stringify(response) + "\n");
+            } catch (err) {
+              const response: IpcSendResponse = {
+                success: false,
+                error: String(err),
+              };
+              conn.write(JSON.stringify(response) + "\n");
+            }
+          }
+        } catch (err) {
+          logger.warn({ error: String(err) }, "failed to parse IPC request");
+          const response: IpcSendResponse = {
+            success: false,
+            error: "Invalid request format",
+          };
+          conn.write(JSON.stringify(response) + "\n");
+        }
+      }
+    });
+
+    conn.on("error", (err) => {
+      logger.debug({ error: String(err) }, "IPC connection error");
+    });
+  });
+
+  server.listen(SOCKET_PATH, () => {
+    logger.info({ socketPath: SOCKET_PATH }, "IPC server started");
+    // Make socket accessible
+    fs.chmodSync(SOCKET_PATH, 0o600);
+  });
+
+  server.on("error", (err) => {
+    logger.error({ error: String(err) }, "IPC server error");
+  });
+}
+
+/**
+ * Stop the IPC server. Called when relay shuts down.
+ */
+export function stopIpcServer(): void {
+  if (server) {
+    server.close();
+    server = null;
+  }
+  try {
+    fs.unlinkSync(SOCKET_PATH);
+  } catch {
+    // Ignore
+  }
+}
+
+/**
+ * Check if the relay IPC server is running.
+ */
+export function isRelayRunning(): boolean {
+  try {
+    fs.accessSync(SOCKET_PATH);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Send a message through the running relay's IPC.
+ * Returns null if relay is not running.
+ */
+export async function sendViaIpc(
+  to: string,
+  message: string,
+  mediaUrl?: string,
+): Promise<IpcSendResponse | null> {
+  if (!isRelayRunning()) {
+    return null;
+  }
+
+  return new Promise((resolve) => {
+    const client = net.createConnection(SOCKET_PATH);
+    let buffer = "";
+    let resolved = false;
+
+    const timeout = setTimeout(() => {
+      if (!resolved) {
+        resolved = true;
+        client.destroy();
+        resolve({ success: false, error: "IPC timeout" });
+      }
+    }, 30000); // 30 second timeout
+
+    client.on("connect", () => {
+      const request: IpcSendRequest = {
+        type: "send",
+        to,
+        message,
+        mediaUrl,
+      };
+      client.write(JSON.stringify(request) + "\n");
+    });
+
+    client.on("data", (data) => {
+      buffer += data.toString();
+      const lines = buffer.split("\n");
+
+      for (const line of lines) {
+        if (!line.trim()) continue;
+        try {
+          const response = JSON.parse(line) as IpcSendResponse;
+          if (!resolved) {
+            resolved = true;
+            clearTimeout(timeout);
+            client.end();
+            resolve(response);
+          }
+          return;
+        } catch {
+          // Keep reading
+        }
+      }
+    });
+
+    client.on("error", (err) => {
+      if (!resolved) {
+        resolved = true;
+        clearTimeout(timeout);
+        // Socket exists but can't connect - relay might have crashed
+        resolve(null);
+      }
+    });
+
+    client.on("close", () => {
+      if (!resolved) {
+        resolved = true;
+        clearTimeout(timeout);
+        resolve({ success: false, error: "Connection closed" });
+      }
+    });
+  });
+}
+
+/**
+ * Get the IPC socket path for debugging/status.
+ */
+export function getSocketPath(): string {
+  return SOCKET_PATH;
+}

src/web/monitor-inbox.test.ts

@@ -9,6 +9,19 @@ vi.mock("../media/store.js", () => ({
   }),
 }));
 
+const mockLoadConfig = vi.fn().mockReturnValue({
+  inbound: {
+    allowFrom: ["*"], // Allow all in tests
+    messagePrefix: undefined,
+    responsePrefix: undefined,
+    timestampPrefix: false,
+  },
+});
+
+vi.mock("../config/config.js", () => ({
+  loadConfig: () => mockLoadConfig(),
+}));
+
 vi.mock("./session.js", () => {
   const { EventEmitter } = require("node:events");
   const ev = new EventEmitter();
@@ -216,4 +229,146 @@ describe("web monitor inbox", () => {
     ]);
     await listener.close();
   });
+
+  it("blocks messages from unauthorized senders not in allowFrom", async () => {
+    // Test for auto-recovery fix: early allowFrom filtering prevents Bad MAC errors
+    // from unauthorized senders corrupting sessions
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["+111"], // Only allow +111
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    const onMessage = vi.fn();
+    const listener = await monitorWebInbox({ verbose: false, onMessage });
+    const sock = await createWaSocket();
+
+    // Message from unauthorized sender +999 (not in allowFrom)
+    const upsert = {
+      type: "notify",
+      messages: [
+        {
+          key: { id: "unauth1", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "unauthorized message" },
+          messageTimestamp: 1_700_000_000,
+        },
+      ],
+    };
+
+    sock.ev.emit("messages.upsert", upsert);
+    await new Promise((resolve) => setImmediate(resolve));
+
+    // Should NOT call onMessage for unauthorized senders
+    expect(onMessage).not.toHaveBeenCalled();
+
+    // Reset mock for other tests
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    await listener.close();
+  });
+
+  it("allows messages from senders in allowFrom list", async () => {
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["+111", "+999"], // Allow +999
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    const onMessage = vi.fn();
+    const listener = await monitorWebInbox({ verbose: false, onMessage });
+    const sock = await createWaSocket();
+
+    const upsert = {
+      type: "notify",
+      messages: [
+        {
+          key: { id: "auth1", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "authorized message" },
+          messageTimestamp: 1_700_000_000,
+        },
+      ],
+    };
+
+    sock.ev.emit("messages.upsert", upsert);
+    await new Promise((resolve) => setImmediate(resolve));
+
+    // Should call onMessage for authorized senders
+    expect(onMessage).toHaveBeenCalledWith(
+      expect.objectContaining({ body: "authorized message", from: "+999" }),
+    );
+
+    // Reset mock for other tests
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    await listener.close();
+  });
+
+  it("allows same-phone messages even if not in allowFrom", async () => {
+    // Same-phone mode: when from === selfJid, should always be allowed
+    // This allows users to message themselves even with restrictive allowFrom
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["+111"], // Only allow +111, but self is +123
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    const onMessage = vi.fn();
+    const listener = await monitorWebInbox({ verbose: false, onMessage });
+    const sock = await createWaSocket();
+
+    // Message from self (sock.user.id is "123@s.whatsapp.net" in mock)
+    const upsert = {
+      type: "notify",
+      messages: [
+        {
+          key: { id: "self1", fromMe: false, remoteJid: "123@s.whatsapp.net" },
+          message: { conversation: "self message" },
+          messageTimestamp: 1_700_000_000,
+        },
+      ],
+    };
+
+    sock.ev.emit("messages.upsert", upsert);
+    await new Promise((resolve) => setImmediate(resolve));
+
+    // Should allow self-messages even if not in allowFrom
+    expect(onMessage).toHaveBeenCalledWith(
+      expect.objectContaining({ body: "self message", from: "+123" }),
+    );
+
+    // Reset mock for other tests
+    mockLoadConfig.mockReturnValue({
+      inbound: {
+        allowFrom: ["*"],
+        messagePrefix: undefined,
+        responsePrefix: undefined,
+        timestampPrefix: false,
+      },
+    });
+
+    await listener.close();
+  });
 });

src/web/test-helpers.ts

@@ -3,18 +3,37 @@ import { vi } from "vitest";
 import type { MockBaileysSocket } from "../../test/mocks/baileys.js";
 import { createMockBaileys } from "../../test/mocks/baileys.js";
 
-let loadConfigMock: () => unknown = () => ({});
+// Use globalThis to store the mock config so it survives vi.mock hoisting
+const CONFIG_KEY = Symbol.for("warelay:testConfigMock");
+const DEFAULT_CONFIG = {
+  inbound: {
+    allowFrom: ["*"], // Allow all in tests by default
+    messagePrefix: undefined, // No message prefix in tests
+    responsePrefix: undefined, // No response prefix in tests
+    timestampPrefix: false, // No timestamp in tests
+  },
+};
 
-export function setLoadConfigMock(fn: () => unknown) {
+// Initialize default if not set
-  loadConfigMock = fn;
+if (!(globalThis as Record<symbol, unknown>)[CONFIG_KEY]) {
+  (globalThis as Record<symbol, unknown>)[CONFIG_KEY] = () => DEFAULT_CONFIG;
+}
+
+export function setLoadConfigMock(fn: (() => unknown) | unknown) {
+  (globalThis as Record<symbol, unknown>)[CONFIG_KEY] =
+    typeof fn === "function" ? fn : () => fn;
 }
 
 export function resetLoadConfigMock() {
-  loadConfigMock = () => ({});
+  (globalThis as Record<symbol, unknown>)[CONFIG_KEY] = () => DEFAULT_CONFIG;
 }
 
 vi.mock("../config/config.js", () => ({
-  loadConfig: () => loadConfigMock(),
+  loadConfig: () => {
+    const getter = (globalThis as Record<symbol, unknown>)[CONFIG_KEY];
+    if (typeof getter === "function") return getter();
+    return DEFAULT_CONFIG;
+  },
 }));
 
 vi.mock("../media/store.js", () => ({