Repository navigation
fix(wecom): release inbound dedup claims when the handler fails, without re-running completed turns - #62861
fix(wecom): release inbound dedup claims when the handler fails, without re-running completed turns#62861irresi wants to merge 3 commits into
Conversation
…liveries are not dropped MessageDeduplicator.is_duplicate() marks a message seen at check time, before processing finishes. WeCom acks a callback only via the response frame sent after processing and redelivers the same msgid on silence (NousResearch#47573), so a failure after the mark left a stale entry that swallowed every redelivery for the full 300s TTL — silent message loss. Add MessageDeduplicator.remove() (matching the method NousResearch#36758 introduces for the weixin success path) and release the claim when processing raises, both in the synchronous _on_message path and in the background text-batch flush, covering every chunk merged into a failed batch. Fixes NousResearch#62860 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
teknium1
left a comment
There was a problem hiding this comment.
Thanks for tracing the stale dedup claim through the WeCom callback path. The pre-dispatch failure case is real on current main: plugins/platforms/wecom/adapter.py:499-500 marks the ID before parsing and dispatch.
Problems
- The new
tryblocks do not cover the normal agent-processing failure path.gateway/platforms/base.py:4779makeshandle_message()schedule_process_message_background()and return; the real_message_handlerruns in that background task atgateway/platforms/base.py:4855-4859, whose failures are handled internally atgateway/platforms/base.py:5203-5231. Consequently, the new catches do not release a WeCom claim when that normal background path fails. The added tests mockhandle_message, so they do not exercise this lifecycle.
Suggested changes
- Carry the WeCom claim to an appropriate processing-completion failure/cancellation path and add a regression test using the real background dispatch lifecycle.
Automated hermes-sweeper review.
| return | ||
| try: | ||
| await self._process_message(payload, body, msg_id) | ||
| except BaseException: |
There was a problem hiding this comment.
handle_message() normally returns after scheduling _process_message_background() (gateway/platforms/base.py:4779); actual handler failures are caught inside that background task (gateway/platforms/base.py:5203-5231). This catch therefore misses the normal agent-processing failure that can leave the claim stale. Please wire the release to the relevant completion/failure path and test the real lifecycle rather than a mocked handle_message exception.
…serve dedup claim ids across event merges ProcessingOutcome.FAILURE conflates two very different situations: the handler raised before completing (the event was never processed), and the handler completed but delivery went wrong (processing_ok is a delivery heuristic — a failed reply send, or an image-only reply that never records delivery, both classify FAILURE). Adapters that release per-message claims (inbound dedup) on FAILURE would re-run completed turns — duplicate tool side effects — on platform redelivery. Add an additive on_handler_failure lifecycle hook that fires only when the message handler raised or was unexpectedly cancelled before returning, gated by a handler_completed flag in _process_message_background. The busy-session inline dispatch paths (bypass commands, clarify text-intercept) swallow handler exceptions by design; they now report through the same hook, gated the same way, so a failed /approve or clarify answer no longer silently keeps its claim. Existing on_processing_complete overrides are unaffected. Also preserve metadata["dedup_msg_ids"] when base merges or replaces buffered events (merge_pending_message_event's three merge branches and its replace fallthrough, and _queue_text_debounce's merge): dropping the ids left the dropped event claimed for the dedup TTL after a failure, silently discarding the platform's redelivery. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…patch from supersede-cancel Follow-ups to the claim-release fix, addressing review: - Background pipeline failures never reach the intake try blocks (handle_message returns after scheduling the background task). Adopt the new on_handler_failure hook: claims are released exactly when the handler did not complete. Completed turns whose reply delivery failed keep their claims — re-processing them would duplicate side effects — as do expected cancellations (/stop, session reset). - Every dispatched event carries its claim ids in MessageEvent.metadata["dedup_msg_ids"] (seeded in _process_message, enforced again at the batch-registry boundary so raw test events cannot KeyError the merge branch). - The batch flush now shields its dispatch: a supersede-cancel from a newer chunk can land before gateway acceptance (losing the popped batch) or after it (double-processing on redelivery if released) — letting the in-flight dispatch run to completion removes both cases. A shielded dispatch that still fails synchronously releases its claims via a done-callback. - Tests drive the real dispatch lifecycle: background handler failure releases, delivery-failure-after-completion does not, busy-session inline bypass failure releases, supersede neither loses nor duplicates, and a failed merged flush releases the head and appended chunks independently. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…sion + best-effort finalize) Two failing repro tests for the '同一回复发两条' bug, authored to drive the fix (they assert the DESIRED post-fix state, so they FAIL today): 1. test_slow_ack_beyond_gateway_join_causes_double_send — timeout inversion: finalize ack (0.3s) slower than the gateway join window (0.1s) → gateway cancels the consumer mid-finalize → final_content_delivered stays False while the finalize frame was already rendered → duplicate. Uses the REAL WeComAdapter + REAL GatewayStreamConsumer.run() lifecycle (only WS byte writer + ack timing controlled), avoiding the mock-handle_message anti-pattern that got PR NousResearch#62861 bounced. 2. test_best_effort_finalize_renders_but_flag_unset_double_send — the stream_consumer.py:2064-2094 'DO NOT mark' path: native fails mid-stream, best-effort finalize frame renders, then fallback send delivers the same content again. Control: test_fast_ack_within_join_window_single_send passes today and after the fix (regression guard). The gateway suppression predicate is mirrored (not called) because run.py's _stream_confirmed_final_delivery is a nested fn inside _run_agent_inner; the mirror reads the REAL consumer attributes run.py consults. Authored by the review counterpart; the fix will be authored separately then cross-reviewed.
What this fixes
MessageDeduplicator.is_duplicate()marks a message seen at check time, before processing finishes. WeCom acks a callback only via the reply frame and redelivers the same msgid on silence (#47573), so any failure between the mark and a completed run left a stale claim that swallowed every redelivery for the dedup TTL (300s) — silent message loss with only a DEBUG trace.Fixes #62860.
How the claim lifecycle works now
A claim is released exactly when the message was NOT processed, and kept when it was:
_on_message): release and re-raise._flush_text_batch): the dispatch is wrapped inasyncio.shield— a supersede-cancel from a newer chunk can land before gateway acceptance (dropping the popped batch) or after it (double-processing on redelivery if released); letting the in-flight dispatch complete removes both cases. A shielded dispatch that still fails synchronously releases its claims via a done-callback.handle_messagereturns after scheduling_process_message_background): a new additiveon_handler_failurelifecycle hook in base fires only when the message handler raised or was unexpectedly cancelled before completing, gated by ahandler_completedflag.ProcessingOutcome.FAILUREalone is a delivery heuristic (a failed reply send, or an image-only reply that never records delivery, both classify FAILURE after a fully completed run) — releasing on it would re-run completed turns and duplicate tool side effects. The busy-session inline dispatch paths (bypass commands, clarify text-intercept) swallow handler exceptions by design; they now report through the same hook, so a failed/approveor clarify answer no longer keeps its claim while the agent stays blocked./stop, session reset), and policy-blocked messages.Claim ids ride
MessageEvent.metadata["dedup_msg_ids"]; base's event merges (merge_pending_message_event,_queue_text_debounce) now preserve their union, since a dropped id stayed claimed forever after a failure. The key is unprefixed because base owns its merge semantics (_merge_dedup_claim_ids) — it is shared claim infrastructure, not adapter-private state.Scope boundary
WeCom-only adoption by design. DingTalk's stream handler acks
STATUS_OKbefore processing, so it never redelivers on later failure — a release there would be dead code. Other adapters sharing the helper (google_chat is at-least-once Pub/Sub, slack redelivers on Socket-Mode reconnects per its own comments) have the same mark-before-success shape and can adoptremove()+on_handler_failurein follow-ups; the base hook and merge preservation shipped here are what make that adoption one method per adapter.Tests
tests/gateway/test_wecom_dedup_redelivery.pyexercises the real dispatch lifecycle (set_message_handler+ background tasks, nohandle_messagemock where the contract depends on base):/statuson an occupied session) → releasedpytestacross the wecom/batching/dedup/base-session neighborhood: 157 passed, 1 pre-existing environmental failure (TestWeComZombieSessionFix, reproduces identically without these commits — missing optional aiohttp in the minimal dev env).ruffclean;tydiagnostics net −1 vs the previous head (macOS 15, Python 3.12; behavior is platform-independent).