fix(routing): unify session delivery invariants for duplicate suppression (#33786)

* Routing: unify session delivery invariants

* Routing: address PR review feedback

* Routing: tighten topic and session-scope suppression

* fix(chat): inherit routes for per-account channel-peer sessions

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

@@ -1,5 +1,8 @@
 import { describe, expect, it } from "vitest";
-import { filterMessagingToolMediaDuplicates } from "./reply-payloads.js";
+import {
+  filterMessagingToolMediaDuplicates,
+  shouldSuppressMessagingToolReplies,
+} from "./reply-payloads.js";
 
 describe("filterMessagingToolMediaDuplicates", () => {
   it("strips mediaUrl when it matches sentMediaUrls", () => {
@@ -75,3 +78,79 @@ describe("filterMessagingToolMediaDuplicates", () => {
     expect(result).toEqual([{ text: "hello", mediaUrl: undefined, mediaUrls: undefined }]);
   });
 });
+
+describe("shouldSuppressMessagingToolReplies", () => {
+  it("suppresses when target provider is missing but target matches current provider route", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "123",
+        messagingToolSentTargets: [{ tool: "message", provider: "", to: "123" }],
+      }),
+    ).toBe(true);
+  });
+
+  it('suppresses when target provider uses "message" placeholder and target matches', () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "123",
+        messagingToolSentTargets: [{ tool: "message", provider: "message", to: "123" }],
+      }),
+    ).toBe(true);
+  });
+
+  it("does not suppress when providerless target does not match origin route", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "123",
+        messagingToolSentTargets: [{ tool: "message", provider: "", to: "456" }],
+      }),
+    ).toBe(false);
+  });
+
+  it("suppresses telegram topic-origin replies when explicit threadId matches", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "telegram:group:-100123:topic:77",
+        messagingToolSentTargets: [
+          { tool: "message", provider: "telegram", to: "-100123", threadId: "77" },
+        ],
+      }),
+    ).toBe(true);
+  });
+
+  it("does not suppress telegram topic-origin replies when explicit threadId differs", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "telegram:group:-100123:topic:77",
+        messagingToolSentTargets: [
+          { tool: "message", provider: "telegram", to: "-100123", threadId: "88" },
+        ],
+      }),
+    ).toBe(false);
+  });
+
+  it("does not suppress telegram topic-origin replies when target omits topic metadata", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "telegram:group:-100123:topic:77",
+        messagingToolSentTargets: [{ tool: "message", provider: "telegram", to: "-100123" }],
+      }),
+    ).toBe(false);
+  });
+
+  it("suppresses telegram replies when chatId matches but target forms differ", () => {
+    expect(
+      shouldSuppressMessagingToolReplies({
+        messageProvider: "telegram",
+        originatingTo: "telegram:group:-100123",
+        messagingToolSentTargets: [{ tool: "message", provider: "telegram", to: "-100123" }],
+      }),
+    ).toBe(true);
+  });
+});

src/auto-reply/reply/reply-payloads.ts

@@ -4,6 +4,7 @@ import { normalizeChannelId } from "../../channels/plugins/index.js";
 import type { ReplyToMode } from "../../config/types.js";
 import { normalizeTargetForProvider } from "../../infra/outbound/target-normalization.js";
 import { normalizeOptionalAccountId } from "../../routing/account-id.js";
+import { parseTelegramTarget } from "../../telegram/targets.js";
 import type { OriginatingChannelType } from "../templating.js";
 import type { ReplyPayload } from "../types.js";
 import { extractReplyToTag } from "./reply-tags.js";
@@ -162,6 +163,62 @@ function normalizeProviderForComparison(value?: string): string | undefined {
   return PROVIDER_ALIAS_MAP[lowered] ?? lowered;
 }
 
