fix(channels): run each IM message once across instances, restarts and late redeliveries (#7104)

* fix(channels): run each IM message once across instances, restarts and late redeliveries

Inbound IM messages were deduplicated only by the 10-minute inbound dedupe
store (#7103). Discord direct messages had no key at all, a Discord thread
reply got an instance-dependent key, a Discord thread was created before
dedupe ran, a redelivery past the window started a second run, and two
instances racing on a conversation's first messages bound it to two
threads.

- Discord messages are keyed (discord, guild_id or "dm", posting channel,
  message_id); the posting channel is the same on every instance.
- MessageBus.claim_inbound lets an adapter record the manager's dedupe key
  before a side effect; Discord claims it on the Gateway loop before
  _create_thread, marks the published message dedupe_claimed, and gives the
  claim back when the message is never committed.
- runs.wait / runs.stream / runs.create send an Idempotency-Key derived from
  the dedupe key, and keyed human message ids are uuid5 of it (a uuid4 per
  attempt made every keyed retry a 409). A reused run's status-only /wait or
  stream gap is resolved to this message's own turn via threads.get_state;
  409s about the message's own keyed run keep the key instead of reading as
  a busy thread; an ended reused GitHub run gets no follow-up watcher.
- A conversation's first thread binding is claimed insert-if-absent
  (claim_thread_id on channel_conversations and channel_thread_bindings);
  the loser adopts the winner and deletes the empty thread it minted.
  /new and /agent use still rebind.

Docs: IM_CHANNEL_CONNECTIONS.md, CONFIGURATION.md, channels AGENTS.md,
CHANGELOG (en/zh).

* fix(channels): give back Discord dedupe claims for messages never processed

Review round 1 on #7104. A Discord claim records the manager's dedupe key
before the message is queued, so every path that drops the message unrun
must release it, or a peer refuses the redelivery for the rest of the TTL:

- The hand-off task releases the claim on cancellation too (stop() drains it
  while it resolves the connection identity), via await_drained so the
  cancellation still propagates.
- ChannelManager._discard_pending_inbound releases the key of a claimed
  message that shutdown drops from the queue, after the grace period and on
  a cancelled shutdown.
- The adapter tracks claims until they are committed. A release refused
  because stop() already closed cross-loop intake, or a hand-off cancelled
  before it ran, leaves the claim tracked, and stop() releases it last.

* fix(gateway): keep IM run idempotency stable when /new rebinds the conversation

Review round 2 on #7104. The Gateway scoped every Idempotency-Key by owner
and thread. /new rebinds an IM conversation to a fresh thread, so a late
redelivery of a message admitted on the old thread was requested on the new
one, got a different scoped key and started a second run.

- Keys from trusted internal callers (the IM channel manager) are scoped by
  owner only (`internal-run:` prefix); browser and API keys keep the
  owner-and-thread scope (`http-run:`).
- A key that resolves to a run of another thread or user raises
  RunIdempotencyKeyMismatch, a RuntimeError subclass so in-process launchers
  see the same type, and start_run answers it with 409 "Idempotency-Key
  already used with a different request" instead of a 500. The channel
  manager already treats that 409 as the message's admitted run.

Docs: API.md, IM_CHANNEL_CONNECTIONS.md, gateway AGENTS.md (reworded to stay
under its budget), CHANGELOG (en/zh).

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -1265,6 +1265,30 @@ This release closes that milestone with **621 merged pull requests**.
 
 ### Fixed
 
+- **channels:** An IM message delivered twice no longer runs twice, and two
+  Gateway instances no longer split a new conversation across two threads.
+  Discord direct messages had no inbound dedupe key (no guild id), and a reply
+  in a Discord thread got a different key on an instance that does not track
+  that thread, so with every instance's session receiving every message both
+  answered. Discord messages are now keyed by guild (`dm` for a direct message)
+  and by the channel they were posted in, and the adapter claims that key
+  before it creates a Discord thread, so only one instance creates one. Past
+  the 10-minute dedupe window, after a restart, or while the shared dedupe
+  store is unavailable, a redelivery used to start a second run (a GitHub
+  "Redeliver" could post a duplicate comment). Every IM run request for a keyed
+  message now carries an `Idempotency-Key` derived from the dedupe key, with a
+  human message id derived from the same key, so the Gateway returns the run
+  it admitted for the first delivery: the redelivery replays its stream, reads
+  this message's turn from the thread state, or replies that the message is
+  still being processed, and never runs the turn again, even when the first
+  run failed or `/new` has since moved the conversation to a new thread: the
+  Gateway scopes `Idempotency-Key`s from trusted internal callers by owner
+  instead of owner and thread, and a key already used on another thread now
+  answers `409` instead of `500`. When two instances create a thread for a
+  conversation's first message, the binding is now claimed insert-if-absent in
+  `channel_conversations` / `channel_thread_bindings`: the instance that loses
+  adopts the winner's thread and deletes its own empty one, and `/new` still
+  rebinds. (issue #7103, [#7104])
 - **skills:** A skill the agent creates, edits, patches or deletes with the
   `skill_manage` tool now reaches every Gateway process sharing the home
   volume, as a change made through the skills API already did. The tool
@@ -11877,3 +11901,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
 [#7040]: https://github.com/bytedance/deer-flow/pull/7040
 [#7099]: https://github.com/bytedance/deer-flow/pull/7099
+[#7104]: https://github.com/bytedance/deer-flow/pull/7104

CHANGELOG_zh.md

@@ -1015,6 +1015,21 @@
 
 ### 修复
 
+- **渠道：** 同一条 IM 消息被投递两次不再被执行两次，两个 Gateway 实例也不再把一个新
+  会话拆到两个线程上。Discord 私信此前没有入站去重键（没有 guild id），而 Discord 线程
+  内的回复在不跟踪该线程的实例上会得到另一个键；由于每个实例的会话都会收到每条消息，
+  两个实例都会作答。Discord 消息现在按 guild（私信记为 `dm`）和消息所在频道生成去重键，
+  适配器在创建 Discord 线程之前先认领该键，因此只有一个实例会创建线程。超过 10 分钟
+  去重窗口、重启之后或共享去重存储不可用时，重投递此前会启动第二次运行（GitHub 的
+  “Redeliver”可能发出重复评论）。现在带去重键的消息发起的每个 IM 运行请求都携带由去重键
+  派生的 `Idempotency-Key`，人类消息 id 也由同一个键派生，因此 Gateway 会返回首次投递
+  时已接纳的运行：重投递会重放其事件流、从线程状态读取这条消息对应的轮次，或回复“该消息
+  仍在处理中”，而不会把这一轮再执行一遍——即使首次运行失败，或 `/new` 之后会话已换到新
+  线程也是如此：Gateway 对受信内部调用方的 `Idempotency-Key` 只按 owner 而不再按 owner 与
+  线程划分作用域，在另一个线程上复用已用过的 key 现在返回 `409` 而不是 `500`。
+  两个实例为同一会话的第一条消息各自创建线程时，绑定现在以“不存在才插入”的方式在
+  `channel_conversations` / `channel_thread_bindings` 中认领：落败的实例改用胜出者的线程
+  并删除自己创建的空线程，`/new` 仍会重新绑定到新线程。 (issue #7103, [#7104])
 - **技能：** agent 通过 `skill_manage` 工具创建、编辑、修补或删除的技能，现在会像通过
   技能 API 所做的修改一样，传达到共享同一 home 卷的每个 Gateway 进程。此前该工具只刷新
   本进程的技能提示缓存，对端副本会继续为该用户渲染旧的技能列表；现在它在本地刷新之后、
@@ -9439,3 +9454,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
 [#7040]: https://github.com/bytedance/deer-flow/pull/7040
 [#7099]: https://github.com/bytedance/deer-flow/pull/7099
+[#7104]: https://github.com/bytedance/deer-flow/pull/7104

backend/app/channels/AGENTS.md

@@ -13,10 +13,10 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors.
 
 **Components**:
 - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks.
-- `store.py` - `ChannelStore` protocol (async `get_thread_id`/`set_thread_id`/`remove`/`list_entries`) for `channel_name:chat_id[:topic_id]` → `thread_id`. `SqlChannelStore` (`channel_thread_bindings`, harness `persistence/channel_thread_bindings/`; upsert keeps `created_at`, topic-less `remove` deletes by key prefix) for sqlite/postgres, so replicas share bindings; `JsonChannelStore` (`{base_dir}/channels/store.json`, loaded once per process, file I/O via `to_thread` under `_lock`) only for `database.backend: memory`. `resolve_channel_store` picks; `ChannelService.start()` runs `SqlChannelStore.import_legacy_json` once (file + empty table → `INSERT … ON CONFLICT DO NOTHING`, rename to `store.json.migrated`, a peer's rename tolerated; populated table → file kept). Feishu's sync lark callback (on the SDK's own loop) bridges `_resolve_routing_bounded` to the Gateway loop via `_run_store_call` (tracked future; refuses only a wait from the Gateway loop itself). Store calls run under `FEISHU_STORE_LOOKUP_TIMEOUT_SECONDS` (3s; the SDK loop also services ws keepalives); timeout or store error = *unresolved* (`StoreLookupUnresolved`), never a miss: the message joins a per-chat FIFO drained on the Gateway loop with backoff up to `FEISHU_STORE_RESOLVE_TOTAL_BUDGET_SECONDS` (30s), then is dropped with a warning, never routed to a new conversation; each deferred message holds a bus intake reservation (the `inbound_queue_maxsize` bound) and its 30s budget runs from arrival. `discord.py` reads `channels_dir`. Tests: `test_channel_thread_binding_store.py` (also under `blocking_io/`).
-- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`; `runs.wait`/`runs.stream` pass `on_disconnect: continue` (a run outlives its requester; reply lost, thread kept), so once a request is sent, cancellation or a transport error keeps the inbound dedupe key (`_keep_dedupe_key_after`); only Gateway rejections, pre-request failures and `error` frames release it
+- `store.py` - `ChannelStore` protocol (async `get_thread_id`/`set_thread_id`/`claim_thread_id`/`remove`/`list_entries`) for `channel_name:chat_id[:topic_id]` → `thread_id`. `SqlChannelStore` (`channel_thread_bindings`, harness `persistence/channel_thread_bindings/`; upsert keeps `created_at`, topic-less `remove` deletes by key prefix) for sqlite/postgres, so replicas share bindings; `JsonChannelStore` (`{base_dir}/channels/store.json`, loaded once per process, file I/O via `to_thread` under `_lock`) only for `database.backend: memory`. `resolve_channel_store` picks; `ChannelService.start()` runs `SqlChannelStore.import_legacy_json` once (file + empty table → `INSERT … ON CONFLICT DO NOTHING`, rename to `store.json.migrated`, a peer's rename tolerated; populated table → file kept). Feishu's sync lark callback (on the SDK's own loop) bridges `_resolve_routing_bounded` to the Gateway loop via `_run_store_call` (tracked future; refuses only a wait from the Gateway loop itself). Store calls run under `FEISHU_STORE_LOOKUP_TIMEOUT_SECONDS` (3s; the SDK loop also services ws keepalives); timeout or store error = *unresolved* (`StoreLookupUnresolved`), never a miss: the message joins a per-chat FIFO drained on the Gateway loop with backoff up to `FEISHU_STORE_RESOLVE_TOTAL_BUDGET_SECONDS` (30s), then is dropped with a warning, never routed to a new conversation; each deferred message holds a bus intake reservation (the `inbound_queue_maxsize` bound) and its 30s budget runs from arrival. `discord.py` reads `channels_dir`. Tests: `test_channel_thread_binding_store.py` (also under `blocking_io/`).
+- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`; `runs.wait`/`runs.stream` pass `on_disconnect: continue` (a run outlives its requester; reply lost, thread kept), so once a request is sent, cancellation or a transport error keeps the inbound dedupe key (`_keep_dedupe_key_after`); only Gateway rejections, pre-request failures and `error` frames release it. Keyed messages send `Idempotency-Key` (from the dedupe key) and a uuid5 human message id, so a late redelivery gets the admitted run: a status-only `/wait` or a stream `gap` resolves this message's turn via `threads.get_state`; a 409 on its own run (`_ADMITTED_RUN_CONFLICT_DETAILS`) replies `ALREADY_ADMITTED_MESSAGE` and keeps the key. First bindings use `claim_thread_id` (insert-if-absent): a loser adopts the winner and deletes its minted thread (never a preferred id); `/new` rebinds. Discord keys `(discord, guild|dm, channel_id, message_id)` and claims via `MessageBus.claim_inbound` before `_create_thread`; `dedupe_claimed` skips the worker's record; uncommitted (incl. cancelled) or shutdown-discarded claims are released
   Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection.
-  **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `<durable_context_data>` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`).
+  **Stream publishing is an allowlist** (`_accumulate_stream_text`): only assistant types (`"ai"`, `"AIMessageChunk"`, OpenAI-style `"assistant"`) become text, matched by prefix, never substring (`chain`, `domain` contain `"ai"`). The old `"tool"` denylist leaked hidden model context: `DynamicContextMiddleware` (recalled memory, the rewritten user turn) and `DurableContextMiddleware` inject hidden `HumanMessage`s that fan out on `messages-tuple`. `_stream_payload_type` reads the `model_dump()` shape and LangChain's `to_json()` constructor shape (top-level `type` `"constructor"`, class name at the tail of `id`). Bare `str` payloads are rejected: untyped, and `serialize_messages_tuple` always emits `[message_dict, metadata]`.
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`: the shared Gateway loop also serves `POST /api/channels/{name}/restart`; Slack cleanup awaits connect then closes off-loop; a 5s deadline/cancellation retains it for retry. SDK close errors propagate. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push**: WeCom, Feishu, WeChat and QQ C2C share the outbox; `capabilities.py` gates enqueue/UI. Unsupported rows fail terminally. Await ACK; `ChannelUnavailable` parks, business rejection counts. WeCom parks socket `RuntimeError` except `Reply ack error`. Feishu/QQ/WeCom group hooks use owner connection IDs; Feishu p2p resolves owner. `group_name`: OAuth state only, no suffix/raw-ID labels. WeCom legacy-group fallback excludes private bindings. QQ: owner/current-App C2C only, never member OpenIDs/passive context. WeChat: encrypted owner/bot/private context across restart; ignore public tokens. `store_connection_context` locks/checks the active identity before merging encrypted extra; browser responses strip legacy tokens. Preferences gate enqueue/send/retry, cancel pending/sending and fence late writes; reenable never replays. In-flight sends may finish. Binding lookup errors park; revoked targets fail before liveness. `_on_finalization` commits one notice with the outcome (stop > pause > finish > run); recovery dedupes, idle uses `idle:`, trials/interrupts stay silent. Web/background events default off: confirm durable terminal state; recheck Human Input before send/retry. Text: localized, no IDs/links. Delivery starts after channels/stops first; failure detaches enqueue. [IM guide](../../docs/IM_CHANNEL_CONNECTIONS.md#proactive-notification-settings).
 - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`.
 - `service.py` - Drain manager work before transports; retain timed-out instances for retry. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu first disarms lark's `_auto_reconnect`; DingTalk runs its own SDK connect-loop copy), then join; a live join or Slack cleanup raises `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup. Leader lease (`leadership.py`): only the `channel_leases` holder starts a non-GitHub transport (`_start_channel` checks `holds()`; callers acquire first); `_supervise` renews at a fixed rate, runs starts/stops per provider, stops a transport before its lease lapses, re-applies runtime config whose `version` changed; reloads read a disconnect as its tombstone.

backend/app/channels/discord.py

@@ -53,6 +53,10 @@ DISCORD_MAX_INBOUND_FILE_BYTES = 25 * 1024 * 1024
 DISCORD_CDN_HOST_SUFFIXES = frozenset({"discordapp.net", "discordapp.com", "discord.com"})
 
 
+class _NotClaimed(Exception):
+    """This delivery lost the message's dedupe claim: create no thread, publish nothing."""
+
+
 def _parse_discord_guild_id(entry: Any) -> int | None:
     """Return a Discord guild (snowflake) ID, or ``None`` if ``entry`` is not one."""
     if isinstance(entry, int) and not isinstance(entry, bool) and entry > 0:
@@ -131,6 +135,12 @@ class DiscordChannel(Channel):
         # Uses a dedicated JSON file next to the channel store's state, separate from
         # the ChannelStore itself, which maps IM conversations to DeerFlow thread IDs.
         self._active_threads: dict[str, str] = {}
+        # Dedupe claims recorded for messages not yet committed to the bus, by
+        # Discord message id. Each is settled once: a commit hands it to the
+        # queue (the manager releases it if shutdown discards the message),
+        # every other outcome releases it, and stop() releases what is left.
+        self._unsettled_claims: dict[str, InboundMessage] = {}
+        self._unsettled_claims_lock = threading.Lock()
         # Reverse-lookup set for O(1) thread ID checks (avoids O(n) scan of _active_threads.values()).
         self._active_thread_ids: set[str] = set()
         # Lock protecting _active_threads and the JSON file from concurrent access.
@@ -352,6 +362,17 @@ class DiscordChannel(Channel):
         if discord_loop and discord_loop is not current_loop and not discord_loop.is_running():
             self._discard_ephemeral_tasks()
 
+        # Claims the stopped loop left unsettled: a release refused after intake
+        # closed, or a hand-off cancelled before it ran. None was committed.
+        with self._unsettled_claims_lock:
+            leftover = list(self._unsettled_claims.values())
+            self._unsettled_claims.clear()
+        for claim in leftover:
+            try:
+                await self.bus.release_inbound_claim(claim)
+            except Exception:
+                logger.warning("[Discord] failed to release the dedupe claim of uncommitted message %s", claim.metadata.get("message_id"), exc_info=True)
+
         self._client = None
         self._discord_loop = None
         self._discord_module = None
@@ -651,6 +672,14 @@ class DiscordChannel(Channel):
         """Finish Discord routing while owning one bounded intake slot."""
         reservation_transferred = False
         files = files or []
+        # Every instance's session receives every MESSAGE_CREATE; the one that
+        # wins this message's dedupe claim is the only one to create a thread.
+        claim = self._make_inbound(
+            chat_id=str(message.channel.id),
+            user_id=str(message.author.id),
+            text="",
+            metadata={"guild_id": str(guild.id) if guild else None, "channel_id": str(message.channel.id), "message_id": str(message.id)},
+        )
         try:
             # --- Determine thread/channel routing and typing target ---
             thread_id = None
@@ -709,7 +738,7 @@ class DiscordChannel(Channel):
                     return
                 # mention_only + fresh @ → create new thread instead of routing to existing one
                 if self._mention_only and has_mention:
-                    thread_obj = await self._create_thread(message)
+                    thread_obj = await self._create_thread_once_claimed(message, claim)
                     if thread_obj is not None:
                         target_thread_id = str(thread_obj.id)
                         self._record_thread_mapping(channel_id, target_thread_id)
@@ -736,7 +765,7 @@ class DiscordChannel(Channel):
                 return
             elif self._mention_only and has_mention:
                 # First mention in this channel → create thread
-                thread_obj = await self._create_thread(message)
+                thread_obj = await self._create_thread_once_claimed(message, claim)
                 if thread_obj is not None:
                     target_thread_id = str(thread_obj.id)
                     self._record_thread_mapping(channel_id, target_thread_id)
@@ -753,7 +782,7 @@ class DiscordChannel(Channel):
                     typing_target = message.channel  # Type into the channel
             elif self._thread_mode:
                 # thread_mode but mention_only is False → create thread anyway for conversation grouping
-                thread_obj = await self._create_thread(message)
+                thread_obj = await self._create_thread_once_claimed(message, claim)
                 if thread_obj is None:
                     # Thread creation failed (disabled/permissions), fall back to channel replies
                     logger.info("[Discord] thread creation failed in channel %s, falling back to channel replies", channel_id)
@@ -787,15 +816,61 @@ class DiscordChannel(Channel):
                 },
             )
             inbound.topic_id = thread_id
+            inbound.dedupe_claimed = claim.dedupe_claimed
 
             handoff = self._publish_reserved(inbound, reservation, guild_id=str(guild.id) if guild else None)
             if handoff is None:
                 return
             reservation_transferred = True
             await self._acknowledge_after_handoff(message, handoff, typing_target=typing_target, chat_id=chat_id, thread_id=thread_id)
+        except _NotClaimed:
+            logger.info("[Discord] skipping message %s: another delivery claimed it, or the Gateway loop is stopping", message.id)
         finally:
             if not reservation_transferred:
                 reservation.release()
+                if claim.dedupe_claimed:
+                    # Refused once stop() closed cross-loop intake; stop() then releases it.
+                    message_id = claim.metadata["message_id"]
+                    self._submit_threadsafe_coroutine(self._release_unsettled_claim(message_id), self._main_loop, name="release_inbound_claim", msg_id=message_id)
+
+    async def _create_thread_once_claimed(self, message, claim: InboundMessage):
+        """``_create_thread`` after winning *claim*; raises ``_NotClaimed`` on a duplicate."""
+        if not claim.dedupe_claimed and not await self._claim_inbound(claim):
+            raise _NotClaimed
+        return await self._create_thread(message)
+
+    async def _claim_inbound(self, claim: InboundMessage) -> bool:
+        """Run ``MessageBus.claim_inbound`` on the Gateway loop, where the dedupe store's engine lives."""
+        future = self._submit_threadsafe_coroutine_future(self._claim_on_gateway_loop(claim), self._main_loop, name="claim_inbound", msg_id=claim.metadata.get("message_id"))
+        if future is None:
+            return False  # the Gateway loop is gone; the hand-off would be refused as well
+        try:
+            return await asyncio.shield(asyncio.wrap_future(future))
+        except asyncio.CancelledError:
+            if not future.cancelled():
+                raise
+            return False  # drained by stop()
+        except Exception:
+            # Fail open like the dedupe store itself: the worker still dedupes.
+            logger.warning("[Discord] inbound dedupe claim failed for message %s; continuing", claim.metadata.get("message_id"), exc_info=True)
+            return True
+
+    async def _claim_on_gateway_loop(self, claim: InboundMessage) -> bool:
+        claimed = await self.bus.claim_inbound(claim)
+        if claim.dedupe_claimed:
+            # Registered with the claim itself, so no later exit path can lose it.
+            with self._unsettled_claims_lock:
+                self._unsettled_claims[claim.metadata["message_id"]] = claim
+        return claimed
+
+    def _settle_claim(self, message_id: Any) -> InboundMessage | None:
+        """Take *message_id*'s unsettled claim, if any: the caller now decides its outcome."""
+        with self._unsettled_claims_lock:
+            return self._unsettled_claims.pop(str(message_id), None)
+
+    async def _release_unsettled_claim(self, message_id: Any) -> None:
+        if (claim := self._settle_claim(message_id)) is not None:
+            await self.bus.release_inbound_claim(claim)
 
     def _publish_reserved(self, inbound: InboundMessage, reservation: InboundReservation, *, guild_id: str | None) -> Future[bool] | None:
         """Transfer an already-reserved message to the Gateway loop.
@@ -876,9 +951,22 @@ class DiscordChannel(Channel):
         *,
         guild_id: str | None,
     ) -> bool:
-        """Attach connection identity and commit the reservation on the Gateway loop."""
-        inbound = await self._attach_connection_identity(inbound, guild_id=guild_id)
-        return self._commit_reserved_inbound(reservation, inbound)
+        """Attach connection identity and commit the reservation on the Gateway loop.
+
+        A message that was claimed against inbound dedupe but never committed
+        gives its claim back, also when stop() cancels this task.
+        """
+        message_id = inbound.metadata.get("message_id")
+        try:
+            committed = self._commit_reserved_inbound(reservation, await self._attach_connection_identity(inbound, guild_id=guild_id))
+        except BaseException:
+            await await_drained(self._release_unsettled_claim(message_id))
+            raise
+        if committed:
+            self._settle_claim(message_id)
+        else:
+            await self._release_unsettled_claim(message_id)
+        return committed
 
     async def _attach_connection_identity(self, inbound: InboundMessage, guild_id: str | None = None) -> InboundMessage:
         return await attach_connection_identity(

backend/app/channels/manager.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
+import json
 import logging
 import math
 import mimetypes
@@ -115,6 +117,24 @@ THREAD_BUSY_MESSAGE = "This conversation is already processing another request.
 # such policy: its request returns as soon as the run is pending.
 CHANNEL_RUN_ON_DISCONNECT = "continue"
 
+# Run admission is idempotent on this key: a redelivery that outlived the
+# inbound dedupe window (or a restart, or a fail-open dedupe store) gets the
+# run the Gateway already admitted instead of a second one. The digest names
+# the inbound dedupe key. The Gateway scopes an internal caller's key by
+# owner only, so a redelivery requested on the thread ``/new`` bound later
+# still meets the admitted run (a 409) instead of starting another one.
+IDEMPOTENCY_KEY_HEADER = "Idempotency-Key"
+_RUN_IDEMPOTENCY_KEY_PREFIX = "channel-inbound:v1:"
+# The Gateway reuses a keyed run only for an identical input, so a keyed
+# message's human message id is derived from the same key instead of uuid4.
+_HUMAN_MESSAGE_ID_NAMESPACE = uuid.UUID("5b7c1e0e-8f5a-4d43-9a63-0c5b2f1d7e21")
+# 409 details that name this message's own keyed run rather than another
+# active run: the redelivery's input differs (re-ingested attachments get new
+# names), or the run is live on a peer whose stream this instance cannot read.
+_ADMITTED_RUN_CONFLICT_DETAILS = ("Idempotency-Key already used", "is not active on this worker and cannot be streamed")
+ALREADY_ADMITTED_MESSAGE = "This message is already being processed. Its answer will appear in this conversation in DeerFlow."
+_LIVE_RUN_STATUSES = frozenset({"pending", "running"})
+
 
 @dataclass
 class _RunAdmission:
@@ -216,13 +236,12 @@ BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is tempora
 #
 # At the boundary: a manual redelivery (e.g. GitHub's "Redeliver" button)
 # clicked *after* the TTL has elapsed, or any redelivery following a Gateway
-# restart, is no longer recognized as a duplicate — the key has already been
-# evicted, or never existed in the new process — so the agent runs again
-# and may repeat a real side effect (e.g. a duplicate PR comment on
-# GitHub). This is parity with every other IM channel's dedupe (same
-# mechanism, same TTL), not a channel-specific gap. True idempotency against
-# a late/manual redelivery would require persisting the dedupe key in
-# the database-backed ``ChannelStore`` instead, which is not implemented here.
+# restart, is no longer recognized as a duplicate here — the key has already
+# been evicted, or never existed in the new process. Its run request still
+# carries the ``Idempotency-Key`` derived from the same dedupe key, so the
+# Gateway hands back the run it admitted for the first delivery (whatever
+# its outcome) rather than running the turn again. Only messages without a
+# dedupe key (no stable provider message id or workspace) can still run twice.
 # Follow-up buffering for busy fire_and_forget threads (issue #4121 Slice 2).
 # A ConflictError on a channel opted into ChannelRunPolicy.buffer_followups_on_busy
 # buffers the triggering message per-thread instead of only logging it; a
@@ -536,10 +555,44 @@ def _is_thread_busy_error(exc: BaseException | None) -> bool:
     if exc is None:
         return False
     if isinstance(exc, ConflictError):
-        return True
+        return not _is_admitted_run_conflict(exc)
     return "already running a task" in str(exc)
 
 
+def _is_admitted_run_conflict(exc: BaseException | None) -> bool:
+    """True for a 409 about this message's own keyed run (see ``_ADMITTED_RUN_CONFLICT_DETAILS``)."""
+    return isinstance(exc, ConflictError) and any(detail in str(exc) for detail in _ADMITTED_RUN_CONFLICT_DETAILS)
+
+
+def _is_run_status_result(result: Any) -> bool:
+    """A ``/wait`` response carrying run status instead of thread state.
+
+    The Gateway answers a reused keyed run this way, because the thread head may
+    already belong to a later run. A fresh run that left no checkpoint looks the
+    same; both are resolved through ``_admitted_turn``.
+    """
+    return isinstance(result, dict) and "messages" not in result and isinstance(result.get("status"), str)
+
+
+def _turn_of_message(values: Any, message_id: str) -> dict[str, Any] | None:
+    """Thread *values* cut after the turn that answered human message *message_id*.
+
+    The cut ends before the next visible human message, so the response, artifact
+    and clarification extractors (which read the turn after the last human
+    message) see this message's turn rather than a later run's.
+    """
+    if not isinstance(values, dict) or not isinstance(messages := values.get("messages"), list):
+        return None
+    for index, message in enumerate(messages):
+        if isinstance(message, dict) and message.get("type") == "human" and message.get("id") == message_id:
+            end = next(
+                (later for later in range(index + 1, len(messages)) if isinstance(messages[later], dict) and messages[later].get("type") == "human" and not _is_hidden_human_control_message(messages[later])),
+                len(messages),
+            )
+            return {**values, "messages": messages[:end]}
+    return None
+
+
 def _is_model_allowlist_error(exc: BaseException | None) -> bool:
     """True when *exc* is the run-admission rejection for an unlisted model.
 
@@ -1132,7 +1185,14 @@ def _unknown_command_reply(command: str | None = None) -> str:
     return f"Unknown command. Available commands: {available}"
 
 
-def _human_input_message_with_channel_response(content: str, *, metadata: Mapping[str, Any], original_content: str | None = None, files: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+def _human_input_message_with_channel_response(
+    content: str,
+    *,
+    metadata: Mapping[str, Any],
+    original_content: str | None = None,
+    files: list[dict[str, Any]] | None = None,
+    message_id: str | None = None,
+) -> dict[str, Any]:
     """``_human_input_message`` plus a channel-synthesized human-input answer.
 
     Card-driven channels (Feishu button clicks) carry the answer in inbound
@@ -1140,7 +1200,7 @@ def _human_input_message_with_channel_response(content: str, *, metadata: Mappin
     Gateway's human_input protocol, hidden from transcripts like Web card
     replies.
     """
-    message = _human_input_message(content, original_content=original_content, files=files)
+    message = _human_input_message(content, original_content=original_content, files=files, message_id=message_id)
     response = metadata.get(HUMAN_INPUT_RESPONSE_METADATA_KEY)
     if isinstance(response, Mapping):
         additional_kwargs = message.setdefault("additional_kwargs", {})
@@ -1149,11 +1209,12 @@ def _human_input_message_with_channel_response(content: str, *, metadata: Mappin
     return message
 
 
-def _human_input_message(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+def _human_input_message(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | None = None, message_id: str | None = None) -> dict[str, Any]:
     # The Gateway records run input as sent, but add_messages gives an id-less
     # message a fresh uuid in the checkpoint. A web client reconnecting to the
     # run rebuilds its input from both copies and can only match them by id.
-    message: dict[str, Any] = {"id": str(uuid.uuid4()), "role": "human", "content": content}
+    # A keyed inbound message passes an id derived from its idempotency key.
+    message: dict[str, Any] = {"id": message_id or str(uuid.uuid4()), "role": "human", "content": content}
     if original_content is not None and original_content != content or files:
         additional_kwargs: dict[str, Any] = {}
         if original_content is not None and original_content != content:
@@ -1209,6 +1270,13 @@ def _owner_headers(msg: InboundMessage) -> dict[str, str] | None:
     return create_internal_auth_headers(owner_user_id=owner_user_id)
 
 
+def _run_request_headers(msg: InboundMessage, idempotency_key: str | None) -> dict[str, str] | None:
+    headers = dict(_owner_headers(msg) or {})
+    if idempotency_key:
+        headers[IDEMPOTENCY_KEY_HEADER] = idempotency_key
+    return headers or None
+
+
 def _safe_user_id_for_run(raw_user_id: str) -> str:
     from deerflow.config.paths import get_paths
 
@@ -1574,6 +1642,9 @@ class ChannelManager:
         # (pre-#4120 behavior). Multi-pod deployments inject a shared store so
         # duplicate deliveries landing on different pods are collapsed.
         self._inbound_dedupe_store = inbound_dedupe_store if inbound_dedupe_store is not None else MemoryInboundDedupeStore()
+        # Adapters claim the same key through the bus before platform side
+        # effects a duplicate must not repeat (Discord thread creation).
+        self.bus.set_inbound_claim(self._claim_inbound, self._release_inbound_claim)
         # 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],
@@ -2322,7 +2393,12 @@ class ChannelManager:
         logger.info("ChannelManager stopped")
 
     async def _discard_pending_inbound(self) -> int:
-        """Release attachments belonging to messages shutdown never dispatched."""
+        """Release attachments, and adapter dedupe claims, of messages shutdown never dispatched.
+
+        An adapter-claimed message (``dedupe_claimed``) recorded its dedupe key
+        before it was queued; dropped unprocessed, it must give the key back so
+        a redelivery on a peer is not refused for the rest of the TTL.
+        """
         discarded = 0
         while True:
             try:
@@ -2331,6 +2407,11 @@ class ChannelManager:
                 return discarded
             try:
                 await cleanup_inbound_files(msg.files)
+                if msg.dedupe_claimed:
+                    try:
+                        await self._release_inbound_dedupe_key(msg)
+                    except Exception:
+                        logger.exception("[Manager] failed to release the dedupe claim of a discarded inbound message")
             finally:
                 self.bus.inbound_task_done()
             discarded += 1
@@ -2359,8 +2440,9 @@ class ChannelManager:
                 try:
                     # Dedupe before logging "received" so a provider retrying an
                     # event N times does not log N accepts. Provider ack side
-                    # effects may still happen before this manager-level dedupe.
-                    if await self._is_duplicate_inbound(msg):
+                    # effects may still happen before this manager-level dedupe,
+                    # unless the adapter claimed the key ahead of them.
+                    if not msg.dedupe_claimed and await self._is_duplicate_inbound(msg):
                         continue
                     dedupe_recorded = self._inbound_dedupe_key(msg) is not None
                     logger.info(
@@ -2424,6 +2506,14 @@ class ChannelManager:
         if message_id is None:
             return None
 
+        if msg.channel_name == "discord":
+            # Message ids are global snowflakes. ``channel_id`` is where the
+            # message was posted, which every instance sees the same way;
+            # ``chat_id`` depends on this instance's thread tracking (a reply
+            # in an untracked thread is routed as an orphan). A direct message
+            # has no guild.
+            return ("discord", str(metadata.get("guild_id") or "dm"), str(metadata.get("channel_id") or msg.chat_id), message_id)
+
         # Fail closed: without a workspace/team/guild identifier we cannot tell two
         # workspaces apart (e.g. Slack channel ids are not globally unique), so
         # skip dedupe rather than risk collapsing distinct workspaces' messages.
@@ -2440,6 +2530,41 @@ class ChannelManager:
             return None
         return (msg.channel_name, str(workspace_id), msg.chat_id, message_id)
 
+    @staticmethod
+    def _run_idempotency_key(msg: InboundMessage) -> str | None:
+        """``Idempotency-Key`` for *msg*'s run, or ``None`` when it has no dedupe key."""
+        key = ChannelManager._inbound_dedupe_key(msg)
+        if key is None:
+            return None
+        digest = hashlib.sha256(json.dumps(list(key), ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
+        return f"{_RUN_IDEMPOTENCY_KEY_PREFIX}{digest}"
+
+    @staticmethod
+    def _human_message_id(msg: InboundMessage) -> str | None:
+        """Human message id stable across redeliveries of *msg*, or ``None`` when unkeyed."""
+        idempotency_key = ChannelManager._run_idempotency_key(msg)
+        return str(uuid.uuid5(_HUMAN_MESSAGE_ID_NAMESPACE, idempotency_key)) if idempotency_key else None
+
+    async def _claim_inbound(self, msg: InboundMessage) -> bool:
+        """Record *msg*'s dedupe key before the adapter's platform side effects.
+
+        Installed as the bus's inbound claim. ``False`` means another delivery
+        (a peer instance's session, an earlier redelivery) owns the message.
+        Once recorded, ``msg.dedupe_claimed`` tells the worker not to record
+        it again; the adapter copies the flag onto the message it publishes.
+        """
+        if self._inbound_dedupe_key(msg) is None:
+            return True
+        if await self._is_duplicate_inbound(msg):
+            return False
+        msg.dedupe_claimed = True
+        return True
+
+    async def _release_inbound_claim(self, msg: InboundMessage) -> None:
+        """Undo ``_claim_inbound`` for a message the adapter never handed off."""
+        if msg.dedupe_claimed:
+            await self._release_inbound_dedupe_key(msg)
+
     async def _is_duplicate_inbound(self, msg: InboundMessage) -> bool:
         key = self._inbound_dedupe_key(msg)
         if key is None:
@@ -2662,6 +2787,19 @@ class ChannelManager:
             user_id=msg.user_id,
         )
 
+    async def _claim_thread_id(self, msg: InboundMessage, thread_id: str) -> str:
+        """Bind *msg*'s conversation to *thread_id* unless bound already; return the bound thread."""
+        if msg.connection_id and msg.owner_user_id and self._connection_repo is not None:
+            return await self._connection_repo.claim_thread_id(
+                connection_id=msg.connection_id,
+                owner_user_id=msg.owner_user_id,
+                provider=msg.channel_name,
+                external_conversation_id=msg.chat_id,
+                external_topic_id=msg.topic_id,
+                thread_id=thread_id,
+            )
+        return await self.store.claim_thread_id(msg.channel_name, msg.chat_id, thread_id, topic_id=msg.topic_id, user_id=msg.user_id)
+
     def _remember_thread_agent(self, thread_id: str, agent_name: str | None) -> None:
         if len(self._thread_agent_names) > IN_PROCESS_CACHE_MAX_ENTRIES:
             self._thread_agent_names.clear()
@@ -2871,7 +3009,61 @@ class ChannelManager:
         *,
         agent_name: str | None = None,
     ) -> str:
-        """Create a new thread through Gateway and store the mapping."""
+        """Create a new thread through Gateway and rebind the conversation to it (``/new``)."""
+        thread_id, _ = await self._create_gateway_thread(client, msg, agent_name=agent_name)
+        await self._store_thread_id(msg, thread_id)
+        self._remember_thread_agent(thread_id, agent_name)
+        logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
+        return thread_id
+
+    async def _create_first_thread(self, client, msg: InboundMessage) -> tuple[str, bool]:
+        """Create the conversation's first thread unless another instance bound one first.
+
+        Returns ``(thread_id, created)``. Two instances that both missed the
+        lookup each create a Gateway thread; the binding is claimed
+        insert-if-absent, so the loser adopts the winner's thread (and deletes
+        the one it created, which nothing references) instead of overwriting
+        the binding and splitting the conversation.
+        """
+        thread_id, minted = await self._create_gateway_thread(client, msg)
+        bound = await self._claim_thread_id(msg, thread_id)
+        if bound != thread_id:
+            logger.info(
+                "[Manager] chat_id=%s topic_id=%s was bound to thread_id=%s concurrently; adopting it instead of thread_id=%s",
+                msg.chat_id,
+                msg.topic_id,
+                bound,
+                thread_id,
+            )
+            if minted:
+                await self._delete_unbound_thread(client, msg, thread_id)
+            return bound, False
+        self._remember_thread_agent(thread_id, None)
+        logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
+        return thread_id, True
+
+    async def _delete_unbound_thread(self, client, msg: InboundMessage, thread_id: str) -> None:
+        delete_kwargs: dict[str, Any] = {}
+        if owner_headers := _owner_headers(msg):
+            delete_kwargs["headers"] = owner_headers
+        try:
+            await client.threads.delete(thread_id, **delete_kwargs)
+        except Exception:
+            logger.warning("[Manager] failed to delete thread_id=%s after losing the conversation binding; it stays empty", thread_id, exc_info=True)
+
+    async def _create_gateway_thread(
+        self,
+        client,
+        msg: InboundMessage,
+        *,
+        agent_name: str | None = None,
+    ) -> tuple[str, bool]:
+        """Create a thread through Gateway; ``(thread_id, minted)``.
+
+        ``minted`` is False for a channel-supplied preferred id: ``POST
+        /threads`` returns an existing thread for it, which may hold history
+        and must never be deleted as a lost first-binding race.
+        """
         metadata = _thread_channel_metadata(msg)
         if agent_name is not None:
             metadata[CHANNEL_AGENT_METADATA_KEY] = agent_name
@@ -2937,14 +3129,8 @@ class ChannelManager:
                 preferred_thread_id,
                 exc.__class__.__name__,
             )
-            await self._store_thread_id(msg, preferred_thread_id)
-            self._remember_thread_agent(preferred_thread_id, agent_name)
-            return preferred_thread_id
-        thread_id = thread["thread_id"]
-        await self._store_thread_id(msg, thread_id)
-        self._remember_thread_agent(thread_id, agent_name)
-        logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
-        return thread_id
+            return preferred_thread_id, False
+        return thread["thread_id"], "thread_id" not in create_kwargs
 
     async def _get_or_create_thread(self, client, msg: InboundMessage) -> tuple[str, bool]:
         """Return ``(thread_id, created)``, creating a thread only if needed.
@@ -2954,7 +3140,8 @@ class ChannelManager:
         thread and then both create one — the second store silently overwrites
         the first, orphaning a Gateway thread and splitting the conversation.
         Serialize the create path per conversation and re-check inside the lock
-        so only the first message creates a thread and the rest reuse it.
+        so only the first message creates a thread and the rest reuse it; across
+        instances the binding claim in ``_create_first_thread`` converges them.
         """
         thread_id = await self._lookup_thread_id(msg)
         if thread_id:
@@ -2967,7 +3154,7 @@ class ChannelManager:
             thread_id = await self._lookup_thread_id(msg)
             if thread_id:
                 return thread_id, False
-            return await self._create_thread(client, msg), True
+            return await self._create_first_thread(client, msg)
 
     async def _update_thread_channel_metadata(self, client, msg: InboundMessage, thread_id: str) -> None:
         """Best-effort source metadata backfill for existing IM-created threads."""
@@ -3070,6 +3257,9 @@ class ChannelManager:
     ) -> None:
         if storage_user_id is None:
             storage_user_id = _channel_storage_user_id(msg)
+        # Derived before ``receive_file`` may rewrite the message.
+        idempotency_key = self._run_idempotency_key(msg)
+        human_message_id = self._human_message_id(msg)
 
         assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id)
         # The pin this run actually submits — stale-pin recovery compares the
@@ -3101,7 +3291,7 @@ class ChannelManager:
 
         original_text = msg.text
         uploaded = await _ingest_inbound_files(thread_id, msg, user_id=storage_user_id)
-        human_message = _human_input_message_with_channel_response(msg.text, metadata=msg.metadata, original_content=original_text, files=uploaded or None)
+        human_message = _human_input_message_with_channel_response(msg.text, metadata=msg.metadata, original_content=original_text, files=uploaded or None, message_id=human_message_id)
 
         if self._channel_supports_streaming(msg.channel_name):
             await self._handle_streaming_chat(
@@ -3113,6 +3303,7 @@ class ChannelManager:
                 run_context,
                 human_message,
                 storage_user_id=storage_user_id,
+                idempotency_key=idempotency_key,
             )
             return
 
@@ -3123,8 +3314,8 @@ class ChannelManager:
             "multitask_strategy": "reject",
         }
         _apply_run_origin(run_kwargs, msg)
-        if owner_headers := _owner_headers(msg):
-            run_kwargs["headers"] = owner_headers
+        if request_headers := _run_request_headers(msg, idempotency_key):
+            run_kwargs["headers"] = request_headers
 
         if policy is not None and policy.fire_and_forget:
             # Fire-and-forget path: the channel does its own outbound
@@ -3152,6 +3343,11 @@ class ChannelManager:
                 # behaviorally identical to the previous bare ``await``.
                 result = await client.runs.create(thread_id, assistant_id, **run_kwargs)
             except Exception as exc:
+                if _is_admitted_run_conflict(exc):
+                    # An earlier delivery's run; it posts its own outcome.
+                    _note_run_confirmed()
+                    logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
+                    return
                 if _is_thread_busy_error(exc):
                     logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
                     if policy.buffer_followups_on_busy:
@@ -3169,7 +3365,10 @@ class ChannelManager:
                     return
                 raise
             _note_run_confirmed()
-            if policy.buffer_followups_on_busy:
+            # A redelivery gets an already-finished run back; its stream may be
+            # gone, so only a live run is worth watching.
+            status = result.get("status") if isinstance(result, dict) else None
+            if policy.buffer_followups_on_busy and (status is None or status in _LIVE_RUN_STATUSES):
                 self._maybe_spawn_followup_watcher(thread_id, result, msg)
             return
 
@@ -3184,6 +3383,11 @@ class ChannelManager:
                 **run_kwargs,
             )
         except Exception as exc:
+            if _is_admitted_run_conflict(exc):
+                _note_run_confirmed()
+                logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
+                await self._send_error(msg, ALREADY_ADMITTED_MESSAGE)
+                return
             if _is_thread_busy_error(exc):
                 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
@@ -3198,6 +3402,14 @@ class ChannelManager:
             raise
         _note_run_confirmed()
 
+        if _is_run_status_result(result):
+            # A redelivery attached to the run admitted for the first delivery.
+            if result["status"] in _LIVE_RUN_STATUSES:
+                logger.info("[Manager] run already admitted for this message is still %s: thread_id=%s", result["status"], thread_id)
+                await self._send_error(msg, ALREADY_ADMITTED_MESSAGE)
+                return
+            result = await self._admitted_turn(client, msg, thread_id) or result
+
         response_text = _extract_response_text(result)
         pending_clarification = _has_current_turn_clarification(result)
         artifacts = _extract_artifacts(result)
@@ -3235,6 +3447,26 @@ class ChannelManager:
         logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id)
         await self.bus.publish_outbound(outbound)
 
+    async def _admitted_turn(self, client, msg: InboundMessage, thread_id: str) -> dict[str, Any] | None:
+        """This message's turn from the thread's durable state, for a reused run.
+
+        A reused run's ``/wait`` answer carries no state, and its replayed
+        stream may be gone (``gap``). The keyed human message id locates the
+        turn the admitted run answered, even when later runs followed it.
+        """
+        message_id = self._human_message_id(msg)
+        if message_id is None:
+            return None
+        get_kwargs: dict[str, Any] = {}
+        if owner_headers := _owner_headers(msg):
+            get_kwargs["headers"] = owner_headers
+        try:
+            state = await client.threads.get_state(thread_id, **get_kwargs)
+        except Exception:
+            logger.warning("[Manager] failed to read thread state for an admitted run: thread_id=%s", thread_id, exc_info=True)
+            return None
+        return _turn_of_message(state.get("values") if isinstance(state, dict) else None, message_id)
+
     async def _handle_streaming_chat(
         self,
         client,
@@ -3245,6 +3477,7 @@ class ChannelManager:
         run_context: dict[str, Any],
         human_message: dict[str, Any],
         storage_user_id: str | None = None,
+        idempotency_key: str | None = None,
     ) -> None:
         logger.info("[Manager] invoking runs.stream(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))
 
@@ -3256,6 +3489,7 @@ class ChannelManager:
         last_published_len = 0
         last_publish_at = 0.0
         stream_error: BaseException | None = None
+        replay_gap = False
         pin_recovery_text: str | None = None
         pinned_model_value = run_context.get("model_name")
         expected_pin = pinned_model_value.strip() if isinstance(pinned_model_value, str) and pinned_model_value.strip() else None
@@ -3268,8 +3502,8 @@ class ChannelManager:
             "on_disconnect": CHANNEL_RUN_ON_DISCONNECT,
         }
         _apply_run_origin(stream_kwargs, msg)
-        if owner_headers := _owner_headers(msg):
-            stream_kwargs["headers"] = owner_headers
+        if request_headers := _run_request_headers(msg, idempotency_key):
+            stream_kwargs["headers"] = request_headers
 
         _note_run_requested()
         try:
@@ -3298,6 +3532,9 @@ class ChannelManager:
                     # The v1 SDK yields backend errors as frames, not exceptions.
                     # Keep draining the stream and use the existing failure path.
                     stream_error = _RunFailedFrame("Agent run failed.")
+                elif event == "gap":
+                    # A reused run whose events are no longer replayable.
+                    replay_gap = True
                 elif event == "values" and isinstance(data, (dict, list)):
                     last_values = data
                     # Clarification text is only in the values snapshot;
@@ -3336,7 +3573,10 @@ class ChannelManager:
                 last_publish_at = now
         except Exception as exc:
             stream_error = exc
-            if _is_thread_busy_error(exc):
+            if _is_admitted_run_conflict(exc):
+                _note_run_confirmed()
+                logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
+            elif _is_thread_busy_error(exc):
                 logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
             else:
                 logger.exception("[Manager] streaming error: thread_id=%s", thread_id)
@@ -3347,6 +3587,8 @@ class ChannelManager:
                 pin_recovery = await self._stale_pin_recovery(msg, exc, expected_pin=expected_pin, thread_id=thread_id)
                 pin_recovery_text = pin_recovery.text if pin_recovery else None
         finally:
+            if replay_gap and last_values is None and not latest_text and stream_error is None:
+                last_values = await self._admitted_turn(client, msg, thread_id)
             result = last_values if last_values is not None else {"messages": [{"type": "ai", "content": latest_text}]}
             response_text = _extract_response_text(result)
             pending_clarification = _has_current_turn_clarification(result)
@@ -3360,7 +3602,9 @@ class ChannelManager:
                 if attachments:
                     response_text = _format_artifact_text([attachment.virtual_path for attachment in attachments])
                 elif stream_error:
-                    if _is_thread_busy_error(stream_error):
+                    if _is_admitted_run_conflict(stream_error):
+                        response_text = ALREADY_ADMITTED_MESSAGE
+                    elif _is_thread_busy_error(stream_error):
                         response_text = THREAD_BUSY_MESSAGE
                     else:
                         response_text = pin_recovery_text or "An error occurred while processing your request. Please try again."
@@ -3760,7 +4004,7 @@ class ChannelManager:
             return "Goal cleared."
 
         if not thread_id:
-            thread_id = await self._create_thread(self._get_client(), msg)
+            thread_id, _ = await self._create_first_thread(self._get_client(), msg)
 
         try:
             await self._goal_request("put", thread_id, headers=headers, json={"objective": command.objective})

backend/app/channels/message_bus.py

@@ -6,7 +6,7 @@ import asyncio
 import logging
 import threading
 import time
-from collections.abc import Callable, Coroutine
+from collections.abc import Awaitable, Callable, Coroutine
 from contextlib import suppress
 from dataclasses import dataclass, field
 from enum import StrEnum
@@ -93,6 +93,9 @@ class InboundMessage:
         files: Optional list of file attachments (platform-specific dicts).
         metadata: Arbitrary extra data from the channel.
         created_at: Unix timestamp when the message was created.
+        dedupe_claimed: Set by ``MessageBus.claim_inbound`` once the
+            manager's dedupe key for this message is recorded, so the worker
+            does not record it again (it would read as a duplicate).
     """
 
     channel_name: str
@@ -108,6 +111,7 @@ class InboundMessage:
     files: list[dict[str, Any]] = field(default_factory=list)
     metadata: dict[str, Any] = field(default_factory=dict)
     created_at: float = field(default_factory=time.time)
+    dedupe_claimed: bool = False
 
 
 @dataclass
@@ -169,6 +173,8 @@ class OutboundMessage:
 # ---------------------------------------------------------------------------
 
 OutboundCallback = Callable[[OutboundMessage], Coroutine[Any, Any, None]]
+InboundClaim = Callable[[InboundMessage], Awaitable[bool]]
+InboundClaimRelease = Callable[[InboundMessage], Awaitable[None]]
 
 
 class InboundQueueFullError(RuntimeError):
@@ -227,6 +233,8 @@ class MessageBus:
         self._full_rejection_count = 0
         self._last_full_warning_at = 0.0
         self._outbound_listeners: list[OutboundCallback] = []
+        self._inbound_claim: InboundClaim | None = None
+        self._inbound_claim_release: InboundClaimRelease | None = None
 
     # -- inbound -----------------------------------------------------------
 
@@ -320,6 +328,29 @@ class MessageBus:
             self._inbound_queued -= 1
         return msg
 
+    def set_inbound_claim(self, claim: InboundClaim, release: InboundClaimRelease) -> None:
+        """Install the dispatcher's dedupe claim (``ChannelManager._claim_inbound``)."""
+        self._inbound_claim = claim
+        self._inbound_claim_release = release
+
+    async def claim_inbound(self, msg: InboundMessage) -> bool:
+        """Claim *msg* against inbound dedupe before a platform side effect.
+
+        For an adapter whose side effect a duplicate delivery must not repeat
+        (Discord creating a thread): ``False`` means another delivery owns the
+        message. On ``True`` the adapter copies ``msg.dedupe_claimed`` onto the
+        message it publishes, and calls ``release_inbound_claim`` if it never
+        publishes one. Without a dispatcher every message may proceed. Must run
+        on the Gateway loop, where the dedupe store's engine lives.
+        """
+        if self._inbound_claim is None:
+            return True
+        return await self._inbound_claim(msg)
+
+    async def release_inbound_claim(self, msg: InboundMessage) -> None:
+        if self._inbound_claim_release is not None and msg.dedupe_claimed:
+            await self._inbound_claim_release(msg)
+
     def inbound_task_done(self) -> None:
         """Mark one dequeued inbound message as fully handled."""
         self._inbound_queue.task_done()

backend/app/channels/store.py

@@ -67,6 +67,10 @@ class ChannelStore(Protocol):
         """Create or update the mapping for an IM conversation/topic (an update keeps ``created_at``)."""
         ...
 
+    async def claim_thread_id(self, channel_name: str, chat_id: str, thread_id: str, *, topic_id: str | None = None, user_id: str = "") -> str:
+        """Map the conversation/topic to ``thread_id`` unless it is mapped already; return the mapped thread."""
+        ...
+
     async def remove(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> bool:
         """Remove one topic mapping, or — without ``topic_id`` — the base mapping and every topic of that chat."""
         ...
@@ -160,6 +164,17 @@ class JsonChannelStore:
             }
             self._save()
 
+    def _claim_sync(self, key: str, thread_id: str, user_id: str) -> str:
+        with self._lock:
+            data = self._ensure_loaded()
+            existing = data.get(key)
+            if existing:
+                return existing["thread_id"]
+            now = time.time()
+            data[key] = {"thread_id": thread_id, "user_id": user_id, "created_at": now, "updated_at": now}
+            self._save()
+            return thread_id
+
     def _remove_sync(self, channel_name: str, chat_id: str, topic_id: str | None) -> bool:
         with self._lock:
             data = self._ensure_loaded()
@@ -203,6 +218,9 @@ class JsonChannelStore:
     async def set_thread_id(self, channel_name: str, chat_id: str, thread_id: str, *, topic_id: str | None = None, user_id: str = "") -> None:
         await asyncio.to_thread(self._set_sync, binding_key(channel_name, chat_id, topic_id), thread_id, user_id)
 
+    async def claim_thread_id(self, channel_name: str, chat_id: str, thread_id: str, *, topic_id: str | None = None, user_id: str = "") -> str:
+        return await asyncio.to_thread(self._claim_sync, binding_key(channel_name, chat_id, topic_id), thread_id, user_id)
+
     async def remove(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> bool:
         return await asyncio.to_thread(self._remove_sync, channel_name, chat_id, topic_id)
 
@@ -232,6 +250,9 @@ class SqlChannelStore:
     async def set_thread_id(self, channel_name: str, chat_id: str, thread_id: str, *, topic_id: str | None = None, user_id: str = "") -> None:
         await self._repo.upsert(ChannelThreadBinding.build(channel_name, chat_id, thread_id, topic_id=topic_id, user_id=user_id, now=time.time()))
 
+    async def claim_thread_id(self, channel_name: str, chat_id: str, thread_id: str, *, topic_id: str | None = None, user_id: str = "") -> str:
+        return await self._repo.claim(ChannelThreadBinding.build(channel_name, chat_id, thread_id, topic_id=topic_id, user_id=user_id, now=time.time()))
+
     async def remove(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> bool:
         if topic_id is not None:
             return await self._repo.delete(binding_key(channel_name, chat_id, topic_id))

backend/app/gateway/AGENTS.md

@@ -212,7 +212,7 @@ or `deployment.multi_instance: true`.
 - Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state.
 - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy.
 - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection.
-- Thread-scoped create, stream and wait share an optional owner/thread-scoped `Idempotency-Key`; never pass the raw external key to `RunManager`'s process-wide index. Stores resolve an existing key before applying `interrupt`/`rollback`, so a retry reuses rather than interrupts its run. Reuse requires matching input, assistant and conversation references; keyed resume stores a versioned strict typed-JSON SHA-256 (`true`, `1` and `1.0` differ) in the private `runs.idempotency_request_json` column, never public `kwargs`, while ordinary input needs no extra identity. New readers accept the exact old-writer resume identity shape from `kwargs` and every response still strips it; truly identity-less legacy resume fails closed. Validate strict resume JSON with or without a key, but persist the identity only for keyed resume admission. Missing headers stay non-idempotent; stateless `/api/runs/*` is outside this contract because requests without a thread create one before admission.
+- Thread-scoped create, stream and wait share an optional owner/thread-scoped `Idempotency-Key` (internal callers: owner-only, cross-thread reuse 409); never pass the raw external key to `RunManager`'s process-wide index. Stores resolve an existing key before `interrupt`/`rollback`, so a retry reuses rather than interrupts its run. Reuse requires matching input, assistant and conversation references; keyed resume stores a versioned strict typed-JSON SHA-256 (`true`, `1` and `1.0` differ) in the private `runs.idempotency_request_json` column, never public `kwargs`; ordinary input needs no extra identity. New readers accept the exact old-writer resume identity shape from `kwargs`, every response strips it; identity-less legacy resume fails closed. Validate strict resume JSON with or without a key; persist the identity only for keyed resume admission. Missing headers stay non-idempotent; stateless `/api/runs/*` is outside this contract (it creates a thread before admission).
 - For a rolling upgrade, route every keyed resume submission/retry on create/stream/wait only to upgraded workers; otherwise pause that traffic or replace old workers first. Old workers ignore the private identity and can reuse `approve` for `deny` with null input. Old-reader schema compatibility does not protect admission. Upgraded workers reject identity-less legacy resumes with 409. Rollout details: `README.md`, `backend/docs/API.md`.
 - Reused `/wait` returns durable status/error, never the latest thread checkpoint; `task is None` is not completion, and store-backed fields must refresh after waiting. Capture reuse before awaiting because the cached record's `idempotency_reused` flag is sticky. A creating `/runs/stream` retry whose terminal stream is gone opts into `gap`/`stream_replay_gap` via `emit_gap_on_missing_stream`, while observer joins and default/stateless consumers emit `end`; never infer this from disconnect policy or the sticky flag. A reused running `store_only` record on a process-local bridge returns 409 without `Retry-After`.
 - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. In `delta` checkpoint mode the worker rewrites that fork into a linear head write before the graph starts (see "A delta-mode run cannot fork" under Checkpoint Channel Modes), because delta state for a fork replays the abandoned sibling's writes.

backend/app/gateway/routers/thread_runs.py

@@ -104,17 +104,26 @@ IdempotencyKeyHeader = Annotated[
 
 
 def _scope_http_run_idempotency_key(request: Request, thread_id: str, key: str | None) -> str | None:
-    """Namespace a caller key for the process-wide run idempotency index."""
+    """Namespace a caller key for the process-wide run idempotency index.
+
+    Browser and API keys are scoped by owner and thread. A trusted internal
+    caller's key is scoped by owner only: the IM channel manager derives it
+    from the provider message, and a redelivery may be requested on another
+    thread once ``/new`` rebound the conversation; it must then meet the run
+    admitted for the first delivery (a 409) instead of starting a second one.
+    """
     if not isinstance(key, str):
         return None
     key = key.strip()
     if not key:
         raise HTTPException(status_code=422, detail="Idempotency-Key must not be blank")
     owner_id = get_trusted_internal_owner_user_id(request)
+    user = getattr(request.state, "user", None)
     if owner_id is None:
-        user = getattr(request.state, "user", None)
         user_id = getattr(user, "id", None)
         owner_id = str(user_id) if user_id is not None else get_effective_user_id()
+    if getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE:
+        return "internal-run:" + hashlib.sha256(f"{owner_id}\0{key}".encode()).hexdigest()
     digest = hashlib.sha256(f"{owner_id}\0{thread_id}\0{key}".encode()).hexdigest()
     return f"http-run:{digest}"
 

backend/app/gateway/services.py

@@ -69,6 +69,7 @@ from deerflow.runtime import (
     ConflictError,
     DisconnectMode,
     RunContext,
+    RunIdempotencyKeyMismatch,
     RunIdempotencyUnsupported,
     RunManager,
     RunRecord,
@@ -2533,6 +2534,11 @@ async def start_run(
                     )
                 except RunIdempotencyUnsupported as exc:
                     raise HTTPException(status_code=503, detail=str(exc)) from exc
+                except RunIdempotencyKeyMismatch as exc:
+                    # A trusted internal key is scoped by owner, not thread: the
+                    # same message requested on another thread (IM ``/new``) is
+                    # a different request for the run already admitted.
+                    raise HTTPException(status_code=409, detail="Idempotency-Key already used with a different request") from exc
 
                 if record.idempotency_reused:
                     stored = record.kwargs or {}

backend/docs/API.md

@@ -266,7 +266,10 @@ resume value, assistant and conversation references. Resume values must be stric
 JSON with finite numbers, with or without an idempotency key; booleans, integers
 and floats remain distinct when comparing keyed retries. Generate a new key for every intentional user
 action; reuse a key only when retrying that same action after an uncertain HTTP
-result. Keys may be at most 255 characters. Stateless `/api/langgraph/runs/*`
+result. Keys may be at most 255 characters. Trusted internal callers (the IM
+channel manager) get keys scoped by user only: the same key sent for another
+thread returns 409 instead of starting a run, because a redelivered IM message
+may be requested on a new thread after `/new`. Stateless `/api/langgraph/runs/*`
 endpoints do not support this header because requests without an explicit thread
 create a new temporary conversation.
 

backend/docs/CONFIGURATION.md

@@ -826,6 +826,7 @@ Every IM channel except GitHub connects outbound to its platform (Telegram long
 - The block is validated on every topology: an unknown key, a non-mapping value, `enabled` outside `auto | true | false`, `lease_seconds < 3` or `grace_seconds < 0` stop startup with the dotted path in the message.
 - Restart-required (`channels.leader_lease` in the reload-boundary registry): the gate evaluates it once and `ChannelService` captures it at construction. Change it on every instance and restart them together.
 - Known limits: a leader change loses the IM replies in flight on the old holder. The runs themselves continue on the Gateway executing them (channel runs are created with `on_disconnect: continue`) and their thread state is persisted, so the answer stays readable in the web UI, but the transport-local state that would deliver it (a Telegram placeholder message, a Feishu running card, a DingTalk replier) is gone with the old holder; a run executing on the instance that died is recovered as an error like any run of a dead worker.
+- A provider redelivery that reaches an instance after the 10-minute inbound dedupe window, or while the shared dedupe store is unavailable, does not run the turn again: IM run requests carry an `Idempotency-Key` derived from the message, so the Gateway returns the run it already admitted. First messages racing on two instances converge on one thread. See [Duplicate and late deliveries](IM_CHANNEL_CONNECTIONS.md#duplicate-and-late-deliveries).
 
 ### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`)
 

backend/docs/IM_CHANNEL_CONNECTIONS.md

@@ -720,6 +720,43 @@ executing the turn twice; a request the Gateway rejected, or a failure before
 the request was sent, still releases the key for a retry. See
 [IM channel leader lease](CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease).
 
+### Duplicate and late deliveries
+
+The inbound dedupe key is `(channel, workspace, chat, message id)`; with a
+PostgreSQL database every instance shares it for 10 minutes, and it fails open
+on a database error. Discord keys a message by its guild (`dm` for a direct
+message, which used to get no key at all) and by the channel it was posted in,
+so two instances that route a reply in a thread differently still see one
+message. A Discord session on every instance receives every message, so the
+adapter claims that key before it creates a Discord thread for the message:
+only the instance that wins the claim creates one. A claimed message that is
+never processed (a failed or cancelled hand-off, a shutdown that discards it
+from the queue) gives the key back, so a peer can still take it.
+
+Past the dedupe window, after a restart, or while the dedupe store is down, a
+redelivery still reaches the run request. Every run request for a message
+with a dedupe key carries an `Idempotency-Key` derived from it, and the human
+message id is derived from the same key, so the Gateway returns the run it
+admitted for the first delivery instead of starting another one (on a
+`memory` database, only while that Gateway process still holds the run). The
+Gateway scopes these internal keys by owner rather than by thread, so a
+redelivery that arrives after `/new` rebound the conversation to a fresh thread
+gets a 409 for the admitted run, answered with the same note, instead of
+running again there. The
+redelivery then attaches to that run: a streaming channel replays its events,
+or, once they are gone, the manager reads this message's turn from the thread
+state; a `runs.wait` channel reads that turn as well, and a run that is still
+in progress is answered with a short "already being processed" note. A
+redelivered message is not run again even when its first run failed; a new
+message is. Messages without a dedupe key (no stable message id or workspace)
+are not covered.
+
+The first message of a new conversation binds it to a thread with an
+insert-if-absent claim on `channel_conversations` (connected users) or
+`channel_thread_bindings` (unbound chats). An instance that created a thread
+but lost the claim adopts the winner's thread and deletes its own, empty one;
+`/new` still rebinds the conversation to a fresh thread.
+
 ## Security Notes
 
 - Browser APIs remain authenticated and CSRF-protected.

backend/packages/harness/deerflow/persistence/channel_connections/sql.py

@@ -12,6 +12,8 @@ from typing import Any
 
 from cryptography.fernet import Fernet, InvalidToken
 from sqlalchemy import delete, func, select, text, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -601,6 +603,54 @@ class ChannelConnectionRepository:
                     await session.rollback()
             raise last_error  # type: ignore[misc]  # loop runs at least once
 
+    async def claim_thread_id(
+        self,
+        *,
+        connection_id: str,
+        owner_user_id: str,
+        provider: str,
+        external_conversation_id: str,
+        thread_id: str,
+        external_topic_id: str | None = None,
+    ) -> str:
+        """Bind a conversation to ``thread_id`` unless it is bound already; return its thread.
+
+        The first binding of a conversation, unlike ``set_thread_id``'s
+        rebinding (``/new``): two Gateway instances that both created a thread
+        for one new conversation must converge on one of them. A single
+        ``INSERT ... ON CONFLICT DO NOTHING`` lets the first writer win, and
+        the SELECT in the same transaction returns whichever row won (Postgres
+        READ COMMITTED reads the peer's committed row; SQLite serializes
+        writers).
+        """
+        topic_id = external_topic_id or ""
+        async with self.session_factory() as session:
+            insert = pg_insert if session.get_bind().dialect.name == "postgresql" else sqlite_insert
+            await session.execute(
+                insert(ChannelConversationRow)
+                .values(
+                    id=self._new_id(),
+                    connection_id=connection_id,
+                    owner_user_id=owner_user_id,
+                    provider=provider,
+                    external_conversation_id=external_conversation_id,
+                    external_topic_id=topic_id,
+                    thread_id=thread_id,
+                )
+                .on_conflict_do_nothing(index_elements=["connection_id", "external_conversation_id", "external_topic_id"])
+            )
+            bound = (
+                await session.execute(
+                    select(ChannelConversationRow.thread_id).where(
+                        ChannelConversationRow.connection_id == connection_id,
+                        ChannelConversationRow.external_conversation_id == external_conversation_id,
+                        ChannelConversationRow.external_topic_id == topic_id,
+                    )
+                )
+            ).scalar_one()
+            await session.commit()
+            return bound
+
     async def get_thread_id(
         self,
         connection_id: str,

backend/packages/harness/deerflow/persistence/channel_thread_bindings/sql.py

@@ -7,6 +7,10 @@ Concurrency contract, shared with ``app.channels.store.SqlChannelStore``:
   existing ``created_at`` — the overwrite semantics the JSON store had — so two
   replicas recording the same conversation never produce two rows and never
   lose the first stamp.
+- ``claim`` is the first binding of a conversation: ``INSERT ... ON CONFLICT
+  DO NOTHING`` plus a read of the winning row in the same transaction, so two
+  replicas that each created a thread for one new conversation converge on
+  the first one bound.
 - ``insert_missing`` is the one-time import primitive: ``INSERT ... ON CONFLICT
   DO NOTHING`` in bounded batches, returning only the rows it actually inserted,
   so two replicas importing the same ``store.json`` concurrently can neither
@@ -97,6 +101,14 @@ class SqlChannelThreadBindingRepository:
             await session.execute(stmt)
             await session.commit()
 
+    async def claim(self, binding: ChannelThreadBinding) -> str:
+        """Insert the binding unless its key exists; return the ``thread_id`` the key is bound to."""
+        async with self._sf() as session:
+            await session.execute(_insert_for(session)(ChannelThreadBindingRow).values(**_values(binding)).on_conflict_do_nothing(index_elements=[ChannelThreadBindingRow.key]))
+            bound = (await session.execute(select(ChannelThreadBindingRow.thread_id).where(ChannelThreadBindingRow.key == binding.key))).scalar_one()
+            await session.commit()
+            return bound
+
     async def delete(self, key: str) -> bool:
         async with self._sf() as session:
             result = await session.execute(delete(ChannelThreadBindingRow).where(ChannelThreadBindingRow.key == key))

backend/packages/harness/deerflow/runtime/__init__.py

@@ -14,6 +14,7 @@ from .runs import (
     ConflictError,
     DisconnectMode,
     RunContext,
+    RunIdempotencyKeyMismatch,
     RunIdempotencyUnsupported,
     RunManager,
     RunRecord,
@@ -45,6 +46,7 @@ __all__ = [
     "DisconnectMode",
     "ORPHAN_RECOVERY_STOP_REASON",
     "RunContext",
+    "RunIdempotencyKeyMismatch",
     "RunIdempotencyUnsupported",
     "RunManager",
     "RunRecord",

backend/packages/harness/deerflow/runtime/runs/__init__.py

@@ -1,6 +1,6 @@
 """Run lifecycle management for LangGraph Platform API compatibility."""
 
-from .manager import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, RunIdempotencyUnsupported, RunManager, RunRecord, UnsupportedStrategyError
+from .manager import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, RunIdempotencyKeyMismatch, RunIdempotencyUnsupported, RunManager, RunRecord, UnsupportedStrategyError
 from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 from .worker import RunContext, run_agent
 
@@ -10,6 +10,7 @@ __all__ = [
     "DisconnectMode",
     "ORPHAN_RECOVERY_STOP_REASON",
     "RunContext",
+    "RunIdempotencyKeyMismatch",
     "RunIdempotencyUnsupported",
     "RunManager",
     "RunRecord",

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -188,6 +188,10 @@ class RunIdempotencyUnsupported(RuntimeError):
     """The configured store cannot safely persist the requested idempotency fields."""
 
 
+class RunIdempotencyKeyMismatch(RuntimeError):
+    """The idempotency key belongs to a run of another thread or user: a different request."""
+
+
 def _store_accepts_idempotency_parameter(store: RunStore, parameter_name: str) -> bool:
     """Require an explicit keyword parameter or an opt-in owned by the method."""
     callable_ = store.create_thread_operation_atomic
@@ -2063,7 +2067,7 @@ class RunManager:
                     if existing.idempotency_key != idempotency_key:
                         continue
                     if existing.thread_id != thread_id or existing.user_id != user_id:
-                        raise RuntimeError("Run idempotency key resolved to a different thread or user")
+                        raise RunIdempotencyKeyMismatch("Run idempotency key resolved to a different thread or user")
                     existing.idempotency_reused = True
                     return existing
 
@@ -2077,7 +2081,7 @@ class RunManager:
                 # for get(), cancel(), and orphan reconciliation.
                 existing = self._record_from_store(conflict.existing)
                 if existing.thread_id != thread_id or existing.user_id != user_id:
-                    raise RuntimeError("Run idempotency key resolved to a different thread or user") from conflict
+                    raise RunIdempotencyKeyMismatch("Run idempotency key resolved to a different thread or user") from conflict
                 existing.idempotency_reused = True
                 return existing
 

backend/tests/blocking_io/test_channel_thread_binding_store.py

@@ -69,8 +69,10 @@ async def store(request, tmp_path) -> AsyncIterator[ChannelStore]:
 
 
 async def test_store_round_trip_does_not_block_loop(store) -> None:
-    """get / set / overwrite / list / remove on the loop, including the JSON file's atomic rewrite."""
+    """get / claim / set / overwrite / list / remove on the loop, including the JSON file's atomic rewrite."""
     assert await store.get_thread_id("slack", "C1") is None
+    assert await store.claim_thread_id("slack", "C1", "thread-0", user_id="U1") == "thread-0"
+    assert await store.claim_thread_id("slack", "C1", "thread-peer", user_id="U1") == "thread-0"
     await store.set_thread_id("slack", "C1", "thread-1", user_id="U1")
     await store.set_thread_id("slack", "C1", "thread-1b", topic_id="171.1", user_id="U1")
     await store.set_thread_id("slack", "C1", "thread-2", user_id="U2")  # overwrite
@@ -88,6 +90,7 @@ async def test_manager_lookup_and_store_do_not_block_loop(store) -> None:
     msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="hi", topic_id="171.1")
     assert await manager._lookup_thread_id(msg) is None
     assert await lookup_thread_id(msg, repo=None, store=store) is None
+    assert await manager._claim_thread_id(msg, "thread-0") == "thread-0"
     await manager._store_thread_id(msg, "thread-1")
     assert await manager._lookup_thread_id(msg) == "thread-1"
     assert await lookup_thread_id(msg, repo=None, store=store) == "thread-1"

backend/tests/test_channel_inbound_idempotency.py

@@ -0,0 +1,520 @@
+"""IM inbound dedupe and run idempotency across replicas, restarts and late redeliveries.
+
+The inbound dedupe store absorbs a redelivery only inside its 10-minute window
+and only while the store remembers the key: an in-process store forgets it on
+restart, and the shared Postgres store fails open on a database error. Past
+that window the manager used to start a second run for a message the Gateway
+had already admitted. Every keyed run request now carries an ``Idempotency-Key``
+derived from the inbound dedupe key, with a human message id derived from the
+same key so the retry's input is identical: the Gateway reuses the admitted run
+and the redelivery attaches to its outcome instead of running the turn twice.
+
+Discord gets a dedupe key for direct messages (no guild) and one that does not
+depend on replica-local thread routing, and two replicas that each create a
+thread for one new conversation converge on the first one bound.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import tempfile
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+from langgraph_sdk.errors import ConflictError
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.channels.dedupe_store import MemoryInboundDedupeStore
+from app.channels.manager import ALREADY_ADMITTED_MESSAGE, THREAD_BUSY_MESSAGE, ChannelManager
+from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage
+from app.channels.store import JsonChannelStore, SqlChannelStore
+from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME
+from deerflow.persistence.base import Base
+from deerflow.persistence.channel_thread_bindings import ChannelThreadBindingRow
+
+pytestmark = pytest.mark.asyncio
+
+
+def _manager(*, dedupe: MemoryInboundDedupeStore | None = None, store=None, get_stream_bridge=None) -> ChannelManager:
+    return ChannelManager(
+        bus=MessageBus(),
+        store=store or JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"),
+        inbound_dedupe_store=dedupe or MemoryInboundDedupeStore(),
+        max_concurrency=1,
+        shutdown_grace_period_seconds=0,
+        get_stream_bridge=get_stream_bridge,
+    )
+
+
+def _client(thread_id: str = "thread-1") -> MagicMock:
+    client = MagicMock()
+    client.threads.create = AsyncMock(return_value={"thread_id": thread_id})
+    client.threads.update = AsyncMock(return_value={"thread_id": thread_id})
+    client.threads.get = AsyncMock(return_value={"thread_id": thread_id})
+    client.threads.delete = AsyncMock(return_value=None)
+    client.runs.wait = AsyncMock(return_value={"messages": [{"type": "human", "content": "hi"}, {"type": "ai", "content": "fresh answer"}]})
+    client.runs.create = AsyncMock(return_value={"run_id": "run-1", "status": "pending"})
+    return client
+
+
+def _slack(event_id: str = "evt-1", *, team_id: str | None = "T1") -> InboundMessage:
+    metadata = {"event_id": event_id}
+    if team_id:
+        metadata["team_id"] = team_id
+    return InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="hi", metadata=metadata)
+
+
+def _feishu() -> InboundMessage:
+    return InboundMessage(channel_name="feishu", chat_id="oc-1", user_id="ou-1", text="hi", metadata={"message_id": "om-1"})
+
+
+def _github() -> InboundMessage:
+    return InboundMessage(
+        channel_name="github",
+        chat_id="acme/repo",
+        user_id="alice",
+        owner_user_id="alice",
+        text="@bot review",
+        topic_id="7:reviewer",
+        workspace_id="acme/repo",
+        metadata={"message_id": "d1:alice:reviewer", "agent_name": "reviewer"},
+    )
+
+
+def _discord(*, chat_id: str, channel_id: str, guild_id: str | None, message_id: str = "1300000000000000001", topic_id: str | None = None) -> InboundMessage:
+    return InboundMessage(
+        channel_name="discord",
+        chat_id=chat_id,
+        user_id="42",
+        text="hi",
+        topic_id=topic_id,
+        metadata={"guild_id": guild_id, "channel_id": channel_id, "message_id": message_id},
+    )
+
+
+def _outbounds(manager: ChannelManager) -> list[OutboundMessage]:
+    sent: list[OutboundMessage] = []
+
+    async def capture(msg: OutboundMessage) -> None:
+        sent.append(msg)
+
+    manager.bus.subscribe_outbound(capture)
+    return sent
+
+
+def _stream(*parts):
+    def factory(*_args, **_kwargs):
+        async def gen():
+            for event, data in parts:
+                yield SimpleNamespace(event=event, data=data)
+
+        return gen()
+
+    return MagicMock(side_effect=factory)
+
+
+def _request_headers(call) -> dict:
+    return dict(call.kwargs.get("headers") or {})
+
+
+def _human_message(call) -> dict:
+    [message] = call.kwargs["input"]["messages"]
+    return message
+
+
+# -- Discord dedupe key --------------------------------------------------------
+
+
+async def test_discord_direct_message_gets_a_dedupe_key():
+    """A DM has no guild, so it used to get no key at all and every replica ran it."""
+    dm = _discord(chat_id="1200000000000000009", channel_id="1200000000000000009", guild_id=None)
+
+    assert ChannelManager._inbound_dedupe_key(dm) == ("discord", "dm", "1200000000000000009", "1300000000000000001")
+
+
+async def test_discord_key_does_not_depend_on_replica_local_thread_routing():
+    """A replica that tracks thread T routes a reply in T to its parent channel;
+    one that does not treats T as an orphan and routes it to T itself. Both must
+    see one message."""
+    tracked = _discord(chat_id="parent-1", channel_id="thread-T", guild_id="guild-1", topic_id="thread-T")
+    orphaned = _discord(chat_id="thread-T", channel_id="thread-T", guild_id="guild-1", topic_id="thread-new")
+
+    assert ChannelManager._inbound_dedupe_key(tracked) == ChannelManager._inbound_dedupe_key(orphaned)
+    assert ChannelManager._inbound_dedupe_key(tracked) == ("discord", "guild-1", "thread-T", "1300000000000000001")
+
+
+async def test_discord_guild_channel_key_keeps_its_shape():
+    msg = _discord(chat_id="chan-1", channel_id="chan-1", guild_id="guild-1")
+
+    assert ChannelManager._inbound_dedupe_key(msg) == ("discord", "guild-1", "chan-1", "1300000000000000001")
+
+
+async def test_worker_runs_a_message_the_adapter_already_claimed_and_drops_its_redelivery():
+    """An adapter-claimed message has its key recorded already; the worker must
+    not read its own claim as a duplicate, but a second delivery still is one."""
+    manager = _manager()
+    manager._client = client = _client()
+    claimed = _discord(chat_id="chan-1", channel_id="chan-1", guild_id="guild-1")
+    assert await manager.bus.claim_inbound(claimed) is True
+    assert claimed.dedupe_claimed is True
+    redelivery = _discord(chat_id="chan-1", channel_id="chan-1", guild_id="guild-1")
+    assert await manager.bus.claim_inbound(redelivery) is False
+
+    await manager.start()
+    try:
+        await manager.bus.publish_inbound(claimed)
+        await manager.bus.publish_inbound(_discord(chat_id="chan-1", channel_id="chan-1", guild_id="guild-1"))
+        await asyncio.wait_for(manager.bus.join_inbound(), timeout=5)
+    finally:
+        await manager.stop()
+
+    assert client.runs.wait.await_count == 1
+
+
+@pytest.mark.parametrize("cancelled", [False, True], ids=["grace-expired", "shutdown-cancelled"])
+async def test_shutdown_gives_back_the_claim_of_a_queued_message_it_discards(cancelled):
+    """An adapter-claimed message waiting in the queue owns a dedupe key no
+    worker recorded. Shutdown discards it unprocessed, so it must release the
+    key, or a peer drops the redelivery for the rest of the TTL."""
+    dedupe = MemoryInboundDedupeStore()
+    manager = _manager(dedupe=dedupe)
+    manager._client = client = _client()
+    busy = asyncio.Event()
+
+    async def run_forever(*_args, **_kwargs):
+        busy.set()
+        await asyncio.Event().wait()
+
+    client.runs.wait = AsyncMock(side_effect=run_forever)
+    await manager.start()
+    await manager.bus.publish_inbound(_slack("evt-busy"))
+    await asyncio.wait_for(busy.wait(), timeout=5)
+    queued = _discord(chat_id="chan-1", channel_id="chan-1", guild_id="guild-1")
+    assert await manager.bus.claim_inbound(queued) is True
+    await manager.bus.publish_inbound(queued)
+
+    if cancelled:
+        manager._shutdown_grace_period_seconds = 30
+        stopping = asyncio.create_task(manager.stop())
+        await asyncio.sleep(0)  # stop() is now waiting out the grace period
+        stopping.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await stopping
+    else:
+        await manager.stop()
+
+    assert ChannelManager._inbound_dedupe_key(queued) not in dedupe._store
+    if cancelled:
+        await manager.stop()
+
+
+# -- Idempotency-Key on IM run requests ----------------------------------------
+
+
+async def test_wait_request_carries_an_idempotency_key_stable_across_redeliveries():
+    clients = []
+    for _ in range(2):
+        # A fresh dedupe store per delivery: the window expired, or the process restarted.
+        manager = _manager()
+        manager._client = client = _client()
+        await manager._handle_message(_slack())
+        clients.append(client)
+
+    first, retry = (client.runs.wait.call_args for client in clients)
+    key = _request_headers(first)["Idempotency-Key"]
+    assert key == ChannelManager._run_idempotency_key(_slack())
+    assert _request_headers(retry)["Idempotency-Key"] == key
+    # The Gateway reuses a keyed run only for an identical input, so the
+    # human message id must be the same on the redelivery too.
+    assert _human_message(first) == _human_message(retry)
+    uuid.UUID(_human_message(first)["id"])
+
+
+async def test_distinct_messages_get_distinct_keys_and_message_ids():
+    manager = _manager()
+    manager._client = client = _client()
+
+    await manager._handle_message(_slack("evt-1"))
+    await manager._handle_message(_slack("evt-2"))
+
+    first, second = client.runs.wait.call_args_list
+    assert _request_headers(first)["Idempotency-Key"] != _request_headers(second)["Idempotency-Key"]
+    assert _human_message(first)["id"] != _human_message(second)["id"]
+
+
+async def test_message_without_a_dedupe_key_sends_no_idempotency_key():
+    manager = _manager()
+    manager._client = client = _client()
+
+    await manager._handle_message(_slack(team_id=None))
+    await manager._handle_message(_slack(team_id=None))
+
+    first, second = client.runs.wait.call_args_list
+    assert "Idempotency-Key" not in _request_headers(first)
+    assert _human_message(first)["id"] != _human_message(second)["id"]
+
+
+async def test_stream_request_carries_the_idempotency_key():
+    manager = _manager()
+    manager._client = client = _client()
+    client.runs.stream = _stream(("values", {"messages": [{"type": "human", "content": "hi"}, {"type": "ai", "content": "streamed"}]}))
+
+    await manager._handle_message(_feishu())
+
+    assert _request_headers(client.runs.stream.call_args)["Idempotency-Key"] == ChannelManager._run_idempotency_key(_feishu())
+
+
+async def test_fire_and_forget_create_carries_the_idempotency_key():
+    import app.gateway.github.run_policy  # noqa: F401 — registers GitHub's fire-and-forget policy
+
+    manager = _manager()
+    manager._client = client = _client()
+
+    await manager._handle_message(_github())
+
+    assert _request_headers(client.runs.create.call_args)["Idempotency-Key"] == ChannelManager._run_idempotency_key(_github())
+
+
+async def test_idempotency_key_rides_along_with_the_owner_headers():
+    import app.gateway.github.run_policy  # noqa: F401
+
+    msg = _github()
+    manager = _manager()
+    manager._client = client = _client()
+
+    await manager._handle_message(msg)
+
+    headers = _request_headers(client.runs.create.call_args)
+    assert headers["Idempotency-Key"] == ChannelManager._run_idempotency_key(msg)
+    assert headers.get(INTERNAL_OWNER_USER_ID_HEADER_NAME) == "alice"
+
+
+# -- attaching a redelivery to the admitted run --------------------------------
+
+
+def _state_with_turns(manager_msg: InboundMessage) -> dict:
+    """Thread state where this message's turn sits between an older and a later turn."""
+    ours = str(ChannelManager._human_message_id(manager_msg))
+    return {
+        "values": {
+            "messages": [
+                {"type": "human", "id": "older", "content": "older question"},
+                {"type": "ai", "content": "older answer"},
+                {"type": "human", "id": ours, "content": "hi"},
+                {"type": "ai", "content": "the admitted run's answer"},
+                {"type": "human", "id": "later", "content": "a later question"},
+                {"type": "ai", "content": "a later answer"},
+            ]
+        }
+    }
+
+
+async def test_reused_wait_run_replies_with_this_messages_turn_from_thread_state():
+    """``/wait`` on a reused run returns status only (the thread head may be a
+    later run). The manager reads this message's own turn instead."""
+    msg = _slack()
+    manager = _manager()
+    manager._client = client = _client()
+    client.runs.wait = AsyncMock(return_value={"status": "success", "error": None})
+    client.threads.get_state = AsyncMock(return_value=_state_with_turns(msg))
+    sent = _outbounds(manager)
+
+    await manager._handle_message(msg)
+
+    assert [out.text for out in sent] == ["the admitted run's answer"]
+
+
+async def test_reused_wait_run_still_in_flight_says_it_is_already_handled_and_keeps_the_key():
+    msg = _slack()
+    dedupe = MemoryInboundDedupeStore()
+    manager = _manager(dedupe=dedupe)
+    manager._client = client = _client()
+    client.runs.wait = AsyncMock(return_value={"status": "running", "error": None})
+    sent = _outbounds(manager)
+    assert await manager._is_duplicate_inbound(msg) is False
+
+    await manager._handle_message(msg)
+
+    assert [out.text for out in sent] == [ALREADY_ADMITTED_MESSAGE]
+    assert ChannelManager._inbound_dedupe_key(msg) in dedupe._store
+
+
+async def test_reused_stream_with_a_replay_gap_replies_from_thread_state():
+    msg = _feishu()
+    manager = _manager()
+    manager._client = client = _client()
+    client.runs.stream = _stream(("gap", {"code": "stream_replay_gap", "recovery": "reload_durable_state"}))
+    client.threads.get_state = AsyncMock(return_value=_state_with_turns(msg))
+    sent = _outbounds(manager)
+
+    await manager._handle_message(msg)
+
+    assert [out.text for out in sent if out.is_final] == ["the admitted run's answer"]
+
+
+@pytest.mark.parametrize(
+    "detail",
+    [
+        "Idempotency-Key already used with a different request",
+        "Run 3f2a is not active on this worker and cannot be streamed",
+    ],
+    ids=["input-mismatch", "foreign-active-run"],
+)
+async def test_conflict_from_an_admitted_keyed_run_is_not_reported_as_a_busy_thread(detail):
+    """A 409 that names the keyed run (re-ingested attachments changed the
+    input; the run is live on a peer) means the message was already admitted:
+    keep the key and do not tell the user to retry."""
+    msg = _slack()
+    dedupe = MemoryInboundDedupeStore()
+    manager = _manager(dedupe=dedupe)
+    manager._client = client = _client()
+    request = httpx.Request("POST", "http://gateway/api/threads/thread-1/runs/wait")
+    client.runs.wait = AsyncMock(side_effect=ConflictError(detail, response=httpx.Response(409, request=request), body={"detail": detail}))
+    sent = _outbounds(manager)
+    assert await manager._is_duplicate_inbound(msg) is False
+
+    await manager._handle_message(msg)
+
+    assert [out.text for out in sent] == [ALREADY_ADMITTED_MESSAGE]
+    assert ChannelManager._inbound_dedupe_key(msg) in dedupe._store
+
+
+async def test_busy_thread_conflict_keeps_its_reply_and_releases_the_key():
+    msg = _slack()
+    dedupe = MemoryInboundDedupeStore()
+    manager = _manager(dedupe=dedupe)
+    manager._client = client = _client()
+    request = httpx.Request("POST", "http://gateway/api/threads/thread-1/runs/wait")
+    detail = "Thread thread-1 already has an active run"
+    client.runs.wait = AsyncMock(side_effect=ConflictError(detail, response=httpx.Response(409, request=request), body={"detail": detail}))
+    sent = _outbounds(manager)
+    assert await manager._is_duplicate_inbound(msg) is False
+
+    await manager._handle_message(msg)
+
+    assert [out.text for out in sent] == [THREAD_BUSY_MESSAGE]
+    assert ChannelManager._inbound_dedupe_key(msg) not in dedupe._store
+
+
+@pytest.mark.parametrize(("status", "watched"), [("pending", True), ("running", True), ("success", False), ("error", False)])
+async def test_followup_watcher_only_follows_a_live_run(status, watched):
+    """A redelivered GitHub event gets the admitted run back; once that run has
+    ended its stream may be gone, and a watcher would wait on it forever."""
+    import app.gateway.github.run_policy  # noqa: F401
+
+    manager = _manager(get_stream_bridge=lambda: MagicMock())
+    manager._client = client = _client()
+    client.runs.create = AsyncMock(return_value={"run_id": "run-1", "status": status})
+    spawned: list[str] = []
+    manager._watch_run_and_drain_followups = AsyncMock(side_effect=lambda thread_id, run_id, msg: spawned.append(run_id))
+
+    await manager._handle_message(_github())
+    await asyncio.sleep(0)
+
+    assert spawned == (["run-1"] if watched else [])
+
+
+async def test_redelivery_after_new_rebinds_the_conversation_meets_the_admitted_run(tmp_path):
+    """``/new`` points the conversation at a fresh thread, so a late redelivery
+    is requested there. It carries the same key, which the Gateway scopes by
+    owner only for this internal caller, and its 409 must not start a run or
+    read as a busy thread."""
+    engine = await _shared_sql_store(tmp_path)
+    try:
+        store = SqlChannelStore(async_sessionmaker(engine, expire_on_commit=False))
+        msg = _slack()
+        first = _manager(store=store)
+        first._client = first_client = _client("thread-old")
+        await first._handle_message(msg)
+        await first._create_thread(_client("thread-new"), _slack("evt-new-command"))  # /new
+
+        dedupe = MemoryInboundDedupeStore()  # the dedupe window expired
+        redelivery = _manager(dedupe=dedupe, store=store)
+        redelivery._client = client = _client("thread-unused")
+        request = httpx.Request("POST", "http://gateway/api/threads/thread-new/runs/wait")
+        detail = "Idempotency-Key already used with a different request"
+        client.runs.wait = AsyncMock(side_effect=ConflictError(detail, response=httpx.Response(409, request=request), body={"detail": detail}))
+        sent = _outbounds(redelivery)
+        assert await redelivery._is_duplicate_inbound(msg) is False
+
+        await redelivery._handle_message(msg)
+
+        assert client.runs.wait.call_args.args[0] == "thread-new"
+        assert _request_headers(client.runs.wait.call_args)["Idempotency-Key"] == _request_headers(first_client.runs.wait.call_args)["Idempotency-Key"]
+        assert [out.text for out in sent] == [ALREADY_ADMITTED_MESSAGE]
+        assert ChannelManager._inbound_dedupe_key(msg) in dedupe._store
+    finally:
+        await engine.dispose()
+
+
+# -- converging on one thread across replicas ----------------------------------
+
+
+async def _shared_sql_store(tmp_path: Path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{(tmp_path / 'bindings.db').as_posix()}")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all, tables=[ChannelThreadBindingRow.__table__])
+    return engine
+
+
+async def test_replica_that_lost_the_first_binding_adopts_the_winner_and_discards_its_thread(tmp_path):
+    engine = await _shared_sql_store(tmp_path)
+    try:
+        factory = async_sessionmaker(engine, expire_on_commit=False)
+        manager_a, manager_b = _manager(store=SqlChannelStore(factory)), _manager(store=SqlChannelStore(factory))
+        client_a, client_b = _client("thread-a"), _client("thread-b")
+        msg = _slack()
+
+        assert await manager_a._get_or_create_thread(client_a, msg) == ("thread-a", True)
+        # B looked the conversation up before A's binding committed.
+        manager_b._lookup_thread_id = AsyncMock(return_value=None)
+        assert await manager_b._get_or_create_thread(client_b, msg) == ("thread-a", False)
+
+        client_b.threads.delete.assert_awaited_once()
+        assert client_b.threads.delete.await_args.args == ("thread-b",)
+        assert await SqlChannelStore(factory).get_thread_id("slack", "C1") == "thread-a"
+    finally:
+        await engine.dispose()
+
+
+async def test_new_command_still_rebinds_the_conversation(tmp_path):
+    engine = await _shared_sql_store(tmp_path)
+    try:
+        store = SqlChannelStore(async_sessionmaker(engine, expire_on_commit=False))
+        manager = _manager(store=store)
+        msg = _slack()
+        await manager._get_or_create_thread(_client("thread-first"), msg)
+
+        assert await manager._create_thread(_client("thread-after-new"), msg) == "thread-after-new"
+
+        assert await store.get_thread_id("slack", "C1") == "thread-after-new"
+    finally:
+        await engine.dispose()
+
+
+async def test_connection_bound_conversation_converges_through_the_repository(tmp_path):
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'connections.db'}", sqlite_dir=str(tmp_path))
+    try:
+        repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("k"))
+        connection = await repo.upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U1", workspace_id="T1")
+        msg = _slack()
+        msg.connection_id, msg.owner_user_id = connection["id"], "alice"
+        manager_a, manager_b = _manager(), _manager()
+        manager_a._connection_repo = manager_b._connection_repo = repo
+        client_b = _client("thread-b")
+
+        assert await manager_a._get_or_create_thread(_client("thread-a"), msg) == ("thread-a", True)
+        manager_b._lookup_thread_id = AsyncMock(return_value=None)
+        assert await manager_b._get_or_create_thread(client_b, msg) == ("thread-a", False)
+
+        assert client_b.threads.delete.await_args.args == ("thread-b",)
+        assert await repo.get_thread_id(connection["id"], "C1") == "thread-a"
+    finally:
+        await close_engine()

backend/tests/test_channel_thread_claim.py

@@ -0,0 +1,203 @@
+"""First-binding claims for IM conversations: two replicas converge on one thread.
+
+When the first message of a conversation reaches two Gateway replicas at the
+same time (or a replica and a peer that has not committed yet), both miss the
+binding lookup and each creates a Gateway thread. ``set_thread_id`` overwrites,
+so whichever replica writes last wins and the other replica's run lands on a
+thread the conversation no longer points to. ``claim_thread_id`` is the
+insert-if-absent primitive the manager uses for that first binding instead: the
+first claim wins and every later claim returns the winner, so the losers can
+adopt it. ``set_thread_id`` keeps its overwrite semantics for ``/new``.
+
+Each case runs on SQLite and, when ``TEST_POSTGRES_URI`` is set, on PostgreSQL in
+a throwaway schema, with two engines over one database standing in for two
+Gateway replicas.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+
+import pytest
+import pytest_asyncio
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+from support.postgres import asyncpg_test_url
+
+from app.channels.store import JsonChannelStore, SqlChannelStore
+from deerflow.persistence.base import Base
+from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelConnectionRow, ChannelConversationRow, ChannelCredentialCipher
+from deerflow.persistence.channel_thread_bindings import ChannelThreadBindingRow
+from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
+
+pytestmark = pytest.mark.asyncio
+
+_TABLES = [ChannelConnectionRow.__table__, ChannelConversationRow.__table__, ChannelThreadBindingRow.__table__]
+# Enough concurrent first writers that a lost race would show on either dialect.
+_WRITERS = 8
+
+
+@dataclass
+class Replicas:
+    dialect: str
+    engines: list[AsyncEngine]
+
+    def connections(self, index: int) -> ChannelConnectionRepository:
+        return ChannelConnectionRepository(async_sessionmaker(self.engines[index], expire_on_commit=False), cipher=ChannelCredentialCipher.from_key("k"))
+
+    def store(self, index: int) -> SqlChannelStore:
+        return SqlChannelStore(async_sessionmaker(self.engines[index], expire_on_commit=False))
+
+
+@pytest_asyncio.fixture(params=["sqlite", "postgres"])
+async def replicas(request, tmp_path) -> AsyncIterator[Replicas]:
+    if request.param == "sqlite":
+        url = f"sqlite+aiosqlite:///{(tmp_path / 'claims.db').as_posix()}"
+        engines = [create_async_engine(url) for _ in range(2)]
+        async with engines[0].begin() as conn:
+            await conn.run_sync(Base.metadata.create_all, tables=_TABLES)
+        try:
+            yield Replicas("sqlite", engines)
+        finally:
+            for engine in engines:
+                await engine.dispose()
+        return
+
+    uri = os.environ.get("TEST_POSTGRES_URI")
+    if not uri:
+        pytest.skip("requires TEST_POSTGRES_URI (channel conversation claims on PostgreSQL)")
+    schema = f"channel_thread_claim_{uuid.uuid4().hex}"
+    engines = [create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) for _ in range(2)]
+    try:
+        async with engines[0].begin() as conn:
+            await conn.execute(sa.text(f'CREATE SCHEMA "{schema}"'))
+            await conn.run_sync(Base.metadata.create_all, tables=_TABLES)
+        yield Replicas("postgresql", engines)
+    finally:
+        async with engines[0].begin() as conn:
+            await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+        for engine in engines:
+            await engine.dispose()
+
+
+async def _connection_binding(replicas: Replicas, conversation: str = "C1") -> dict:
+    connection = await replicas.connections(0).upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U1", workspace_id="T1")
+    return {
+        "connection_id": connection["id"],
+        "owner_user_id": "alice",
+        "provider": "slack",
+        "external_conversation_id": conversation,
+        "external_topic_id": "1700000000.000100",
+    }
+
+
+async def _conversation_rows(replicas: Replicas, conversation: str) -> list[str]:
+    async with replicas.engines[0].connect() as conn:
+        rows = await conn.execute(sa.select(ChannelConversationRow.thread_id).where(ChannelConversationRow.external_conversation_id == conversation))
+        return [row[0] for row in rows]
+
+
+# -- channel_conversations (connection-bound conversations) -------------------
+
+
+async def test_first_claim_binds_and_a_peer_claim_adopts_it(replicas):
+    binding = await _connection_binding(replicas)
+    a, b = replicas.connections(0), replicas.connections(1)
+
+    assert await a.claim_thread_id(**binding, thread_id="thread-a") == "thread-a"
+    assert await b.claim_thread_id(**binding, thread_id="thread-b") == "thread-a"
+
+    assert await b.get_thread_id(binding["connection_id"], "C1", "1700000000.000100") == "thread-a"
+    assert await _conversation_rows(replicas, "C1") == ["thread-a"]
+
+
+async def test_concurrent_first_claims_converge_on_one_thread(replicas):
+    binding = await _connection_binding(replicas)
+    repos = [replicas.connections(index % 2) for index in range(_WRITERS)]
+
+    winners = await asyncio.gather(*(repo.claim_thread_id(**binding, thread_id=f"thread-{index}") for index, repo in enumerate(repos)))
+
+    assert len(set(winners)) == 1
+    assert await _conversation_rows(replicas, "C1") == [winners[0]]
+
+
+async def test_set_thread_id_still_rebinds_a_claimed_conversation(replicas):
+    """``/new`` rebinds the conversation on purpose; a claim must not pin it."""
+    binding = await _connection_binding(replicas)
+    repo = replicas.connections(0)
+    await repo.claim_thread_id(**binding, thread_id="thread-first")
+
+    await repo.set_thread_id(**binding, thread_id="thread-after-new")
+
+    assert await repo.claim_thread_id(**binding, thread_id="thread-late-peer") == "thread-after-new"
+    assert await _conversation_rows(replicas, "C1") == ["thread-after-new"]
+
+
+async def test_claims_are_scoped_by_topic(replicas):
+    binding = await _connection_binding(replicas)
+    repo = replicas.connections(0)
+
+    assert await repo.claim_thread_id(**binding, thread_id="thread-topic") == "thread-topic"
+    assert await repo.claim_thread_id(**{**binding, "external_topic_id": None}, thread_id="thread-base") == "thread-base"
+
+
+async def test_concurrent_first_set_thread_id_leaves_one_row(replicas):
+    """Pins #6852 on real concurrency and both dialects: racing first binds neither fail nor duplicate."""
+    binding = await _connection_binding(replicas, "C-race")
+    repos = [replicas.connections(index % 2) for index in range(_WRITERS)]
+
+    await asyncio.gather(*(repo.set_thread_id(**binding, thread_id=f"thread-{index}") for index, repo in enumerate(repos)))
+
+    rows = await _conversation_rows(replicas, "C-race")
+    assert len(rows) == 1 and rows[0].startswith("thread-")
+
+
+# -- channel_thread_bindings (unbound conversations) ---------------------------
+
+
+async def test_sql_store_first_claim_wins_across_replicas(replicas):
+    a, b = replicas.store(0), replicas.store(1)
+
+    assert await a.claim_thread_id("telegram", "chat-1", "thread-a", topic_id="t1", user_id="u1") == "thread-a"
+    assert await b.claim_thread_id("telegram", "chat-1", "thread-b", topic_id="t1", user_id="u2") == "thread-a"
+
+    assert await b.get_thread_id("telegram", "chat-1", topic_id="t1") == "thread-a"
+    [entry] = await a.list_entries("telegram")
+    assert (entry["thread_id"], entry["user_id"]) == ("thread-a", "u1")
+
+
+async def test_sql_store_concurrent_first_claims_converge(replicas):
+    stores = [replicas.store(index % 2) for index in range(_WRITERS)]
+
+    winners = await asyncio.gather(*(store.claim_thread_id("discord", "chan-1", f"thread-{index}") for index, store in enumerate(stores)))
+
+    assert len(set(winners)) == 1
+    assert [entry["thread_id"] for entry in await stores[0].list_entries("discord")] == [winners[0]]
+
+
+async def test_sql_store_set_thread_id_still_rebinds_after_a_claim(replicas):
+    store = replicas.store(0)
+    await store.claim_thread_id("slack", "C1", "thread-first")
+
+    await store.set_thread_id("slack", "C1", "thread-after-new")
+
+    assert await store.claim_thread_id("slack", "C1", "thread-late-peer") == "thread-after-new"
+
+
+# -- JSON store (database.backend: memory, one process) ------------------------
+
+
+async def test_json_store_claim_keeps_the_first_binding(tmp_path):
+    store = JsonChannelStore(tmp_path / "store.json")
+
+    assert await store.claim_thread_id("telegram", "chat-1", "thread-a", user_id="u1") == "thread-a"
+    assert await store.claim_thread_id("telegram", "chat-1", "thread-b", user_id="u2") == "thread-a"
+    await store.set_thread_id("telegram", "chat-1", "thread-after-new")
+    assert await store.claim_thread_id("telegram", "chat-1", "thread-c") == "thread-after-new"
+
+    # Persisted like any other write: a fresh process reads the claimed binding.
+    assert await JsonChannelStore(tmp_path / "store.json").get_thread_id("telegram", "chat-1") == "thread-after-new"

backend/tests/test_channels.py

@@ -806,7 +806,7 @@ class TestChannelManager:
                     await allow_third_lookup.wait()
                 return None
 
-            async def create_thread(_client, _msg):
+            async def create_first_thread(_client, _msg):
                 nonlocal create_calls, active_later_creates, max_active_later_creates
                 create_calls += 1
                 call_number = create_calls
@@ -825,10 +825,10 @@ class TestChannelManager:
                     await release_later_creates.wait()
                 finally:
                     active_later_creates -= 1
-                return f"thread-{call_number}"
+                return f"thread-{call_number}", True
 
             manager._lookup_thread_id = lookup_thread_id
-            manager._create_thread = create_thread
+            manager._create_first_thread = create_first_thread
             client = MagicMock()
             first_msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="first")
             second_msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="second")
@@ -6833,6 +6833,10 @@ class _BoundIdentityRepo:
             }
         )
 
+    async def claim_thread_id(self, **binding) -> str:
+        await self.set_thread_id(**binding)
+        return binding["thread_id"]
+
 
 class TestChannelManagerBoundIdentityPolicy:
     def test_unbound_auth_enabled_chat_is_rejected_before_thread_or_run_creation(self, monkeypatch):
@@ -13557,12 +13561,12 @@ class TestHandleGoalCommand:
             chats = []
 
             async def _create(client, msg):
-                return "new-thread"
+                return "new-thread", True
 
             async def _handle_chat(msg, **kwargs):
                 chats.append((msg, kwargs))
 
-            monkeypatch.setattr(manager, "_create_thread", _create)
+            monkeypatch.setattr(manager, "_create_first_thread", _create)
             monkeypatch.setattr(manager, "_get_client", lambda: object())
             monkeypatch.setattr(manager, "_handle_chat", _handle_chat)
 

backend/tests/test_discord_channel.py

@@ -913,3 +913,158 @@ class TestDiscordAllowedChannels:
         await channel._on_message(_make_discord_message("hello"))
 
         assert channel._publish_reserved.called is admitted
+
+
+class TestDiscordDedupeClaimBeforeThreadCreation:
+    """Every instance's Discord session receives every MESSAGE_CREATE. The manager
+    drops the duplicate, but only after both instances created a Discord thread
+    for it. The adapter therefore claims the manager's dedupe key first."""
+
+    @staticmethod
+    def _wire(tmp_path, *, dedupe=None):
+        from app.channels.dedupe_store import MemoryInboundDedupeStore
+        from app.channels.manager import ChannelManager
+        from app.channels.store import JsonChannelStore
+
+        bus = MessageBus()
+        dedupe = dedupe or MemoryInboundDedupeStore()
+        ChannelManager(bus=bus, store=JsonChannelStore(tmp_path / "store.json"), inbound_dedupe_store=dedupe)
+        channel = DiscordChannel(bus=bus, config={"bot_token": "token", "thread_mode": True})
+        channel._running = True
+        channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>"))
+        channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {}))
+        channel._main_loop = asyncio.get_running_loop()
+        channel._create_thread = AsyncMock(return_value=SimpleNamespace(id=777))
+        dedupe.try_record = AsyncMock(wraps=dedupe.try_record)
+        channel._start_typing = AsyncMock(return_value=None)
+        channel._add_reaction = AsyncMock(return_value=None)
+        return bus, channel, dedupe
+
+    @staticmethod
+    def _key():
+        from app.channels.manager import ChannelManager
+
+        return ChannelManager._inbound_dedupe_key(InboundMessage(channel_name="discord", chat_id="456", user_id="123", text="", metadata={"guild_id": "321", "channel_id": "456", "message_id": "111"}))
+
+    @pytest.mark.asyncio
+    async def test_message_claimed_by_a_peer_creates_no_thread(self, tmp_path) -> None:
+        bus, channel, dedupe = self._wire(tmp_path)
+        assert await dedupe.try_record(self._key()) is False  # the peer instance won
+
+        await channel._on_message(_make_discord_message("hello"))
+
+        channel._create_thread.assert_not_awaited()
+        assert bus.inbound_queue.qsize() == 0
+        assert bus._inbound_reservations == set()
+
+    @pytest.mark.asyncio
+    async def test_winning_claim_creates_the_thread_and_marks_the_published_message(self, tmp_path) -> None:
+        bus, channel, dedupe = self._wire(tmp_path)
+
+        await channel._on_message(_make_discord_message("hello"))
+        inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5)
+        bus.inbound_task_done()
+
+        channel._create_thread.assert_awaited_once()
+        assert (inbound.chat_id, inbound.topic_id, inbound.dedupe_claimed) == ("456", "777", True)
+        assert self._key() in dedupe._store
+
+        # The same MESSAGE_CREATE on a second session sharing the dedupe store.
+        _, peer, _ = self._wire(tmp_path, dedupe=dedupe)
+        await peer._on_message(_make_discord_message("hello"))
+        peer._create_thread.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_refused_hand_off_gives_the_claim_back(self, tmp_path) -> None:
+        _, channel, dedupe = self._wire(tmp_path)
+        channel._publish_reserved = MagicMock(return_value=None)
+
+        await channel._on_message(_make_discord_message("hello"))
+        for _ in range(20):
+            if self._key() not in dedupe._store:
+                break
+            await asyncio.sleep(0.01)
+
+        channel._create_thread.assert_awaited_once()
+        dedupe.try_record.assert_awaited_once_with(self._key())
+        assert self._key() not in dedupe._store
+
+    @pytest.mark.asyncio
+    async def test_failed_thread_creation_gives_the_claim_back(self, tmp_path) -> None:
+        _, channel, dedupe = self._wire(tmp_path)
+        channel._create_thread = AsyncMock(side_effect=RuntimeError("thread create failed"))
+
+        with pytest.raises(RuntimeError, match="thread create failed"):
+            await channel._on_message(_make_discord_message("hello"))
+        for _ in range(20):
+            if self._key() not in dedupe._store:
+                break
+            await asyncio.sleep(0.01)
+
+        dedupe.try_record.assert_awaited_once_with(self._key())
+        assert self._key() not in dedupe._store
+
+    @pytest.mark.asyncio
+    async def test_failed_commit_gives_the_claim_back(self, tmp_path) -> None:
+        bus, channel, dedupe = self._wire(tmp_path)
+        # The intake closed (shutdown) between the reservation and the commit.
+        channel._commit_reserved_inbound = MagicMock(return_value=False)
+
+        await channel._on_message(_make_discord_message("hello"))
+        for _ in range(20):
+            if self._key() not in dedupe._store:
+                break
+            await asyncio.sleep(0.01)
+
+        dedupe.try_record.assert_awaited_once_with(self._key())
+        assert self._key() not in dedupe._store
+
+    @pytest.mark.asyncio
+    async def test_stop_during_identity_lookup_gives_the_claim_back(self, tmp_path) -> None:
+        """stop() cancels the hand-off while it resolves the connection identity:
+        the message was never committed, so a peer must be able to take it."""
+        _, channel, dedupe = self._wire(tmp_path)
+        lookup_started = asyncio.Event()
+
+        async def lookup_forever(inbound, guild_id=None):
+            lookup_started.set()
+            await asyncio.Event().wait()
+
+        channel._attach_connection_identity = lookup_forever
+        handling = asyncio.create_task(channel._on_message(_make_discord_message("hello")))
+        await asyncio.wait_for(lookup_started.wait(), timeout=5)
+        assert self._key() in dedupe._store
+
+        await channel.stop()
+        await asyncio.wait_for(handling, timeout=5)
+
+        assert self._key() not in dedupe._store
+        peer_bus, peer, _ = self._wire(tmp_path, dedupe=dedupe)
+        await peer._on_message(_make_discord_message("hello"))
+        await asyncio.wait_for(peer_bus.get_inbound(), timeout=5)
+
+    @pytest.mark.asyncio
+    async def test_claim_whose_release_is_refused_during_stop_is_given_back_by_stop(self, tmp_path) -> None:
+        """Thread creation still in flight when stop() closes cross-loop intake:
+        the release cannot be submitted then, so stop() releases it last."""
+        _, channel, dedupe = self._wire(tmp_path)
+        create_started = asyncio.Event()
+        fail_create = asyncio.Event()
+
+        async def create_then_fail(message):
+            create_started.set()
+            await fail_create.wait()
+            raise RuntimeError("thread create failed")
+
+        channel._create_thread = create_then_fail
+        handling = asyncio.create_task(channel._on_message(_make_discord_message("hello")))
+        await asyncio.wait_for(create_started.wait(), timeout=5)
+        await channel._close_and_drain_threadsafe_futures()  # stop() has begun
+        fail_create.set()
+        with pytest.raises(RuntimeError, match="thread create failed"):
+            await handling
+        assert self._key() in dedupe._store
+
+        await channel.stop()
+
+        assert self._key() not in dedupe._store

backend/tests/test_multi_pod_inbound_dedupe.py

@@ -168,3 +168,66 @@ async def test_expired_unreleased_row_is_reclaimed_on_next_redelivery():
         assert row is not None and row[0] is True
     finally:
         await store.release(key)
+
+
+@pytest.mark.asyncio
+async def test_two_managers_share_dedupe_for_a_discord_direct_message():
+    """A Discord DM has no guild id and used to get no dedupe key, so every
+    instance whose session received the MESSAGE_CREATE ran it."""
+    from app.channels.manager import ChannelManager
+    from app.channels.message_bus import InboundMessage, MessageBus
+    from app.channels.store import JsonChannelStore
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.engine import get_engine, get_session_factory
+    from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
+
+    async with get_engine().begin() as conn:
+        await conn.run_sync(Base.metadata.create_all, tables=[WebhookDeliveryRow.__table__], checkfirst=True)
+    shared_store = PostgresInboundDedupeStore(session_factory=get_session_factory())
+    manager_a = ChannelManager(bus=MessageBus(), store=JsonChannelStore(), inbound_dedupe_store=shared_store)
+    manager_b = ChannelManager(bus=MessageBus(), store=JsonChannelStore(), inbound_dedupe_store=shared_store)
+    dm_channel = str(uuid.uuid4().int)[:19]
+
+    def delivery() -> InboundMessage:
+        return InboundMessage(
+            channel_name="discord",
+            chat_id=dm_channel,
+            user_id="42",
+            text="hi",
+            topic_id=dm_channel,
+            metadata={"guild_id": None, "channel_id": dm_channel, "message_id": "1300000000000000001"},
+        )
+
+    try:
+        assert await manager_a._is_duplicate_inbound(delivery()) is False
+        assert await manager_b._is_duplicate_inbound(delivery()) is True
+    finally:
+        await manager_a._release_inbound_dedupe_key(delivery())
+
+
+@pytest.mark.asyncio
+async def test_two_managers_share_dedupe_for_a_reply_in_a_thread_only_one_tracks():
+    """The instance tracking Discord thread T routes a reply in T to its parent
+    channel; the other treats T as an orphan. One message, one run."""
+    from app.channels.manager import ChannelManager
+    from app.channels.message_bus import InboundMessage, MessageBus
+    from app.channels.store import JsonChannelStore
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.engine import get_engine, get_session_factory
+    from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
+
+    async with get_engine().begin() as conn:
+        await conn.run_sync(Base.metadata.create_all, tables=[WebhookDeliveryRow.__table__], checkfirst=True)
+    shared_store = PostgresInboundDedupeStore(session_factory=get_session_factory())
+    manager_a = ChannelManager(bus=MessageBus(), store=JsonChannelStore(), inbound_dedupe_store=shared_store)
+    manager_b = ChannelManager(bus=MessageBus(), store=JsonChannelStore(), inbound_dedupe_store=shared_store)
+    thread_t = str(uuid.uuid4().int)[:19]
+    metadata = {"guild_id": "guild-1", "channel_id": thread_t, "message_id": "1300000000000000002"}
+    tracked = InboundMessage(channel_name="discord", chat_id="parent-1", user_id="42", text="hi", topic_id=thread_t, metadata=dict(metadata))
+    orphaned = InboundMessage(channel_name="discord", chat_id=thread_t, user_id="42", text="hi", topic_id="new-thread", metadata=dict(metadata))
+
+    try:
+        assert await manager_a._is_duplicate_inbound(tracked) is False
+        assert await manager_b._is_duplicate_inbound(orphaned) is True
+    finally:
+        await manager_a._release_inbound_dedupe_key(tracked)

backend/tests/test_thread_run_idempotency.py

@@ -418,6 +418,23 @@ def test_scope_http_run_idempotency_key_ignores_header_default():
     assert thread_runs._scope_http_run_idempotency_key(request, "thread-1", None) is None
 
 
+def test_internal_caller_key_is_scoped_by_owner_not_thread():
+    """The IM channel manager may route a redelivered message to a different
+    thread after ``/new`` rebound the conversation; its key must still meet the
+    run admitted for the first delivery."""
+    from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE
+
+    internal = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "alice"})
+    other_owner = SimpleNamespace(state=internal.state, headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "bob"})
+    browser = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="alice", system_role="user")), headers={})
+
+    scope = thread_runs._scope_http_run_idempotency_key
+    assert scope(internal, "thread-1", "message-1") == scope(internal, "thread-2", "message-1")
+    assert scope(internal, "thread-1", "message-1") != scope(other_owner, "thread-1", "message-1")
+    assert scope(browser, "thread-1", "message-1") != scope(browser, "thread-2", "message-1")
+    assert scope(internal, "thread-1", "message-1") != scope(browser, "thread-1", "message-1")
+
+
 def test_stream_reused_store_only_running_run_returns_409(monkeypatch):
     """A reused running record on a process-local bridge must not hang on an empty stream."""
 
@@ -725,6 +742,74 @@ async def test_start_run_rejects_reused_key_with_different_input(_stub_app_confi
     assert "different request" in str(excinfo.value.detail)
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(("stable_message_id", "reused"), [(True, True), (False, False)], ids=["keyed-message-id", "random-message-id"])
+async def test_channel_redelivery_on_another_instance_reuses_the_admitted_run(_stub_app_config, stable_message_id, reused):
+    """The IM manager's keyed request, as a second Gateway instance sees it.
+
+    The human message id is part of the compared input, so only a message id
+    derived from the same key lets a redelivery reuse the admitted run; a
+    fresh uuid per delivery is a different request (409).
+    """
+    from app.channels.manager import ChannelManager, _human_input_message
+    from app.channels.message_bus import InboundMessage
+    from app.gateway.services import start_run
+
+    msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="hi", metadata={"event_id": "evt-1", "team_id": "T1"})
+    key = f"http-run:{ChannelManager._run_idempotency_key(msg)}"
+
+    def body() -> RunCreateRequest:
+        message_id = ChannelManager._human_message_id(msg) if stable_message_id else None
+        return RunCreateRequest(input={"messages": [_human_input_message("hi", message_id=message_id)]}, on_disconnect="continue")
+
+    async def fake_run_agent(*args, **kwargs):
+        del args, kwargs
+        await asyncio.Event().wait()
+
+    store = MemoryRunStore()
+    first_instance, second_instance = RunManager(store=store, worker_id="worker-a"), RunManager(store=store, worker_id="worker-b")
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+    ):
+        first = await start_run(body(), "thread-1", _make_start_run_request(first_instance), idempotency_key=key)
+        try:
+            if reused:
+                second = await start_run(body(), "thread-1", _make_start_run_request(second_instance), idempotency_key=key)
+                assert (second.run_id, second.idempotency_reused) == (first.run_id, True)
+            else:
+                with pytest.raises(HTTPException) as excinfo:
+                    await start_run(body(), "thread-1", _make_start_run_request(second_instance), idempotency_key=key)
+                assert excinfo.value.status_code == 409
+                assert "different request" in str(excinfo.value.detail)
+        finally:
+            first.task.cancel()
+            await asyncio.gather(first.task, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_key_already_admitted_on_another_thread_is_a_409_not_a_second_run(_stub_app_config):
+    """A redelivery routed to the conversation's new thread after ``/new``."""
+    from app.gateway.services import start_run
+
+    input_payload = {"messages": [{"role": "user", "content": "hello"}]}
+    store = MemoryRunStore()
+    first_instance, second_instance = RunManager(store=store, worker_id="worker-a"), RunManager(store=store, worker_id="worker-b")
+    await first_instance.create_or_reject("thread-old", user_id=None, idempotency_key="internal-run:same", kwargs={"input": input_payload, "config": None})
+
+    for run_manager in (first_instance, second_instance):  # the local record, then the durable row
+        with (
+            patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+            patch("app.gateway.services.run_agent", side_effect=AssertionError("worker must not attach")),
+            pytest.raises(HTTPException) as excinfo,
+        ):
+            await start_run(RunCreateRequest(input=input_payload), "thread-new", _make_start_run_request(run_manager), idempotency_key="internal-run:same")
+
+        assert excinfo.value.status_code == 409
+        assert "Idempotency-Key already used" in str(excinfo.value.detail)
+    assert await store.list_by_thread("thread-new") == []
+
+
 def test_resume_idempotency_fingerprint_is_typed_and_order_independent():
     from app.gateway.services import _run_idempotency_request