feat(channels): share inbound webhook dedupe across pods via Postgres (#4210)

* feat(channels): share inbound webhook dedupe across pods via Postgres (#4120)

* ci: run cross-pod inbound dedupe integration tests in CI

Expose the job Postgres service via DEDUPE_TEST_POSTGRES_URL so the integration tests (issue #4120) actually execute instead of silently skipping. Normalize the URL for asyncpg (postgresql:// -> +asyncpg, drop libpq-only sslmode) and await the now-async _is_duplicate_inbound in test_github_dispatcher.

.github/workflows/backend-unit-tests.yml

@@ -83,4 +83,10 @@ jobs:
 
       - name: Run unit tests of backend
         working-directory: backend
+        env:
+          # Expose the job's Postgres service to the cross-pod dedupe integration
+          # tests (issue #4120), which read DEDUPE_TEST_POSTGRES_URL. Without this
+          # mapping those tests silently skip and the headline capability is never
+          # exercised in CI.
+          DEDUPE_TEST_POSTGRES_URL: ${{ env.TEST_POSTGRES_URI }}
         run: make test

backend/app/channels/dedupe_store.py

@@ -0,0 +1,276 @@
+"""Inbound webhook dedupe store.
+
+The manager-level inbound dedupe (``ChannelManager._inbound_dedupe_key``) guards an
+agent run / final answer against provider redeliveries. The default store is an
+in-process ``OrderedDict`` (backward compatible, single-pod). A shared store (e.g.
+Postgres) can be injected for multi-pod deployments so a redelivery landing on a
+different pod is still dropped as a duplicate. See issue #4120.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from collections import OrderedDict
+from typing import Any, Protocol
+
+from sqlalchemy import text
+
+logger = logging.getLogger(__name__)
+
+INBOUND_DEDUPE_TTL_SECONDS = 10 * 60
+INBOUND_DEDUPE_MAX_ENTRIES = 4096
+
+# Key tuple matches ChannelManager._inbound_dedupe_key:
+# (channel_name, workspace_id, chat_id, message_id).
+InboundDedupeKey = tuple[str, str, str, str]
+
+
+class InboundDedupeStore(Protocol):
+    """Async contract for recording / releasing inbound dedupe keys.
+
+    ``try_record`` returns ``True`` if the key already existed (duplicate -> drop)
+    and ``False`` if it was newly recorded or its prior entry had expired (proceed).
+    Shared-state implementations must be atomic; the Postgres variant uses a single
+    conditional upsert (``INSERT ... ON CONFLICT DO UPDATE ... WHERE first_seen < TTL``).
+    """
+
+    async def try_record(self, key: InboundDedupeKey) -> bool: ...
+    async def release(self, key: InboundDedupeKey) -> None: ...
+
+
+class MemoryInboundDedupeStore:
+    """Process-local ``OrderedDict`` store — preserves the pre-#4120 behavior exactly."""
+
+    def __init__(
+        self,
+        ttl_seconds: int = INBOUND_DEDUPE_TTL_SECONDS,
+        max_entries: int = INBOUND_DEDUPE_MAX_ENTRIES,
+    ) -> None:
+        self._ttl = ttl_seconds
+        self._max = max_entries
+        # Insertion order == chronological (keys are never re-inserted), so an
+        # OrderedDict lets us evict expired/overflow entries from the front in
+        # O(k) instead of scanning all entries on every inbound message.
+        self._store: OrderedDict[InboundDedupeKey, float] = OrderedDict()
+
+    async def try_record(self, key: InboundDedupeKey) -> bool:
+        now = time.monotonic()
+        # Entries are in chronological insertion order, so expired ones cluster at
+        # the front: pop from the front until we hit a still-live entry.
+        while self._store:
+            _, oldest_at = next(iter(self._store.items()))
+            if now - oldest_at > self._ttl:
+                self._store.popitem(last=False)
+            else:
+                break
+        while len(self._store) > self._max:
+            self._store.popitem(last=False)
+
+        if key in self._store:
+            return True
+
+        self._store[key] = now
+        return False
+
+    async def release(self, key: InboundDedupeKey) -> None:
+        self._store.pop(key, None)
+
+
+class PostgresInboundDedupeStore:
+    """Shared Postgres-backed dedupe store (issue #4120).
+
+    One row per dispatched inbound webhook (keyed by the 4-tuple). A redelivery
+    routed to a different gateway pod hits the same table. The acquire is a single
+    atomic conditional upsert:
+
+        INSERT ... ON CONFLICT (4-tuple) DO UPDATE SET first_seen = now()
+        WHERE first_seen < now() - TTL RETURNING channel
+
+    - No conflict -> row inserted -> proceed.
+    - Conflict + expired row -> DO UPDATE refreshes ``first_seen``, row RETURNED ->
+      proceed (honors the 10-minute ceiling and re-admits a never-released/expired
+      redelivery, e.g. a manual provider "Redeliver").
+    - Conflict + live row -> WHERE fails, no row RETURNED -> drop as duplicate.
+
+    Because the upsert is a single row-locked statement, two pods racing on the same
+    expired key cannot both proceed (one wins the update; the other sees the fresh row
+    and is dropped). Lazy cleanup (``DELETE`` of rows older than the TTL) runs in the
+    same transaction, amortized into the proceed path with no background task.
+
+    Fail-open: any DB error is logged and treated as "allow" so a storage
+    outage never drops a webhook or returns 5xx to the provider.
+    """
+
+    def __init__(self, session_factory: Any | None = None) -> None:
+        # Injected in tests; otherwise resolved lazily from the app engine.
+        self._session_factory = session_factory
+
+    def _resolve_session_factory(self) -> Any:
+        if self._session_factory is not None:
+            return self._session_factory
+        from deerflow.persistence.engine import get_session_factory
+
+        sf = get_session_factory()
+        if sf is None:
+            raise RuntimeError("PostgresInboundDedupeStore requires a Postgres session factory")
+        return sf
+
+    async def try_record(self, key: InboundDedupeKey) -> bool:
+        channel, workspace_id, chat_id, message_id = key
+        try:
+            sf = self._resolve_session_factory()
+            async with sf() as session:
+                async with session.begin():
+                    # Atomic acquire + TTL reclamation in ONE row-locked statement.
+                    #
+                    # - No conflict  -> new row inserted, RETURNING a row -> proceed.
+                    # - Conflict + the existing row is EXPIRED (first_seen older than
+                    #   the TTL) -> DO UPDATE refreshes first_seen to now() and the row
+                    #   is RETURNED, so the redelivery is re-admitted (proceed). This is
+                    #   the cross-pod-safe equivalent of the memory store's "evict
+                    #   expired entries before the membership check", and it honors the
+                    #   10-minute ceiling of issue #4120 even for a row that was never
+                    #   released (e.g. a crashed run): a manual provider "Redeliver" of
+                    #   an old message is re-admitted instead of being dropped forever
+                    #   in a quiet deployment.
+                    # - Conflict + the existing row is still LIVE -> the WHERE fails, no
+                    #   UPDATE, no row RETURNED -> duplicate -> drop.
+                    #
+                    # A single conditional upsert has no TOCTOU window (unlike a separate
+                    # DELETE-then-INSERT), so two pods racing on the same expired key
+                    # cannot both proceed: one wins the update and proceeds, the other
+                    # sees the now-fresh row and is dropped.
+                    result = await session.execute(
+                        text(
+                            "INSERT INTO webhook_deliveries "
+                            "(channel, workspace_id, chat_id, message_id, first_seen) "
+                            "VALUES (:c, :w, :ch, :m, now()) "
+                            "ON CONFLICT (channel, workspace_id, chat_id, message_id) "
+                            "DO UPDATE SET first_seen = now() "
+                            "WHERE webhook_deliveries.first_seen < now() - make_interval(secs => :ttl) "
+                            "RETURNING channel"
+                        ),
+                        {
+                            "c": channel,
+                            "w": workspace_id,
+                            "ch": chat_id,
+                            "m": message_id,
+                            "ttl": INBOUND_DEDUPE_TTL_SECONDS,
+                        },
+                    )
+                    # A returned row means the key was admitted (new delivery, or an
+                    # expired row that was re-admitted). No row means a live duplicate
+                    # that must be dropped. RETURNING (not rowcount) is used because
+                    # rowcount reliability for ON CONFLICT DO NOTHING/DO UPDATE varies
+                    # across DB drivers.
+                    inserted = result.fetchone() is not None
+                    # Lazy cleanup in the same transaction: drop rows older than the
+                    # TTL. Only when a row was admitted (proceed path) so the periodic
+                    # sweep is amortized into normal inbound traffic and keys never
+                    # re-accessed still get reclaimed.
+                    if inserted:
+                        await session.execute(
+                            text("DELETE FROM webhook_deliveries WHERE first_seen < now() - make_interval(secs => :ttl)"),
+                            {"ttl": INBOUND_DEDUPE_TTL_SECONDS},
+                        )
+            # inserted=True -> admitted (proceed, not a duplicate).
+            return not inserted
+        except Exception:
+            # Fail-open: if the store is unavailable we must NOT drop the
+            # message. Return False so the caller treats it as a new delivery
+            # and proceeds (at worst a possible duplicate, never silent loss).
+            logger.exception("PostgresInboundDedupeStore.try_record failed; proceeding without dedupe (fail-open)")
+            return False
+
+    async def release(self, key: InboundDedupeKey) -> None:
+        channel, workspace_id, chat_id, message_id = key
+        try:
+            sf = self._resolve_session_factory()
+            async with sf() as session:
+                async with session.begin():
+                    await session.execute(
+                        text("DELETE FROM webhook_deliveries WHERE channel = :c AND workspace_id = :w AND chat_id = :ch AND message_id = :m"),
+                        {"c": channel, "w": workspace_id, "ch": chat_id, "m": message_id},
+                    )
+        except Exception:
+            logger.exception("PostgresInboundDedupeStore.release failed; key left for TTL expiry (fail-open)")
+
+
+def _gateway_workers() -> int:
+    """Mirror deps._enforce_postgres_for_multi_worker's worker detection."""
+    try:
+        return int(os.environ.get("GATEWAY_WORKERS", "1") or 1)
+    except (TypeError, ValueError):
+        return 1
+
+
+def _build_postgres_store() -> InboundDedupeStore:
+    """Build the shared Postgres dedupe store."""
+    return PostgresInboundDedupeStore()
+
+
+def make_inbound_dedupe_store(app_config: Any | None = None) -> InboundDedupeStore:
+    """Resolve the inbound dedupe store from app config.
+
+    - ``memory`` -> in-process store (per-pod; a redelivery routed to a different
+      replica is NOT deduped).
+    - ``postgres`` -> shared Postgres store. Requires ``database.backend='postgres'``;
+      if the application DB is not Postgres the store falls back to the in-process
+      memory store and logs a WARNING (otherwise cross-pod dedupe would be silently
+      disabled).
+    - ``auto`` (default) -> shared Postgres store whenever the application DB is
+      Postgres. This is the recommended setting for any deployment that may run more
+      than one replica, including Kubernetes with ``GATEWAY_WORKERS=1`` per pod where
+      multiple pods still share the single Postgres DB. For non-Postgres DBs ``auto``
+      falls back to the cheaper in-process memory store (correct for a single-DB,
+      single-pod deployment).
+
+    Emits a WARNING when a multi-worker/multi-replica deployment cannot use the shared
+    Postgres store (the cross-pod dedupe gap becomes an explicit misconfiguration
+    rather than silent default behavior).
+    """
+    backend = "auto"
+    db_is_postgres = False
+    db_backend = None
+    if app_config is not None:
+        dedupe_cfg = getattr(app_config, "dedupe_storage", None)
+        if dedupe_cfg is not None:
+            backend = str(dedupe_cfg.backend.value if hasattr(dedupe_cfg.backend, "value") else dedupe_cfg.backend)
+        db = getattr(app_config, "database", None)
+        db_backend = getattr(db, "backend", None)
+        db_is_postgres = db_backend == "postgres"
+
+    multi_worker = _gateway_workers() > 1
+
+    if backend == "postgres":
+        if not db_is_postgres:
+            logger.warning(
+                "dedupe_storage=postgres requires database.backend='postgres' (got '%s'). Falling back to the in-process memory store; inbound webhook dedupe is per-pod and cross-pod redeliveries will NOT be deduped. See issue #4120.",
+                db_backend,
+            )
+            return MemoryInboundDedupeStore()
+        return _build_postgres_store()
+    if backend == "memory":
+        if multi_worker:
+            logger.warning(
+                "dedupe_storage=memory with GATEWAY_WORKERS>1: inbound webhook dedupe "
+                "is per-pod and will NOT drop redeliveries routed to a different replica. "
+                "Use dedupe_storage=postgres (or remove the setting to let 'auto' pick it) "
+                "for multi-worker deployments. See issue #4120."
+            )
+        return MemoryInboundDedupeStore()
+    # auto
+    if db_is_postgres:
+        logger.info("dedupe_storage=auto resolved to the shared Postgres store (database.backend=postgres); inbound webhook dedupe is shared across pods.")
+        return _build_postgres_store()
+    if multi_worker:
+        logger.warning(
+            "Multi-worker deployment detected but dedupe_storage=auto resolved to the "
+            "in-process memory store (application database is not Postgres). Inbound "
+            "webhook dedupe is per-pod and will NOT drop redeliveries routed to a different "
+            "replica. Set database.backend=postgres (required for multi-worker) so dedupe "
+            "shares state across pods. See issue #4120."
+        )
+    return MemoryInboundDedupeStore()

backend/app/channels/manager.py

@@ -20,6 +20,7 @@ from langgraph_sdk.errors import ConflictError
 
 from app.channels import feishu_run_policy as _feishu_run_policy  # noqa: F401
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
+from app.channels.dedupe_store import InboundDedupeStore, MemoryInboundDedupeStore
 from app.channels.message_bus import (
     INBOUND_FILE_CONTENT_KEY,
     PENDING_CLARIFICATION_METADATA_KEY,
@@ -75,15 +76,14 @@ MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages")
 THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."
 BOUND_IDENTITY_REQUIRED_MESSAGE = "Connect this channel from DeerFlow Settings, complete the in-channel connect step, then send your message again."
 BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is temporarily unavailable. Please try again later or contact the DeerFlow operator."
-# Inbound-redelivery dedup window. ``_recent_inbound_events`` (below) is a
+# Inbound-redelivery dedup window. The dedupe state lives in
-# process-local, in-memory-only OrderedDict — it is never persisted to
+# ``self._inbound_dedupe_store``: the default in-process Memory store is
-# ``ChannelStore`` — so a recorded key survives only for this TTL (or until
+# local to this Gateway process (a recorded key survives only for the store's
-# evicted by the entry cap below) and is gone entirely across a Gateway
+# TTL / entry cap and is gone across a restart), while a Postgres-backed store
-# restart. 10 minutes is a deliberately bounded window: long enough to
+# is shared across pods. 10 minutes is a deliberately bounded window: long
-# absorb a near-term redelivery of the same event — whether a provider's
+# enough to absorb a near-term redelivery of the same event — whether a
-# own automatic retry (where the provider implements one) or an operator
+# provider's own automatic retry or an operator resend — without keeping a
-# explicitly triggering a resend — without keeping a growing in-memory
+# growing ledger.
-# ledger.
 #
 # For GitHub specifically: GitHub does NOT automatically retry or redeliver
 # a failed delivery (non-2xx response, timeout, or connection error) — it
@@ -114,8 +114,6 @@ BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is tempora
 FOLLOWUP_BUFFER_MAX_PER_THREAD = 20
 FOLLOWUP_DRAIN_BATCH_SIZE = 10
 FOLLOWUP_BLOCK_TAG = "followups-while-busy"
-INBOUND_DEDUPE_TTL_SECONDS = 10 * 60
-INBOUND_DEDUPE_MAX_ENTRIES = 4096
 # Only server-stable provider message ids: client-generated ids (client_msg_id,
 # client_id) are not guaranteed identical across a provider's own redelivery, so
 # keying dedupe on them would miss exactly the retries we want to absorb.
@@ -917,6 +915,7 @@ class ChannelManager:
         channel_sessions: dict[str, Any] | None = None,
         connection_repo: Any | None = None,
         require_bound_identity: bool = False,
+        inbound_dedupe_store: InboundDedupeStore | None = None,
         get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
     ) -> None:
         self.bus = bus
@@ -956,14 +955,14 @@ class ChannelManager:
         # stop() has run, for the follow-up drain guard below.
         self._stopped = False
         self._task: asyncio.Task | None = None
-        # Insertion order == chronological (keys are never re-inserted), so an
+        # Inbound webhook dedupe store. Defaults to the in-process Memory store
-        # OrderedDict lets us evict expired/overflow entries from the front in
+        # (pre-#4120 behavior). Multi-pod deployments inject a shared store so
-        # O(k) instead of scanning all entries on every inbound message.
+        # duplicate deliveries landing on different pods are collapsed.
-        self._recent_inbound_events: OrderedDict[tuple[str, str, str, str], float] = OrderedDict()
+        self._inbound_dedupe_store = inbound_dedupe_store if inbound_dedupe_store is not None else MemoryInboundDedupeStore()
         # Per-thread follow-up buffers for busy fire_and_forget channels that
         # opted into ChannelRunPolicy.buffer_followups_on_busy (issue #4121
         # Slice 2). Keyed by thread_id -> OrderedDict[dedupe_key -> entry],
-        # oldest-first, mirroring _recent_inbound_events's shape but scoped
+        # oldest-first, mirroring the dedupe store's shape but scoped
         # per-thread with a hard cap instead of a global TTL (see
         # _buffer_followup / _enforce_followup_cap).
         self._followup_buffers: dict[str, OrderedDict[str, _FollowupEntry]] = {}
@@ -1549,7 +1548,7 @@ class ChannelManager:
             # answer. Provider adapters may emit ack side-effects (a "Working on
             # it…" reply, an "eyes" reaction) before publish_inbound, so those are
             # intentionally not deduped here.
-            if self._is_duplicate_inbound(msg):
+            if await self._is_duplicate_inbound(msg):
                 continue
             logger.info(
                 "[Manager] received inbound: channel=%s, chat_id=%s, type=%s, text_len=%d, files=%d",
@@ -1598,36 +1597,25 @@ class ChannelManager:
             return None
         return (msg.channel_name, str(workspace_id), msg.chat_id, message_id)
 
-    def _is_duplicate_inbound(self, msg: InboundMessage) -> bool:
+    async def _is_duplicate_inbound(self, msg: InboundMessage) -> bool:
         key = self._inbound_dedupe_key(msg)
         if key is None:
             return False
 
-        now = time.monotonic()
+        # Delegated to the shared/per-pod dedupe store. The store owns TTL eviction
-        # Entries are in chronological insertion order, so expired ones cluster at
+        # and capacity bounds; try_record returns True when the key was already
-        # the front: pop from the front until we hit a still-live entry.
+        # present (i.e. this is a duplicate delivery to drop).
-        while self._recent_inbound_events:
+        is_duplicate = await self._inbound_dedupe_store.try_record(key)
-            _, oldest_at = next(iter(self._recent_inbound_events.items()))
+        if is_duplicate:
-            if now - oldest_at > INBOUND_DEDUPE_TTL_SECONDS:
-                self._recent_inbound_events.popitem(last=False)
-            else:
-                break
-        while len(self._recent_inbound_events) > INBOUND_DEDUPE_MAX_ENTRIES:
-            self._recent_inbound_events.popitem(last=False)
-
-        if key in self._recent_inbound_events:
             logger.info(
                 "[Manager] duplicate inbound ignored: channel=%s, chat_id=%s, message_id=%s",
                 msg.channel_name,
                 msg.chat_id,
                 key[-1],
             )
-            return True
+        return is_duplicate
 
-        self._recent_inbound_events[key] = now
+    async def _release_inbound_dedupe_key(self, msg: InboundMessage) -> None:
-        return False
-
-    def _release_inbound_dedupe_key(self, msg: InboundMessage) -> None:
         """Drop a recorded dedupe key so a provider redelivery can be reprocessed.
 
         Called only on transient/unexpected handling failures: the key was
@@ -1637,7 +1625,7 @@ class ChannelManager:
         """
         key = self._inbound_dedupe_key(msg)
         if key is not None:
-            self._recent_inbound_events.pop(key, None)
+            await self._inbound_dedupe_store.release(key)
 
     @staticmethod
     def _log_task_error(task: asyncio.Task) -> None:
@@ -1692,7 +1680,7 @@ class ChannelManager:
             # Transient/unexpected failure: release the dedupe key so a provider
             # redelivery of the same message can recover instead of being dropped
             # for the dedupe TTL.
-            self._release_inbound_dedupe_key(msg)
+            await self._release_inbound_dedupe_key(msg)
             await self._send_error(msg, "An internal error occurred. Please try again.")
 
     # -- chat handling -----------------------------------------------------
@@ -2064,7 +2052,7 @@ class ChannelManager:
                         # Swallowed like the generic handler would not be: release the
                         # key so the provider's redelivery can retry once the thread
                         # frees, instead of being dropped for the dedupe TTL.
-                        self._release_inbound_dedupe_key(msg)
+                        await self._release_inbound_dedupe_key(msg)
                     await self._send_error(msg, THREAD_BUSY_MESSAGE)
                     return
                 raise
@@ -2084,7 +2072,7 @@ class ChannelManager:
                 logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
                 # Same reason as the fire-and-forget branch above: this error is
                 # handled here rather than re-raised, so release explicitly.
-                self._release_inbound_dedupe_key(msg)
+                await self._release_inbound_dedupe_key(msg)
                 await self._send_error(msg, THREAD_BUSY_MESSAGE)
                 return
             else:
@@ -2261,7 +2249,7 @@ class ChannelManager:
                 # handler never runs and never releases the key. Release only
                 # after publishing the final outbound so a provider redelivery
                 # cannot overtake this attempt's terminal reply.
-                self._release_inbound_dedupe_key(msg)
+                await self._release_inbound_dedupe_key(msg)
 
     # -- command handling --------------------------------------------------
 

backend/app/channels/service.py

@@ -99,6 +99,7 @@ class ChannelService:
         *,
         connection_repo: Any | None = None,
         require_bound_identity: bool = False,
+        app_config: AppConfig | None = None,
         get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
     ) -> None:
         self.bus = MessageBus()
@@ -110,6 +111,8 @@ class ChannelService:
         gateway_url = _resolve_service_url(config, "gateway_url", _CHANNELS_GATEWAY_URL_ENV, DEFAULT_GATEWAY_URL)
         default_session = config.pop("session", None)
         channel_sessions = {name: channel_config.get("session") for name, channel_config in config.items() if isinstance(channel_config, dict)}
+        from app.channels.dedupe_store import make_inbound_dedupe_store
+
         self.manager = ChannelManager(
             bus=self.bus,
             store=self.store,
@@ -119,6 +122,7 @@ class ChannelService:
             channel_sessions=channel_sessions,
             connection_repo=connection_repo,
             require_bound_identity=require_bound_identity,
+            inbound_dedupe_store=make_inbound_dedupe_store(app_config),
             get_stream_bridge=get_stream_bridge,
         )
         self._channels: dict[str, Any] = {}  # name -> Channel instance
@@ -157,6 +161,7 @@ class ChannelService:
             channels_config=channels_config,
             connection_repo=_make_connection_repo(connection_config),
             require_bound_identity=require_bound_identity,
+            app_config=app_config,
             get_stream_bridge=get_stream_bridge,
         )
 

backend/packages/harness/deerflow/config/app_config.py

@@ -17,6 +17,7 @@ from deerflow.config.authorization_config import AuthorizationConfig, load_autho
 from deerflow.config.channel_connections_config import ChannelConnectionsConfig
 from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
 from deerflow.config.database_config import DatabaseConfig
+from deerflow.config.dedupe_storage_config import DedupeStorageConfig
 from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
 from deerflow.config.file_signature import get_config_signature as _get_config_signature
@@ -296,6 +297,13 @@ class AppConfig(BaseModel):
             field_doc="Run ownership and lease configuration for multi-worker deployments.",
         ),
     )