+function normalizeThreadIdForComparison(value?: string): string | undefined {
+  const trimmed = value?.trim();
+  if (!trimmed) {
+    return undefined;
+  }
+  if (/^-?\d+$/.test(trimmed)) {
+    return String(Number.parseInt(trimmed, 10));
+  }
+  return trimmed.toLowerCase();
+}
+
+function resolveTargetProviderForComparison(params: {
+  currentProvider: string;
+  targetProvider?: string;
+}): string {
+  const targetProvider = normalizeProviderForComparison(params.targetProvider);
+  if (!targetProvider || targetProvider === "message") {
+    return params.currentProvider;
+  }
+  return targetProvider;
+}
+
+function targetsMatchForSuppression(params: {
+  provider: string;
+  originTarget: string;
+  targetKey: string;
+  targetThreadId?: string;
+}): boolean {
+  if (params.provider !== "telegram") {
+    return params.targetKey === params.originTarget;
+  }
+
+  const origin = parseTelegramTarget(params.originTarget);
+  const target = parseTelegramTarget(params.targetKey);
+  const explicitTargetThreadId = normalizeThreadIdForComparison(params.targetThreadId);
+  const targetThreadId =
+    explicitTargetThreadId ??
+    (target.messageThreadId != null ? String(target.messageThreadId) : undefined);
+  const originThreadId =
+    origin.messageThreadId != null ? String(origin.messageThreadId) : undefined;
+  if (origin.chatId.trim().toLowerCase() !== target.chatId.trim().toLowerCase()) {
+    return false;
+  }
+  if (originThreadId && targetThreadId != null) {
+    return originThreadId === targetThreadId;
+  }
+  if (originThreadId && targetThreadId == null) {
+    return false;
+  }
+  if (!originThreadId && targetThreadId != null) {
+    return false;
+  }
+  // chatId already matched and neither side carries thread context.
+  return true;
+}
+
 export function shouldSuppressMessagingToolReplies(params: {
   messageProvider?: string;
   messagingToolSentTargets?: MessagingToolSend[];
@@ -182,16 +239,14 @@ export function shouldSuppressMessagingToolReplies(params: {
     return false;
   }
   return sentTargets.some((target) => {
-    const targetProvider = normalizeProviderForComparison(target?.provider);
-    if (!targetProvider) {
+    const targetProvider = resolveTargetProviderForComparison({
+      currentProvider: provider,
+      targetProvider: target?.provider,
+    });
+    if (targetProvider !== provider) {
       return false;
     }
-    const isGenericMessageProvider = targetProvider === "message";
-    if (!isGenericMessageProvider && targetProvider !== provider) {
-      return false;
-    }
-    const targetNormalizationProvider = isGenericMessageProvider ? provider : targetProvider;
-    const targetKey = normalizeTargetForProvider(targetNormalizationProvider, target.to);
+    const targetKey = normalizeTargetForProvider(targetProvider, target.to);
     if (!targetKey) {
       return false;
     }
@@ -199,6 +254,11 @@ export function shouldSuppressMessagingToolReplies(params: {
     if (originAccount && targetAccount && originAccount !== targetAccount) {
       return false;
     }
-    return targetKey === originTarget;
+    return targetsMatchForSuppression({
+      provider,
+      originTarget,
+      targetKey,
+      targetThreadId: target.threadId,
+    });
   });
 }

src/auto-reply/reply/session-delivery.ts

@@ -30,6 +30,14 @@ function resolveSessionKeyChannelHint(sessionKey?: string): string | undefined {
   return normalizeMessageChannel(head);
 }
 
+function isMainSessionKey(sessionKey?: string): boolean {
+  const parsed = parseAgentSessionKey(sessionKey);
+  if (!parsed) {
+    return (sessionKey ?? "").trim().toLowerCase() === "main";
+  }
+  return parsed.rest.trim().toLowerCase() === "main";
+}
+
 function isExternalRoutingChannel(channel?: string): channel is string {
   return Boolean(
     channel && channel !== INTERNAL_MESSAGE_CHANNEL && isDeliverableMessageChannel(channel),
@@ -42,6 +50,9 @@ export function resolveLastChannelRaw(params: {
   sessionKey?: string;
 }): string | undefined {
   const originatingChannel = normalizeMessageChannel(params.originatingChannelRaw);
+  if (originatingChannel === INTERNAL_MESSAGE_CHANNEL && isMainSessionKey(params.sessionKey)) {
+    return params.originatingChannelRaw;
+  }
   const persistedChannel = normalizeMessageChannel(params.persistedLastChannel);
   const sessionKeyChannelHint = resolveSessionKeyChannelHint(params.sessionKey);
   let resolved = params.originatingChannelRaw || params.persistedLastChannel;
@@ -66,6 +77,9 @@ export function resolveLastToRaw(params: {
   sessionKey?: string;
 }): string | undefined {
   const originatingChannel = normalizeMessageChannel(params.originatingChannelRaw);
+  if (originatingChannel === INTERNAL_MESSAGE_CHANNEL && isMainSessionKey(params.sessionKey)) {
+    return params.originatingToRaw || params.toRaw;
+  }
   const persistedChannel = normalizeMessageChannel(params.persistedLastChannel);
   const sessionKeyChannelHint = resolveSessionKeyChannelHint(params.sessionKey);
 

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

@@ -1609,4 +1609,69 @@ describe("initSessionState internal channel routing preservation", () => {
 
     expect(result.sessionEntry.lastChannel).toBe("webchat");
   });
+
+  it("does not reuse stale external lastTo for webchat/main turns without destination", async () => {
+    const storePath = await createStorePath("webchat-main-no-stale-lastto-");
+    const sessionKey = "agent:main:main";
+    await writeSessionStoreFast(storePath, {
+      [sessionKey]: {
+        sessionId: "sess-webchat-main-1",
+        updatedAt: Date.now(),
+        lastChannel: "whatsapp",
+        lastTo: "+15555550123",
+        deliveryContext: {
+          channel: "whatsapp",
+          to: "+15555550123",
+        },
+      },
+    });
+    const cfg = { session: { store: storePath } } as OpenClawConfig;
+
+    const result = await initSessionState({
+      ctx: {
+        Body: "webchat follow-up",
+        SessionKey: sessionKey,
+        OriginatingChannel: "webchat",
+      },
+      cfg,
+      commandAuthorized: true,
+    });
+
+    expect(result.sessionEntry.lastChannel).toBe("webchat");
+    expect(result.sessionEntry.lastTo).toBeUndefined();
+  });
+
+  it("prefers webchat route over persisted external route for main session turns", async () => {
+    const storePath = await createStorePath("prefer-webchat-main-route-");
+    const sessionKey = "agent:main:main";
+    await writeSessionStoreFast(storePath, {
+      [sessionKey]: {
+        sessionId: "sess-webchat-main-2",
+        updatedAt: Date.now(),
+        lastChannel: "whatsapp",
+        lastTo: "+15555550123",
+        deliveryContext: {
+          channel: "whatsapp",
+          to: "+15555550123",
+        },
+      },
+    });
+    const cfg = { session: { store: storePath } } as OpenClawConfig;
+
+    const result = await initSessionState({
+      ctx: {
+        Body: "reply only here",
+        SessionKey: sessionKey,
+        OriginatingChannel: "webchat",
+        OriginatingTo: "session:webchat-main",
+      },
+      cfg,
+      commandAuthorized: true,
+    });
+
+    expect(result.sessionEntry.lastChannel).toBe("webchat");
+    expect(result.sessionEntry.lastTo).toBe("session:webchat-main");
+    expect(result.sessionEntry.deliveryContext?.channel).toBe("webchat");
+    expect(result.sessionEntry.deliveryContext?.to).toBe("session:webchat-main");
+  });
 });