diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index b6405bbb8..212da3ad8 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -72,6 +72,36 @@ MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages") THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again." BOUND_IDENTITY_REQUIRED_MESSAGE = "Connect this channel from DeerFlow Settings, complete the in-channel connect step, then send your message again." BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is temporarily unavailable. Please try again later or contact the DeerFlow operator." +# Inbound-redelivery dedup window. ``_recent_inbound_events`` (below) is a +# process-local, in-memory-only OrderedDict — it is never persisted to +# ``ChannelStore`` — so a recorded key survives only for this TTL (or until +# evicted by the entry cap below) and is gone entirely across a Gateway +# restart. 10 minutes is a deliberately bounded window: long enough to +# absorb a near-term redelivery of the same event — whether a provider's +# own automatic retry (where the provider implements one) or an operator +# explicitly triggering a resend — without keeping a growing in-memory +# ledger. +# +# For GitHub specifically: GitHub does NOT automatically retry or redeliver +# a failed delivery (non-2xx response, timeout, or connection error) — it +# is simply recorded as failed. See GitHub's own documentation: +# https://docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries. +# Every redelivery of the same ``X-GitHub-Delivery`` GUID is therefore an +# explicit action — the repo/App "Redeliver" button, the REST API, or an +# operator's own scheduled recovery script polling the failed-deliveries +# endpoint (the pattern GitHub's own docs recommend) — never an automatic +# GitHub-side retry. This TTL exists to absorb exactly those explicit +# near-term replays. +# +# 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 +# ``ChannelStore`` instead, which is not implemented here. INBOUND_DEDUPE_TTL_SECONDS = 10 * 60 INBOUND_DEDUPE_MAX_ENTRIES = 4096 # Only server-stable provider message ids: client-generated ids (client_msg_id, diff --git a/backend/app/gateway/github/dispatcher.py b/backend/app/gateway/github/dispatcher.py index f8d886a00..0c0e8233a 100644 --- a/backend/app/gateway/github/dispatcher.py +++ b/backend/app/gateway/github/dispatcher.py @@ -402,11 +402,18 @@ async def fanout_event( # mirroring the stable per-message id the other channels stamp (Slack # `ts`, Telegram `message_id`, WeChat/WeCom `message_id`, …) that the # dedupe added in PR #3584 keys on. ``X-GitHub-Delivery`` is GitHub's - # globally-unique per-delivery GUID; it is reused verbatim when a - # delivery is retried after a timeout or replayed via the repo/App - # "Redeliver" button, so keying on it lets the manager absorb those - # replays instead of re-running the agent (and its real side effects, - # e.g. a duplicate PR comment). One delivery fans out to N agents + # globally-unique per-delivery GUID; it is reused verbatim on any + # redelivery of the same event. GitHub does not automatically retry + # a failed delivery — every redelivery is explicit: the repo/App + # "Redeliver" button, the REST API, or an operator's own recovery + # script (see + # https://docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries). + # Keying on the GUID lets the manager absorb those replays (within + # the in-memory dedupe TTL — see ``INBOUND_DEDUPE_TTL_SECONDS`` in + # ``app.channels.manager``; a late manual "Redeliver" or one after a + # Gateway restart is not deduped) instead of re-running the agent + # (and its real side effects, e.g. a duplicate PR comment). One + # delivery fans out to N agents # across potentially N different owning users, so the per-message id # is scoped to (delivery, user, agent): an identical redelivery # reproduces the same triples (deduped) while two agents matching the diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index bf883351d..94fc5a526 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -1418,9 +1418,11 @@ class TestChannelManager: PR #3584 added inbound dedupe for the IM channels; the GitHub channel added in PR #3754 never stamped the ``message_id`` / workspace the - dedupe keys on, so GitHub's native "Redeliver" button or a - retry-on-timeout re-ran the agent with real side effects (e.g. a - duplicate PR comment). The dispatcher now stamps the X-GitHub-Delivery + dedupe keys on, so a redelivered GitHub webhook (the native + "Redeliver" button, the REST API, or an operator's own recovery + script — GitHub does not auto-retry a failed delivery) re-ran the + agent with real side effects (e.g. a duplicate PR comment). The + dispatcher now stamps the X-GitHub-Delivery GUID (scoped per agent) plus the repo, so the same manager dedupe absorbs the replay — while a second agent bound to the same delivery, and a genuinely new delivery, still fire. @@ -1429,20 +1431,25 @@ class TestChannelManager: manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=tmp_path / "store.json")) - def _gh(delivery: str, agent: str = "reviewer") -> InboundMessage: - # Shaped exactly as app.gateway.github.dispatcher.fanout_event emits. + def _gh(delivery: str, agent: str = "reviewer", owner_user_id: str = "alice") -> InboundMessage: + # Shaped exactly as app.gateway.github.dispatcher.fanout_event + # emits: a 3-part (delivery, owner_user_id, agent) message_id — + # ``dedupe_message_id = f"{delivery_id}:{match.user_id}:{agent.name}"`` + # — plus the matching ``owner_user_id`` field fanout_event sets + # from ``match.user_id``. return InboundMessage( channel_name="github", chat_id="zhfeng/llm-gateway", user_id="alice", + owner_user_id=owner_user_id, text="@bot please review", topic_id=f"7:{agent}", workspace_id="zhfeng/llm-gateway", - metadata={"message_id": f"{delivery}:{agent}", "agent_name": agent}, + metadata={"message_id": f"{delivery}:{owner_user_id}:{agent}", "agent_name": agent}, ) # The dedupe key matches the other channels' 4-tuple shape. - assert ChannelManager._inbound_dedupe_key(_gh("d1")) == ("github", "zhfeng/llm-gateway", "zhfeng/llm-gateway", "d1:reviewer") + assert ChannelManager._inbound_dedupe_key(_gh("d1")) == ("github", "zhfeng/llm-gateway", "zhfeng/llm-gateway", "d1:alice:reviewer") # First delivery fires; an identical redelivery of the same GUID is dropped. assert manager._is_duplicate_inbound(_gh("d1")) is False @@ -1451,6 +1458,12 @@ class TestChannelManager: assert manager._is_duplicate_inbound(_gh("d2")) is False # A second agent fanned out from the SAME delivery is not cross-deduped. assert manager._is_duplicate_inbound(_gh("d1", agent="coder")) is False + # A second user's SAME-named agent on the SAME delivery is not + # cross-deduped either. A helper still stamping the old 2-part + # (delivery, agent) id could not even express this case — it would + # collide with the very first assertion's "d1"+"reviewer" key and + # silently drop this user's run (willem-bd, PR #4104 review). + assert manager._is_duplicate_inbound(_gh("d1", owner_user_id="bob")) is False def test_dispatch_loop_releases_dedupe_key_when_handling_fails(self, tmp_path): """A transient handling failure must not black-hole a provider redelivery (ShenAC #1).""" diff --git a/backend/tests/test_github_dispatcher.py b/backend/tests/test_github_dispatcher.py index fc50f83f0..98d3c7692 100644 --- a/backend/tests/test_github_dispatcher.py +++ b/backend/tests/test_github_dispatcher.py @@ -1037,7 +1037,7 @@ async def test_coder_and_reviewer_on_same_pr_get_distinct_threads(base_dir: Path # --------------------------------------------------------------------------- -# Inbound dedupe identity — redelivery / retry-on-timeout protection +# Inbound dedupe identity — redelivery / replay protection # --------------------------------------------------------------------------- @@ -1048,8 +1048,9 @@ async def test_delivery_id_populates_inbound_dedupe_identity(base_dir: Path) -> The inbound dedupe added for the IM channels in PR #3584 keys on a top-level ``metadata["message_id"]`` plus a workspace id. The GitHub channel added later (PR #3754) never populated either, so a redelivered - webhook (native "Redeliver" button / retry-on-timeout) re-ran the agent. - Fan-out now stamps the ``X-GitHub-Delivery`` GUID (scoped per owning + webhook (native "Redeliver" button, REST API, or an operator's own + recovery script — GitHub does not auto-retry a failed delivery) re-ran + the agent. Fan-out now stamps the ``X-GitHub-Delivery`` GUID (scoped per owning user + agent) as the message id and the repo as the workspace id, exactly where ``ChannelManager._inbound_dedupe_key`` looks. """ @@ -1181,7 +1182,19 @@ async def test_missing_delivery_header_leaves_dedupe_open(base_dir: Path) -> Non ``delivery_id`` originates from an optional header and can be empty. An empty value must not become a constant key that would silently drop distinct deliveries — it yields no dedupe id, i.e. the pre-fix behavior. + + This asserts the actual manager-level consequence, not just the raw + dispatcher-layer id: today ``_inbound_dedupe_key`` returns ``None`` for a + falsy ``message_id``, so ``_is_duplicate_inbound`` returns ``False`` and + never records a key. A future change that let a missing id fall through + as a real (constant) key would silently collapse every header-less + delivery into "the same" message; asserting on two separate header-less + deliveries pins that neither is ever treated as a duplicate of the other + (willem-bd, PR #4104 review). """ + from app.channels.manager import ChannelManager + from app.channels.store import ChannelStore + bus = MessageBus() _write_agent(base_dir, "default", "reviewer", {"name": "reviewer", "github": {"bindings": [{"repo": "a/b", "triggers": {"pull_request": {"actions": ["opened"]}}}]}}) payload = { @@ -1190,9 +1203,17 @@ async def test_missing_delivery_header_leaves_dedupe_open(base_dir: Path) -> Non "repository": {"full_name": "a/b"}, "sender": {"login": "u"}, } + manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=base_dir / "dedupe-store.json")) + await fanout_event(bus, "pull_request", "", payload) - (msg,) = await _drain(bus) - assert msg.metadata["message_id"] is None + (first,) = await _drain(bus) + assert first.metadata["message_id"] is None + assert manager._is_duplicate_inbound(first) is False + + await fanout_event(bus, "pull_request", "", payload) + (second,) = await _drain(bus) + assert second.metadata["message_id"] is None + assert manager._is_duplicate_inbound(second) is False # ---------------------------------------------------------------------------