diff --git a/backend/app/gateway/github/dispatcher.py b/backend/app/gateway/github/dispatcher.py index 15cf45b1a..f8d886a00 100644 --- a/backend/app/gateway/github/dispatcher.py +++ b/backend/app/gateway/github/dispatcher.py @@ -398,6 +398,28 @@ async def fanout_event( # (The store accepts a pre-known thread id; manager will fall # through to _create_thread() the very first time.) topic_id = f"{number}:{agent.name}" + # Inbound dedupe identity for ChannelManager._is_duplicate_inbound, + # 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 + # 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 + # same delivery — including two different users who each bind an + # agent of the same name to the same repo+event — keep distinct ids + # and both still fire. ``ChannelManager._inbound_dedupe_key`` indexes + # on (channel, workspace_id, chat_id, message_id); workspace_id and + # chat_id are both the repo here, so match.user_id is the only thing + # that can separate two users in that key — see + # test_dedupe_identity_distinguishes_same_agent_name_across_users. + # Left None when the header is absent, so the manager fails open (no + # dedupe) exactly as before rather than collapsing distinct deliveries. + dedupe_message_id = f"{delivery_id}:{match.user_id}:{agent.name}" if delivery_id else None msg = InboundMessage( channel_name="github", chat_id=repo, @@ -408,7 +430,15 @@ async def fanout_event( # owner_user_id drives which user bucket the run executes in # (custom agent lookup, sandbox, memory). owner_user_id=match.user_id, + # Tenant key the manager's dedupe requires (it fails closed without + # one, to avoid collapsing two workspaces). The repo is globally + # unique and always present — mirrors Telegram/WeChat keying the + # workspace on the chat id. + workspace_id=repo, metadata={ + # Stable inbound-dedupe id keyed by the manager — see + # ``dedupe_message_id`` above. + "message_id": dedupe_message_id, # Routes to the right custom agent inside the manager: # _resolve_run_params() pulls this and writes it into # run_context["agent_name"]. diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 1c5edd4b2..04ce43160 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -1039,6 +1039,45 @@ class TestChannelManager: ) assert ChannelManager._inbound_dedupe_key(without_workspace) is None + def test_github_redelivery_is_deduped_like_other_channels(self, tmp_path): + """A redelivered GitHub webhook must dispatch the agent only once. + + PR #3584 added inbound dedupe for the IM channels; the GitHub channel + 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 + 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. + """ + from app.channels.manager import ChannelManager + + 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. + return InboundMessage( + channel_name="github", + chat_id="zhfeng/llm-gateway", + user_id="alice", + text="@bot please review", + topic_id=f"7:{agent}", + workspace_id="zhfeng/llm-gateway", + metadata={"message_id": f"{delivery}:{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") + + # First delivery fires; an identical redelivery of the same GUID is dropped. + assert manager._is_duplicate_inbound(_gh("d1")) is False + assert manager._is_duplicate_inbound(_gh("d1")) is True + # A genuinely new delivery still fires. + 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 + 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).""" from app.channels.manager import ChannelManager diff --git a/backend/tests/test_github_dispatcher.py b/backend/tests/test_github_dispatcher.py index 07d02fa0e..fc50f83f0 100644 --- a/backend/tests/test_github_dispatcher.py +++ b/backend/tests/test_github_dispatcher.py @@ -1036,6 +1036,165 @@ async def test_coder_and_reviewer_on_same_pr_get_distinct_threads(base_dir: Path assert reviewer.metadata["preferred_thread_id"] == reviewer.metadata["github"]["thread_id"] +# --------------------------------------------------------------------------- +# Inbound dedupe identity — redelivery / retry-on-timeout protection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_delivery_id_populates_inbound_dedupe_identity(base_dir: Path) -> None: + """Each fanned-out message carries the identity ChannelManager dedupes on. + + 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 + user + agent) as the message id and the repo as the workspace id, + exactly where ``ChannelManager._inbound_dedupe_key`` looks. + """ + bus = MessageBus() + _write_agent( + base_dir, + "default", + "reviewer", + {"name": "reviewer", "github": {"installation_id": 1234, "bindings": [{"repo": "zhfeng/llm-gateway", "triggers": {"pull_request": {"actions": ["opened"]}}}]}}, + ) + payload = { + "action": "opened", + "pull_request": {"number": 7, "title": "x", "user": {"login": "zhfeng"}, "body": ""}, + "repository": {"full_name": "zhfeng/llm-gateway"}, + "sender": {"login": "zhfeng"}, + } + await fanout_event(bus, "pull_request", "del-abc", payload) + (msg,) = await _drain(bus) + + # Workspace id: the manager fails closed without one; repo is stable + unique. + assert msg.workspace_id == "zhfeng/llm-gateway" + # Stable per-(delivery, user, agent) message id the manager keys dedupe + # on, read from the top level of metadata (not the nested ``github`` + # block). ``user_id`` here is "default" — the owning user this agent + # config lives under. + assert msg.metadata["message_id"] == "del-abc:default:reviewer" + + +@pytest.mark.asyncio +async def test_dedupe_identity_stable_across_redelivery_and_distinct_per_agent(base_dir: Path) -> None: + """Redelivery reproduces the same ids (deduped); agents/deliveries differ. + + A single delivery fans out to N agents, so the dedupe id is scoped to + (delivery, agent): replaying the same ``X-GitHub-Delivery`` yields identical + ids per agent (the manager drops the replay) while two agents on one + delivery — or a genuinely new delivery — keep distinct ids and still fire. + """ + bus = MessageBus() + for n in ("coder", "reviewer"): + _write_agent(base_dir, "default", n, {"name": n, "github": {"bindings": [{"repo": "a/b", "triggers": {"pull_request": {"actions": ["opened"]}}}]}}) + payload = { + "action": "opened", + "pull_request": {"number": 1, "title": "x", "user": {"login": "u"}, "body": ""}, + "repository": {"full_name": "a/b"}, + "sender": {"login": "u"}, + } + + async def _ids(delivery: str) -> dict[str, str]: + await fanout_event(bus, "pull_request", delivery, payload) + return {m.metadata["agent_name"]: m.metadata["message_id"] for m in await _drain(bus)} + + first = await _ids("del-1") + redelivery = await _ids("del-1") + second = await _ids("del-2") + + # Per-agent within one delivery: distinct, so neither agent is dropped. + assert first["coder"] != first["reviewer"] + # Redelivery of the same GUID: identical per agent (manager will dedupe). + assert redelivery == first + # New delivery: every id changes, so both agents fire again. + assert second["coder"] != first["coder"] + assert second["reviewer"] != first["reviewer"] + + +@pytest.mark.asyncio +async def test_dedupe_identity_distinguishes_same_agent_name_across_users(base_dir: Path) -> None: + """Two different users' same-named agents must not collide (willem-bd, PR #4104). + + ``ChannelManager._inbound_dedupe_key`` indexes on + ``(channel, workspace_id, chat_id, message_id)``. For GitHub both + ``workspace_id`` and ``chat_id`` are the repo, so ``owner_user_id`` was + never represented anywhere in the key. Before folding ``match.user_id`` + into the per-message id, two users each binding an agent named + ``reviewer`` to the same repo+event produced the *identical* id + ``f"{delivery_id}:reviewer"`` for both fan-out messages, so + ``ChannelManager._is_duplicate_inbound`` silently dropped the second + user's run as a false-positive duplicate of the first — even though + GitHub only delivered the webhook once and both users' agents matched. + """ + bus = MessageBus() + for user_id in ("alice", "bob"): + _write_agent( + base_dir, + user_id, + "reviewer", + { + "name": "reviewer", + "github": { + "bindings": [ + {"repo": "a/b", "triggers": {"pull_request": {"actions": ["opened"]}}}, + ], + }, + }, + ) + payload = { + "action": "opened", + "pull_request": {"number": 1, "title": "x", "user": {"login": "u"}, "body": ""}, + "repository": {"full_name": "a/b"}, + "sender": {"login": "u"}, + } + result = await fanout_event(bus, "pull_request", "del-cross-user", payload) + assert result["fired_agents"] == ["reviewer", "reviewer"] + + messages = await _drain(bus) + assert len(messages) == 2 + by_owner = {m.owner_user_id: m for m in messages} + assert set(by_owner) == {"alice", "bob"} + + # The dedupe id must differ even though the agent name is identical — + # otherwise the two users' fan-out messages are indistinguishable to + # the manager's dedupe. + assert by_owner["alice"].metadata["message_id"] != by_owner["bob"].metadata["message_id"] + + # Prove the actual observable consequence, not just that the raw ids + # differ: neither user's message is treated as a duplicate of the + # other inside the same ChannelManager dedupe window. + from app.channels.manager import ChannelManager + from app.channels.store import ChannelStore + + manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=base_dir / "dedupe-store.json")) + assert manager._is_duplicate_inbound(by_owner["alice"]) is False + assert manager._is_duplicate_inbound(by_owner["bob"]) is False + + +@pytest.mark.asyncio +async def test_missing_delivery_header_leaves_dedupe_open(base_dir: Path) -> None: + """Absent ``X-GitHub-Delivery`` fails open (no dedupe id), never collapses. + + ``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. + """ + bus = MessageBus() + _write_agent(base_dir, "default", "reviewer", {"name": "reviewer", "github": {"bindings": [{"repo": "a/b", "triggers": {"pull_request": {"actions": ["opened"]}}}]}}) + payload = { + "action": "opened", + "pull_request": {"number": 1, "title": "x", "user": {"login": "u"}, "body": ""}, + "repository": {"full_name": "a/b"}, + "sender": {"login": "u"}, + } + await fanout_event(bus, "pull_request", "", payload) + (msg,) = await _drain(bus) + assert msg.metadata["message_id"] is None + + # --------------------------------------------------------------------------- # Redundant review-comment fan-out suppression (issue #4121, narrower slice) #