mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-21 10:15:47 +00:00
fix(channels): document GitHub dedupe TTL, tighten redelivery tests (#4274)
* fix(channels): document GitHub dedupe TTL, tighten redelivery tests Follow-up to PR #4104's review, whose non-blocking notes were left unaddressed at merge time: - Document the previously-undocumented inbound-dedupe TTL (INBOUND_DEDUPE_TTL_SECONDS, 10 min, in-memory only) directly above the constant in ChannelManager: what it is, why the window, and what happens at the boundary (a manual "Redeliver" clicked after the TTL has elapsed, or any redelivery after a Gateway restart, is not deduped, since _recent_inbound_events is never persisted to ChannelStore). Cross-reference it from the GitHub dispatcher's fan-out comment, which previously implied unconditional redelivery coverage. - test_channels.py::test_github_redelivery_is_deduped_like_other_channels: the _gh helper stamped a 2-part delivery:agent id while production stamps a 3-part delivery:user:agent id. Align the helper with the real shape and add a cross-user assertion the old 2-part shape could not even express. - test_github_dispatcher.py::test_missing_delivery_header_leaves_dedupe_open: previously only asserted the dispatcher-layer id was None. Add the manager-level _is_duplicate_inbound assertion across two header-less deliveries so a future regression that treats a missing id as a real key is caught here too, not only at the dispatcher layer. All four affected test files (test_github_dispatcher.py, test_channels.py, test_github_channel.py, test_github_registry.py) pass; ruff check and ruff format --check are clean. Verified the two tightened tests actually discriminate by temporarily reintroducing each bug they target (2-part id shape; missing-id treated as a real key) and confirming the new assertions fail, then reverting. * fix(channels): correct GitHub redelivery claim in dedupe comments fancyboi999's review on this PR flagged that the previous commit's new comments (manager.py, dispatcher.py) justified the 10-minute dedupe TTL by claiming GitHub automatically retries a failed delivery after its 10-second timeout. GitHub's own documentation says the opposite: failed deliveries (non-2xx, timeout, connection error) are recorded as failed and never automatically redelivered, for both repository and GitHub App webhooks alike. Every redelivery is explicit: the "Redeliver" button, the REST API, or an operator's own recovery script. - manager.py: rework the INBOUND_DEDUPE_TTL_SECONDS comment to state GitHub's actual behavior (no auto-retry), cite the authoritative doc (docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries), and justify the 10-minute window as a bound on explicit near-term replays rather than an automatic-retry window. No longer asserts unverified retry behavior for the other channels either. - dispatcher.py: same correction for the dedupe_message_id comment, which previously implied "retried after a timeout" as a distinct, automatic case alongside the manual "Redeliver" button. - test_channels.py / test_github_dispatcher.py: reworded the two docstrings and one section comment that repeated the same "retry-on-timeout" framing, so the corrected mental model doesn't sit next to contradicting prose nearby. No assertions changed. Verified directly against GitHub's current documentation (fetched, not paraphrased from memory): "Handling failed webhook deliveries" states plainly that GitHub does not auto-redeliver; "Automatically redelivering failed deliveries for a GitHub App webhook" / "...for a repository webhook" are both guides for a user-written recovery script (GitHub's own example runs on a 6-hour cron), not a native GitHub feature. No difference in this behavior between GitHub Apps and repository webhooks. All four affected test files pass; the two `test_bot_login_whitespace_only` / `test_trigger_mention_login_whitespace_only` failures in test_github_dispatcher.py are pre-existing on this branch's unmodified HEAD (confirmed against a clean checkout of the same commit) and are unrelated to this change. ruff check and ruff format --check are clean on all four touched files.
This commit is contained in:
Daoyuan Li
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)."""
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user