+    dedupe_storage: DedupeStorageConfig = Field(
+        default_factory=DedupeStorageConfig,
+        description=format_field_description(
+            "dedupe_storage",
+            field_doc="Inbound webhook dedupe storage backend (memory / postgres / auto) for cross-pod redelivery dedup. See issue #4120.",
+        ),
+    )
 
     # Name -> config lookup tables, (re)built after validation by
     # ``_build_name_indexes``. They make ``get_model_config`` / ``get_tool_config``

backend/packages/harness/deerflow/config/dedupe_storage_config.py

@@ -0,0 +1,41 @@
+"""Configuration for inbound webhook dedupe storage.
+
+Controls where the ChannelManager's inbound dedupe state lives. See issue #4120
+(cross-pod webhook dedupe). The default ``auto`` reuses the Postgres application
+database whenever database.backend='postgres', otherwise an in-process memory
+store. ``memory`` is per-pod and not shared across replicas; ``postgres`` shares
+state across pods.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+from pydantic import BaseModel, Field
+
+from deerflow.config.reload_boundary import format_field_description
+
+
+class DedupeStorageBackend(StrEnum):
+    AUTO = "auto"
+    MEMORY = "memory"
+    POSTGRES = "postgres"
+
+
+class DedupeStorageConfig(BaseModel):
+    """Where inbound webhook dedupe state lives."""
+
+    backend: DedupeStorageBackend = Field(
+        default=DedupeStorageBackend.AUTO,
+        description=format_field_description(
+            "dedupe_storage",
+            field_doc=(
+                "Storage backend for inbound webhook dedupe state. "
+                "'auto' uses the Postgres application database whenever database.backend='postgres', "
+                "otherwise an in-process memory store (single-pod). "
+                "'memory' forces the in-process store (per-pod; not shared across replicas). "
+                "'postgres' shares dedupe state across pods via the application database. "
+                "See issue #4120."
+            ),
+        ),
+    )

backend/packages/harness/deerflow/config/reload_boundary.py

@@ -73,6 +73,10 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
         "RunOwnershipConfig is captured once into RunManager at langgraph_runtime() startup; the lease heartbeat background task is created and "
         "started there, and heartbeat_enabled / lease_seconds / grace_seconds are not re-read on config.yaml edits."
     ),
+    "dedupe_storage": (
+        "make_inbound_dedupe_store() resolves the inbound dedupe store once when ChannelService is constructed at startup; the store "
+        "(in-process memory or shared Postgres) is captured onto ChannelManager and is not rebuilt on config.yaml edits."
+    ),
 }
 
 

backend/packages/harness/deerflow/persistence/migrations/versions/0009_webhook_dedupe.py

@@ -0,0 +1,61 @@
+"""Cross-pod inbound webhook dedupe table (issue #4120).
+
+Revision ID: 0009_webhook_dedupe
+Revises: 0008_thread_operation_kind
+Create Date: 2026-07-15
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0009_webhook_dedupe"
+down_revision: str | Sequence[str] | None = "0008_thread_operation_kind"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if inspector.has_table("webhook_deliveries"):
+        # Idempotent: a DB whose full-metadata create_all already provisioned
+        # the table (e.g. a fresh DB, or a legacy test seed) must not have it
+        # re-created here.
+        return
+    op.create_table(
+        "webhook_deliveries",
+        sa.Column("channel", sa.String(length=64), nullable=False),
+        sa.Column("workspace_id", sa.String(length=512), nullable=False),
+        sa.Column("chat_id", sa.String(length=512), nullable=False),
+        sa.Column("message_id", sa.String(length=1024), nullable=False),
+        sa.Column(
+            "first_seen",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        # Composite PK mirrors ChannelManager._inbound_dedupe_key exactly. A
+        # single joined surrogate key is deliberately avoided: the components
+        # can contain characters (e.g. NUL) that are illegal in a Postgres TEXT
+        # column, and a joined key would also risk length overflow.
+        sa.PrimaryKeyConstraint(
+            "channel",
+            "workspace_id",
+            "chat_id",
+            "message_id",
+            name="pk_webhook_deliveries",
+        ),
+    )
+    op.create_index("ix_webhook_deliveries_first_seen", "webhook_deliveries", ["first_seen"], unique=False)
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if inspector.has_table("webhook_deliveries"):
+        op.drop_index("ix_webhook_deliveries_first_seen", table_name="webhook_deliveries")
+        op.drop_table("webhook_deliveries")

backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -28,6 +28,7 @@ from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
 from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
 
 __all__ = [
     "AgentRow",
@@ -42,4 +43,5 @@ __all__ = [
     "ScheduledTaskRunRow",
     "ThreadMetaRow",
     "UserRow",
+    "WebhookDeliveryRow",
 ]

backend/packages/harness/deerflow/persistence/webhook_delivery/__init__.py

@@ -0,0 +1,5 @@
+"""Shared inbound webhook dedupe table (issue #4120).
+
+ORM model only; row writes/reads go through raw SQL in
+``app.channels.dedupe_store.PostgresInboundDedupeStore``.
+"""

backend/packages/harness/deerflow/persistence/webhook_delivery/model.py

@@ -0,0 +1,37 @@
+"""ORM model for the shared inbound webhook dedupe table (issue #4120).
+
+A single row records that a particular inbound webhook (identified by the
+``_inbound_dedupe_key`` 4-tuple) has already been dispatched, so a redelivery
+routed to a different gateway pod is still dropped as a duplicate. Rows expire
+via lazy cleanup (see ``PostgresInboundDedupeStore``), which deletes rows older
+than ``INBOUND_DEDUPE_TTL_SECONDS`` using the ``first_seen`` column.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Index, PrimaryKeyConstraint, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+class WebhookDeliveryRow(Base):
+    __tablename__ = "webhook_deliveries"
+
+    # Composite primary key mirrors ChannelManager._inbound_dedupe_key exactly:
+    # (channel, workspace_id, chat_id, message_id). Using the four columns
+    # directly avoids any string-joined surrogate key — important because the
+    # components can contain characters (e.g. NUL) that are illegal in a single
+    # Postgres TEXT column, and keeps the ON CONFLICT target natural.
+    channel: Mapped[str] = mapped_column(String(64), nullable=False)
+    workspace_id: Mapped[str] = mapped_column(String(512), nullable=False)
+    chat_id: Mapped[str] = mapped_column(String(512), nullable=False)
+    message_id: Mapped[str] = mapped_column(String(1024), nullable=False)
+    first_seen: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+    __table_args__ = (
+        PrimaryKeyConstraint("channel", "workspace_id", "chat_id", "message_id", name="pk_webhook_deliveries"),
+        Index("ix_webhook_deliveries_first_seen", "first_seen"),
+    )

backend/tests/test_channels.py

@@ -1264,7 +1264,7 @@ class TestChannelManager:
                 outbound_received.append(msg)
                 if msg.is_final:
                     key = manager._inbound_dedupe_key(_inbound())
-                    key_present_during_final_publish.append(key in manager._recent_inbound_events)
+                    key_present_during_final_publish.append(key in manager._inbound_dedupe_store._store)
 
             bus.subscribe_outbound(capture_outbound)
 
@@ -1428,7 +1428,8 @@ class TestChannelManager:
             else:
                 CHANNEL_RUN_POLICY[channel_name] = original
 
-    def test_github_redelivery_is_deduped_like_other_channels(self, tmp_path):
+    @pytest.mark.asyncio
+    async def test_github_redelivery_is_deduped_like_other_channels(self, tmp_path):
         """A redelivered GitHub webhook must dispatch the agent only once.
 
         PR #3584 added inbound dedupe for the IM channels; the GitHub channel
@@ -1467,18 +1468,18 @@ class TestChannelManager:
         assert ChannelManager._inbound_dedupe_key(_gh("d1")) == ("github", "zhfeng/llm-gateway", "zhfeng/llm-gateway", "d1:alice:reviewer")
 
         # First delivery fires; an identical redelivery of the same GUID is dropped.
-        assert manager._is_duplicate_inbound(_gh("d1")) is False
+        assert await manager._is_duplicate_inbound(_gh("d1")) is False
-        assert manager._is_duplicate_inbound(_gh("d1")) is True
+        assert await manager._is_duplicate_inbound(_gh("d1")) is True
         # A genuinely new delivery still fires.
-        assert manager._is_duplicate_inbound(_gh("d2")) is False
+        assert await manager._is_duplicate_inbound(_gh("d2")) is False
         # A second agent fanned out from the SAME delivery is not cross-deduped.
-        assert manager._is_duplicate_inbound(_gh("d1", agent="coder")) is False
+        assert await manager._is_duplicate_inbound(_gh("d1", agent="coder")) is False
         # A second user's SAME-named agent on the SAME delivery is not
         # cross-deduped either. A helper still stamping the old 2-part
         # (delivery, agent) id could not even express this case — it would
         # collide with the very first assertion's "d1"+"reviewer" key and
         # silently drop this user's run (willem-bd, PR #4104 review).
-        assert manager._is_duplicate_inbound(_gh("d1", owner_user_id="bob")) is False
+        assert await manager._is_duplicate_inbound(_gh("d1", owner_user_id="bob")) is False
 
     def test_dispatch_loop_releases_dedupe_key_when_handling_fails(self, tmp_path):
         """A transient handling failure must not black-hole a provider redelivery (ShenAC #1)."""

backend/tests/test_github_dispatcher.py

@@ -1171,8 +1171,8 @@ async def test_dedupe_identity_distinguishes_same_agent_name_across_users(base_d
     from app.channels.store import ChannelStore
 
     manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=base_dir / "dedupe-store.json"))
-    assert manager._is_duplicate_inbound(by_owner["alice"]) is False
+    assert await manager._is_duplicate_inbound(by_owner["alice"]) is False
-    assert manager._is_duplicate_inbound(by_owner["bob"]) is False
+    assert await manager._is_duplicate_inbound(by_owner["bob"]) is False
 
 
 @pytest.mark.asyncio
@@ -1208,12 +1208,12 @@ async def test_missing_delivery_header_leaves_dedupe_open(base_dir: Path) -> Non
     await fanout_event(bus, "pull_request", "", payload)
     (first,) = await _drain(bus)
     assert first.metadata["message_id"] is None
-    assert manager._is_duplicate_inbound(first) is False
+    assert await manager._is_duplicate_inbound(first) is False
 
     await fanout_event(bus, "pull_request", "", payload)
     (second,) = await _drain(bus)
     assert second.metadata["message_id"] is None
-    assert manager._is_duplicate_inbound(second) is False
+    assert await manager._is_duplicate_inbound(second) is False
 
 
 # ---------------------------------------------------------------------------

backend/tests/test_inbound_dedupe.py

@@ -0,0 +1,252 @@
+"""Unit tests for the inbound dedupe store (issue #4120).
+
+Covers config resolution (memory / postgres / auto), the multi-worker
+misconfiguration WARNING, and the Postgres store's atomic insert / fail-open
+behavior with a mocked session factory.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app.channels.dedupe_store import (
+    MemoryInboundDedupeStore,
+    PostgresInboundDedupeStore,
+    make_inbound_dedupe_store,
+)
+
+
+class _FakeDedupe:
+    def __init__(self, backend: str) -> None:
+        self.backend = backend  # plain string, mimics DedupeStorageBackend.value
+
+
+class _FakeDb:
+    def __init__(self, backend: str) -> None:
+        self.backend = backend
+
+
+class _FakeApp:
+    def __init__(self, dedupe: str, db: str) -> None:
+        self.dedupe_storage = _FakeDedupe(dedupe)
+        self.database = _FakeDb(db)
+
+
+def test_factory_default_is_memory():
+    assert isinstance(make_inbound_dedupe_store(None), MemoryInboundDedupeStore)
+
+
+def test_factory_explicit_memory_is_memory():
+    app = _FakeApp("memory", "sqlite")
+    assert isinstance(make_inbound_dedupe_store(app), MemoryInboundDedupeStore)
+
+
+def test_factory_auto_with_sqlite_resolves_memory():
+    app = _FakeApp("auto", "sqlite")
+    assert isinstance(make_inbound_dedupe_store(app), MemoryInboundDedupeStore)
+
+
+def test_factory_auto_with_postgres_and_multi_worker_resolves_postgres_store(monkeypatch):
+    # 'auto' shares Postgres state whenever the DB is Postgres, even with a single
+    # worker: the common K8s case is N pods x 1 worker sharing one DB, and gating on
+    # the in-pod worker count would silently disable cross-pod dedupe there. See the
+    # sibling ..._regardless_of_workers test for the documented behavior.
+    monkeypatch.setenv("GATEWAY_WORKERS", "2")
+    app = _FakeApp("auto", "postgres")
+    assert isinstance(make_inbound_dedupe_store(app), PostgresInboundDedupeStore)
+
+
+def test_factory_auto_with_postgres_resolves_postgres_store_regardless_of_workers():
+    # 'auto' shares Postgres dedupe state whenever the DB is Postgres, including
+    # the common Kubernetes case of many replicas each running a single worker
+    # (GATEWAY_WORKERS=1) that still share one Postgres DB. Cross-pod dedupe must
+    # not depend on the in-pod worker count (issue #4120).
+    app = _FakeApp("auto", "postgres")
+    assert isinstance(make_inbound_dedupe_store(app), PostgresInboundDedupeStore)
+
+
+def test_factory_explicit_postgres_resolves_postgres_store():
+    app = _FakeApp("postgres", "postgres")
+    assert isinstance(make_inbound_dedupe_store(app), PostgresInboundDedupeStore)
+
+
+def test_factory_explicit_postgres_without_postgres_db_falls_back_to_memory_and_warns(monkeypatch, caplog):
+    import logging
+
+    # Explicit 'postgres' but the application DB is not Postgres: must warn and
+    # fall back to the in-process store rather than silently disabling dedupe.
+    app = _FakeApp("postgres", "sqlite")
+    with caplog.at_level(logging.WARNING):
+        store = make_inbound_dedupe_store(app)
+    assert isinstance(store, MemoryInboundDedupeStore)
+    assert any("dedupe_storage=postgres requires database.backend='postgres'" in r.message for r in caplog.records)
+
+
+def test_factory_warns_when_memory_explicit_under_multi_worker(monkeypatch, caplog):
+    import logging
+
+    monkeypatch.setenv("GATEWAY_WORKERS", "2")
+    app = _FakeApp("memory", "sqlite")
+    with caplog.at_level(logging.WARNING):
+        store = make_inbound_dedupe_store(app)
+    assert isinstance(store, MemoryInboundDedupeStore)
+    assert any("dedupe_storage=memory with GATEWAY_WORKERS>1" in r.message for r in caplog.records)
+
+
+def test_factory_warns_when_auto_resolves_memory_under_multi_worker(monkeypatch, caplog):
+    import logging
+
+    monkeypatch.setenv("GATEWAY_WORKERS", "2")
+    app = _FakeApp("auto", "sqlite")
+    with caplog.at_level(logging.WARNING):
+        store = make_inbound_dedupe_store(app)
+    assert isinstance(store, MemoryInboundDedupeStore)
+    assert any("Multi-worker deployment detected but dedupe_storage=auto" in r.message for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# PostgresInboundDedupeStore (unit, mocked session factory)
+# ---------------------------------------------------------------------------
+
+
+def _fake_session_factory(*, insert_rowcount: int = 1, execute_raises: BaseException | None = None):
+    """Build a fake async session factory whose single session returns the
+    given upsert result (fetchone -> a row when admitted, None on live conflict)
+    or raises for ``execute``.
+
+    ``try_record`` issues up to two executes: (0) the conditional upsert
+    ``INSERT ... ON CONFLICT DO UPDATE ... RETURNING`` (admitted -> row returned)
+    and, only when admitted, (1) the cross-table lazy cleanup ``DELETE``. On a
+    live conflict only the upsert runs (no row returned -> no cleanup).
+    """
+    insert_result = MagicMock()
+    insert_result.rowcount = insert_rowcount
+    # A truthy fetchone means the key was admitted (new delivery or an expired row
+    # re-admitted); None means a live conflict (duplicate). Mirrors the RETURNING
+    # channel contract.
+    insert_result.fetchone.return_value = ("c",)
+    session = AsyncMock()
+    if execute_raises is not None:
+        session.execute = AsyncMock(side_effect=execute_raises)
+    else:
+        session.execute = AsyncMock(
+            side_effect=[
+                insert_result,  # conditional upsert result (fetchone -> row when admitted)
+                MagicMock(),  # lazy cleanup DELETE result (only when admitted)
+            ]
+        )
+    begin_cm = AsyncMock()
+    begin_cm.__aenter__ = AsyncMock(return_value=None)
+    begin_cm.__aexit__ = AsyncMock(return_value=False)
+    session.begin = MagicMock(return_value=begin_cm)
+
+    factory_cm = AsyncMock()
+    factory_cm.__aenter__ = AsyncMock(return_value=session)
+    factory_cm.__aexit__ = AsyncMock(return_value=False)
+    factory = MagicMock()
+    factory.return_value = factory_cm
+    return factory, session
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_new_then_conflict():
+    factory, session = _fake_session_factory(insert_rowcount=1)
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    key = ("github", "repo", "repo", "d1:uA:agentX")
+    # First delivery is inserted -> proceed (not a duplicate).
+    assert await store.try_record(key) is False
+    # Redelivery (same key) on a still-live row: ON CONFLICT -> DO UPDATE WHERE
+    # fails -> no row returned -> duplicate -> drop. Only the upsert runs.
+    session.execute.side_effect = [MagicMock(fetchone=MagicMock(return_value=None))]
+    assert await store.try_record(key) is True
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_uses_atomic_on_conflict_and_lazy_cleanup():
+    factory, session = _fake_session_factory()
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    await store.try_record(("slack", "T1", "C1", "123.456"))
+    # call 0: conditional upsert (INSERT ... ON CONFLICT DO UPDATE ... WHERE TTL);
+    # call 1: lazy cross-table cleanup (only on admit).
+    upsert_sql = session.execute.call_args_list[0].args[0].text
+    cleanup_sql = session.execute.call_args_list[1].args[0].text
+    assert "ON CONFLICT" in upsert_sql
+    assert "DO UPDATE SET first_seen = now()" in upsert_sql
+    assert "first_seen < now()" in upsert_sql  # TTL reclamation condition
+    assert "RETURNING" in upsert_sql
+    assert "make_interval" in cleanup_sql
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_reclaims_expired_unreleased_row():
+    # Reproduces the P1 defect: a key's row has passed the TTL but was never
+    # released (e.g. a crashed run). With the conditional upsert, the conflict
+    # fires DO UPDATE (WHERE first_seen < TTL is true), refreshes first_seen and
+    # RETURNS the row, so the redelivery is re-admitted (proceed, not dropped).
+    factory, session = _fake_session_factory()
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    key = ("slack", "T1", "C1", "123.456")
+    # The upsert result returns a row, meaning the expired row was re-admitted
+    # (proceed, not a duplicate).
+    assert await store.try_record(key) is False
+
+    upsert_sql = session.execute.call_args_list[0].args[0].text
+    # The reclaim is part of the atomic upsert: a conditional DO UPDATE gated on
+    # the TTL, not a separate PK-scoped DELETE.
+    assert "ON CONFLICT" in upsert_sql
+    assert "DO UPDATE SET first_seen = now()" in upsert_sql
+    assert "first_seen < now() - make_interval" in upsert_sql
+    # Exactly two executes: the upsert (admitted/refreshed=True) + lazy cleanup.
+    assert len(session.execute.call_args_list) == 2
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_alive_row_still_deduped_without_cleanup():
+    # A still-live (not expired) row for this key must remain a duplicate, and
+    # because no row is returned the lazy cleanup is skipped (only 1 execute).
+    factory, session = _fake_session_factory()
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    key = ("slack", "T1", "C1", "123.456")
+    session.execute.side_effect = [
+        MagicMock(fetchone=MagicMock(return_value=None)),  # upsert: live conflict, no row
+    ]
+    assert await store.try_record(key) is True
+    assert len(session.execute.call_args_list) == 1
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_fail_open_on_exception():
+    factory, _ = _fake_session_factory(execute_raises=RuntimeError("db down"))
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    # Storage error must NOT drop the webhook: fail open = proceed (return False).
+    assert await store.try_record(("discord", "G1", "C1", "111")) is False
+
+
+@pytest.mark.asyncio
+async def test_postgres_release_deletes_key():
+    factory, session = _fake_session_factory()
+    store = PostgresInboundDedupeStore(session_factory=factory)
+    await store.release(("telegram", "chat1", "chat1", "55"))
+    sql = session.execute.call_args_list[0].args[0].text
+    assert "DELETE FROM webhook_deliveries WHERE channel = " in sql
+    assert "AND message_id = " in sql
+
+
+@pytest.mark.asyncio
+async def test_postgres_try_record_fail_open_when_no_session_factory(monkeypatch):
+    # When Postgres is selected but no session factory is available, the store
+    # must fail open (allow the message) rather than crash startup/handling.
+    import deerflow.persistence.engine as engine_mod
+
+    monkeypatch.setattr(engine_mod, "get_session_factory", lambda: None)
+    store = PostgresInboundDedupeStore()  # no injected factory -> resolved lazily
+    # No DB available must NOT drop the message: fail open = proceed (return False).
+    assert await store.try_record(("discord", "G1", "C1", "111")) is False
+
+
+def test_factory_resolves_postgres_store_when_db_is_postgres():
+    app = _FakeApp("postgres", "postgres")
+    store = make_inbound_dedupe_store(app)
+    assert isinstance(store, PostgresInboundDedupeStore)

backend/tests/test_migration_0004_run_ownership_dedupe.py

@@ -157,7 +157,7 @@ async def test_migration_dedupes_duplicate_active_rows_before_unique_index(tmp_p
         with sqlite3.connect(db_path) as raw:
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
         # Bootstrap upgrades through the later revisions after 0004.
-        assert version_row[0] == "0008_thread_operation_kind"
+        assert version_row[0] == "0009_webhook_dedupe"
 
         # Sanity: the invariant the index enforces is now true — at most one
         # active row per thread.

backend/tests/test_migration_0007_scheduled_run_active_dedupe.py

@@ -169,7 +169,7 @@ async def test_migration_supersedes_duplicate_active_runs_before_unique_index(tm
 
         with sqlite3.connect(db_path) as raw:
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-        assert version_row[0] == "0008_thread_operation_kind"
+        assert version_row[0] == "0009_webhook_dedupe"
 
         # Sanity: the invariant the index enforces now holds — at most one
         # active row per task_id.

backend/tests/test_migration_0009_webhook_dedupe.py

@@ -0,0 +1,53 @@
+"""Migration ``0009_webhook_dedupe`` regression test (issue #4120).
+
+Verifies the migration creates ``webhook_deliveries`` with the composite
+primary key (channel, workspace_id, chat_id, message_id) and no legacy
+``dedupe_key`` column, and that re-running it is idempotent.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import create_async_engine
+
+import deerflow.persistence.models  # noqa: F401  -- registers ORM models
+from deerflow.persistence.base import Base
+from deerflow.persistence.bootstrap import bootstrap_schema
+from deerflow.persistence.engine import close_engine, init_engine
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_migration_0009_creates_composite_pk_table_and_is_idempotent(tmp_path: Path) -> None:
+    db_path = tmp_path / "deer.db"
+    url = f"sqlite+aiosqlite:///{db_path}"
+    engine = create_async_engine(url)
+    try:
+        # Seed all baseline tables, then drop ONLY webhook_deliveries so the
+        # 0009 upgrade actually exercises its create_table path (not the
+        # idempotent early-return). Stamp at 0004 so bootstrap upgrades to head.
+        sync = sa.create_engine(f"sqlite:///{db_path}")
+        Base.metadata.create_all(sync)
+        with sync.begin() as conn:
+            conn.execute(sa.text("DROP TABLE IF EXISTS webhook_deliveries"))
+            conn.execute(sa.text("CREATE TABLE IF NOT EXISTS alembic_version (version_num VARCHAR(32) NOT NULL)"))
+            conn.execute(sa.text("DELETE FROM alembic_version"))
+            conn.execute(sa.text("INSERT INTO alembic_version (version_num) VALUES ('0004_run_ownership')"))
+
+        await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+        # Runs upgrade head -> executes 0009.create_table.
+        await bootstrap_schema(engine, backend="sqlite")
+
+        async with engine.connect() as conn:
+            cols = {row["name"] for row in await conn.run_sync(lambda c: sa.inspect(c).get_columns("webhook_deliveries"))}
+        # Composite PK columns only; the old single-column ``dedupe_key`` must
+        # NOT exist (it is illegal in Postgres TEXT and caused schema drift).
+        assert cols == {"channel", "workspace_id", "chat_id", "message_id", "first_seen"}
+
+        # Idempotent: re-running bootstrap at head must not raise (table exists).
+        await bootstrap_schema(engine, backend="sqlite")
+    finally:
+        await close_engine()

backend/tests/test_multi_pod_inbound_dedupe.py

@@ -0,0 +1,170 @@
+"""Integration test: two ChannelManagers share one Postgres dedupe table.
+
+Mirrors ``test_multi_worker_run_ownership.py``'s "two managers sharing a DB"
+pattern. Skipped unless a real Postgres backend is configured via
+``DEDUPE_TEST_POSTGRES_URL`` (the dev container runs sqlite), so it only runs
+where cross-pod dedupe actually matters — CI with Postgres, or a local run with
+a throwaway Postgres. See issue #4120.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import pytest
+import pytest_asyncio
+
+from app.channels.dedupe_store import INBOUND_DEDUPE_TTL_SECONDS, PostgresInboundDedupeStore
+
+POSTGRES_URL = os.environ.get("DEDUPE_TEST_POSTGRES_URL")
+
+# libpq/psycopg-only query parameters that asyncpg's ``connect()`` rejects. CI
+# hands over ``TEST_POSTGRES_URI`` as ``postgresql://...?sslmode=disable``; left
+# in place these raise ``TypeError: connect() got an unexpected keyword argument
+# 'sslmode'`` once the URL is routed to the asyncpg driver, so strip them here.
+_LIBPQ_ONLY_QUERY_KEYS = {"sslmode", "channel_binding"}
+
+
+def _asyncpg_url(url: str | None) -> str | None:
+    """Normalize a sync ``postgresql://`` URL to the async driver SQLAlchemy needs.
+
+    ``init_engine`` builds an *async* engine, so a bare ``postgresql://`` would fall
+    back to the synchronous psycopg2 driver and fail. Accept either form so the test
+    runs whether CI hands over ``TEST_POSTGRES_URI`` (``postgresql://...``) or a
+    caller passes ``postgresql+asyncpg://...`` directly. libpq-only query params
+    (e.g. ``sslmode``) are dropped because asyncpg does not understand them.
+    """
+    if not url:
+        return url
+    if url.startswith("postgresql://"):
+        url = "postgresql+asyncpg://" + url[len("postgresql://") :]
+    parts = urlsplit(url)
+    if parts.query:
+        kept = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True) if k not in _LIBPQ_ONLY_QUERY_KEYS]
+        url = urlunsplit(parts._replace(query=urlencode(kept)))
+    return url
+
+
+pytestmark = pytest.mark.skipif(
+    not POSTGRES_URL,
+    reason="requires DEDUPE_TEST_POSTGRES_URL (real Postgres for cross-pod dedupe)",
+)
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _postgres_engine():
+    """Initialize the Postgres engine within the test's event loop.
+
+    Doing this inside the loop (not at import time) avoids the "connection
+    attached to a different loop" error that arises when the engine is built in
+    a loop that pytest-asyncio later closes.
+    """
+    from deerflow.persistence.engine import close_engine, init_engine
+
+    await init_engine("postgres", url=_asyncpg_url(POSTGRES_URL))
+    try:
+        yield
+    finally:
+        await close_engine()
+
+
+@pytest.mark.asyncio
+async def test_two_stores_share_dedupe_state_across_pods():
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.engine import get_engine, get_session_factory
+    from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
+
+    engine = get_engine()
+    sf = get_session_factory()
+    # Ensure the table exists regardless of whether the alembic migration ran.
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all, tables=[WebhookDeliveryRow.__table__], checkfirst=True)
+
+    store_a = PostgresInboundDedupeStore(session_factory=sf)
+    store_b = PostgresInboundDedupeStore(session_factory=sf)
+    unique = uuid.uuid4().hex
+    key = ("github", "repo", "repo", f"d-{unique}:uA:agentX")
+    try:
+        # First pod records the delivery and proceeds.
+        assert await store_a.try_record(key) is False
+        # A redelivery landing on the second pod hits the same table -> duplicate.
+        assert await store_b.try_record(key) is True
+    finally:
+        await store_a.release(key)
+
+
+@pytest.mark.asyncio
+async def test_manager_injects_shared_store_and_dedupes_cross_pod():
+    """End-to-end: two ChannelManagers wired to the *same* Postgres store drop a
+    redelivery routed to the second manager, exactly as the production dispatch
+    loop would (``if await self._is_duplicate_inbound(msg): continue``)."""
+    from app.channels.manager import ChannelManager
+    from app.channels.message_bus import InboundMessage, MessageBus
+    from app.channels.store import ChannelStore
+    from deerflow.persistence.engine import get_session_factory
+
+    sf = get_session_factory()
+    shared_store = PostgresInboundDedupeStore(session_factory=sf)
+    manager_a = ChannelManager(bus=MessageBus(), store=ChannelStore(), inbound_dedupe_store=shared_store)
+    manager_b = ChannelManager(bus=MessageBus(), store=ChannelStore(), inbound_dedupe_store=shared_store)
+
+    msg = InboundMessage(
+        channel_name="github",
+        chat_id="repo",
+        user_id="alice",
+        text="@bot review",
+        topic_id="7:agent",
+        workspace_id="repo",
+        metadata={"message_id": "d-crosspod:agent"},
+    )
+    try:
+        # First manager records the delivery and lets it through.
+        assert await manager_a._is_duplicate_inbound(msg) is False
+        # Same redelivery on the second manager hits the shared table -> dropped.
+        assert await manager_b._is_duplicate_inbound(msg) is True
+    finally:
+        await manager_a._release_inbound_dedupe_key(msg)
+
+
+@pytest.mark.asyncio
+async def test_expired_unreleased_row_is_reclaimed_on_next_redelivery():
+    """P1 end-to-end: a row past the TTL that was never released must be
+    reclaimed so the next redelivery is re-admitted (not dropped forever)."""
+    from sqlalchemy import text
+
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.engine import get_engine, get_session_factory
+    from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
+
+    engine = get_engine()
+    sf = get_session_factory()
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all, tables=[WebhookDeliveryRow.__table__], checkfirst=True)
+
+    store = PostgresInboundDedupeStore(session_factory=sf)
+    unique = uuid.uuid4().hex
+    channel, workspace_id, chat_id, message_id = ("github", "repo", "repo", f"d-{unique}:expired")
+    key = (channel, workspace_id, chat_id, message_id)
+    try:
+        # Seed an already-expired, unreleased row for this key.
+        async with sf() as session:
+            async with session.begin():
+                await session.execute(
+                    text("INSERT INTO webhook_deliveries (channel, workspace_id, chat_id, message_id, first_seen) VALUES (:c, :w, :ch, :m, now() - make_interval(secs => :age))"),
+                    {"c": channel, "w": workspace_id, "ch": chat_id, "m": message_id, "age": INBOUND_DEDUPE_TTL_SECONDS + 1},
+                )
+        # The expired row is reclaimed -> redelivery re-admitted (not a duplicate).
+        assert await store.try_record(key) is False
+        # Its first_seen is refreshed to ~now (well within the TTL).
+        async with sf() as session:
+            row = (
+                await session.execute(
+                    text("SELECT (now() - first_seen) < make_interval(secs => :ttl) AS fresh FROM webhook_deliveries WHERE channel = :c AND workspace_id = :w AND chat_id = :ch AND message_id = :m"),
+                    {"c": channel, "w": workspace_id, "ch": chat_id, "m": message_id, "ttl": INBOUND_DEDUPE_TTL_SECONDS},
+                )
+            ).fetchone()
+        assert row is not None and row[0] is True
+    finally:
+        await store.release(key)

backend/tests/test_persistence_bootstrap.py

@@ -48,7 +48,7 @@ from deerflow.persistence.migrations._helpers import _normalize_default
 asyncio_test = pytest.mark.asyncio
 
 
-HEAD = "0008_thread_operation_kind"
+HEAD = "0009_webhook_dedupe"
 BASELINE = "0001_baseline"
 
 

backend/tests/test_persistence_bootstrap_concurrency.py

@@ -28,7 +28,7 @@ from deerflow.persistence.bootstrap import bootstrap_schema
 pytestmark = pytest.mark.asyncio
 
 
-HEAD = "0008_thread_operation_kind"
+HEAD = "0009_webhook_dedupe"
 
 
 def _url(tmp_path: Path) -> str:

backend/tests/test_persistence_bootstrap_regression.py

@@ -76,7 +76,7 @@ async def test_legacy_database_recovers_token_usage_column(tmp_path: Path) -> No
             cols = {row[1] for row in raw.execute("PRAGMA table_info(runs)").fetchall()}
             assert "token_usage_by_model" in cols
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-            assert version_row[0] == "0008_thread_operation_kind"
+            assert version_row[0] == "0009_webhook_dedupe"
 
         # And the read path that originally 500'd must now succeed.
         sf = get_session_factory()
@@ -116,6 +116,6 @@ async def test_legacy_database_with_manual_alter_still_bootstraps(tmp_path: Path
             # No duplicate column -- list, not set, to catch dupes.
             assert cols.count("token_usage_by_model") == 1
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-            assert version_row[0] == "0008_thread_operation_kind"
+            assert version_row[0] == "0009_webhook_dedupe"
     finally:
         await close_engine()

config.example.yaml

@@ -15,7 +15,7 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 29
+config_version: 30
 
 # ============================================================================
 # Logging
@@ -1825,6 +1825,30 @@ skill_evolution:
 database:
   backend: sqlite
   sqlite_dir: .deer-flow/data
+
+# ============================================================================
+# Inbound webhook dedupe storage (cross-pod redelivery dedup — issue #4120)
+# ============================================================================
+# Where the ChannelManager records inbound webhook dedupe state.
+#
+# backend: auto     -- (default) Postgres whenever database.backend=postgres,
+#                       including multi-replica K8s where each pod runs a single
+#                       worker but shares one DB. Otherwise an in-process memory
+#                       store (single-DB, single-pod).
+#                       NOTE: multi-replica deployments must use a Postgres DB;
+#                       if database.backend is sqlite/memory, 'auto' falls back to
+#                       the per-pod memory store and cross-pod dedupe is disabled.
+# backend: memory   -- In-process store only. Per-pod: a webhook redelivered to
+#                       a DIFFERENT replica is NOT deduped. Not recommended for
+#                       multi-worker deployments (logs a startup WARNING).
+# backend: postgres -- Share dedupe state across pods via the application DB.
+#                       REQUIRED for any multi-replica deployment. If
+#                       database.backend is not 'postgres', a WARNING is logged and
+#                       the store falls back to the in-process memory store.
+#
+# dedupe_storage:
+#   backend: auto
+
   # Recycle app ORM PostgreSQL connections before the environment's idle cutoff.
   pool_recycle: 300
   # App ORM PostgreSQL command timeout. Set to null to disable it or raise it

deploy/helm/deer-flow/README.md

@@ -124,7 +124,7 @@ they resolve from the `secrets` map):
 
 ```yaml
 config: |
-  config_version: 29
+  config_version: 30
   models:
     - name: gpt-4
       use: langchain_openai:ChatOpenAI

deploy/helm/deer-flow/values.yaml

@@ -240,7 +240,7 @@ ingress:
 # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
 #    inline literal secret values here. The default enables provisioner sandbox.
 config: |
-  config_version: 29
+  config_version: 30
   log_level: info
 
   models: []