From 4436eacebff6b5fc77f80b1187fe58c447c801bd Mon Sep 17 00:00:00 2001 From: rhylryan21 Date: Sat, 11 Jul 2026 16:06:28 +0100 Subject: [PATCH] fix(gateway): kanban notifier delivery reliability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - honor SendResult(success=False) instead of discarding it, so an adapter that REPORTS (not raises) a soft send failure — e.g. the Telegram adapter's "Not connected" mid-reconnect — no longer advances the cursor past an undelivered event and silently loses the notification. Addresses the notifier half of #31901. - add block_loop_detected to the notifier's TERMINAL_KINDS so a task routed to triage for a human decision (re-blocked past the recurrence limit) actually pings its subscribers instead of stalling silently. - raise MAX_SEND_FAILURES 3 -> 12 (~60s at the 5s tick) so a transient Telegram/API outage does not permanently unsubscribe a live channel now that reported soft-failures also reach this counter. - route active-profile-stamped subscriptions via the primary adapter on a single-profile gateway (self.adapters[platform] when the stamped notifier_profile equals the active profile). Related to #56802. Adds test_kanban_notifier_rewinds_claim_on_reported_send_failure asserting a reported send failure leaves the event unseen (rewound) rather than consumed. Co-Authored-By: Claude Fable 5 --- gateway/authz_mixin.py | 10 +++ gateway/kanban_watchers.py | 29 +++++++- tests/gateway/test_kanban_notifier.py | 69 +++++++++++++++++++ tests/gateway/test_multiplex_profile_authz.py | 8 +++ 4 files changed, 114 insertions(+), 2 deletions(-) diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 45f7f32cf2ab7..be57b3f03ef21 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -78,6 +78,16 @@ class GatewayAuthorizationMixin: return None profile_name = (profile or "").strip() or None if profile_name and profile_name != "default": + active_profile = None + active_profile_fn = getattr(self, "_active_profile_name", None) + if callable(active_profile_fn): + try: + active_profile = active_profile_fn() + except Exception: + active_profile = None + if profile_name == active_profile: + adapters = getattr(self, "adapters", None) or {} + return adapters.get(platform) profile_adapters = getattr(self, "_profile_adapters", None) or {} if profile_name in profile_adapters: return profile_adapters[profile_name].get(platform) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index baa59fd9661a2..2563091f4fc41 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -164,7 +164,7 @@ class GatewayKanbanWatchersMixin: # "status" covers dashboard drag-drop and `_set_status_direct()` # writes — surface those transitions to subscribers too. - TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked") + TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked", "block_loop_detected") # Subscriptions are removed only when the task reaches a truly final # status (done / archived). We used to also unsub on any terminal # event kind (gave_up / crashed / timed_out / blocked), but that @@ -181,7 +181,13 @@ class GatewayKanbanWatchersMixin: # means the chat is dead (deleted, bot kicked, etc.) — after N # consecutive send failures the sub is dropped so we don't spin # against a dead chat every 5 seconds forever. - MAX_SEND_FAILURES = 3 + # Raised from 3 to 12 (~60s at the 5s tick cadence): now that a + # reported SendResult(success=False) also lands here (see the + # delivery loop below), a transient Telegram/API outage of a few + # ticks must NOT permanently unsubscribe a live review-gate channel. + # A genuinely dead chat still drops, just ~60s later — a fine trade + # for an unattended gate where a false drop means silent work pileup. + MAX_SEND_FAILURES = 12 sub_fail_counts: dict[tuple, int] = getattr( self, "_kanban_sub_fail_counts", {} ) @@ -413,6 +419,25 @@ class GatewayKanbanWatchersMixin: if ev.payload and ev.payload.get("status"): new_status = str(ev.payload["status"]) msg = f"🔄 {board_tag}{tag}Kanban {sub['task_id']} → {new_status}" + elif kind == "block_loop_detected": + # A task re-blocked for the same cause past the + # recurrence limit and was routed to `triage` for a + # human decision. This is the ONE transition that + # exists to force human attention, yet it emits no + # `blocked`/`status` event — so before adding it to + # TERMINAL_KINDS it produced zero notification and + # the task stalled in triage silently. Ping loudly. + reason = "" + recurrences = None + if ev.payload: + if ev.payload.get("reason"): + reason = f": {str(ev.payload['reason'])[:160]}" + recurrences = ev.payload.get("recurrences") + rc = f" (blocked {recurrences}x for the same cause)" if recurrences else "" + msg = ( + f"🛑 {board_tag}{tag}Kanban {sub['task_id']} routed to TRIAGE" + f" — needs a human decision{rc}{reason}" + ) else: # archived / unblocked are claimed by TERMINAL_KINDS # (so the cursor advances past them and they can't diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 45a22cb78a1ab..f5551899c589f 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -226,6 +226,45 @@ def test_kanban_notifier_rewinds_claim_on_send_exception(tmp_path, monkeypatch): assert [ev.kind for ev in _unseen_terminal_events(tid)] == ["completed"] +class ReportedFailureAdapter: + """Adapter that REPORTS failure via SendResult(success=False) instead of + raising — the exact contract the Telegram adapter uses for 'Not connected' + and degraded-send paths.""" + + def __init__(self): + self.attempts = 0 + + async def send(self, chat_id, text, metadata=None): + self.attempts += 1 + from gateway.platforms.base import SendResult + return SendResult(success=False, error="Not connected") + + +def test_kanban_notifier_rewinds_claim_on_reported_send_failure(tmp_path, monkeypatch): + """A non-raising SendResult(success=False) must NOT advance the cursor. + + Regression for the silent-drop bug: the notifier used to discard send()'s + return value, so a reported (not raised) failure — e.g. Telegram mid- + reconnect after a gateway restart — fell through to the success branch, + marked the event seen, and lost the notification forever. The event must + remain unseen for retry, exactly like the raised-exception path. + """ + db_path = tmp_path / "reported-failure.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + tid = _create_completed_subscription() + + adapter = ReportedFailureAdapter() + runner = _make_runner(adapter) + + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert adapter.attempts >= 1, "send should have been attempted" + assert [ev.kind for ev in _unseen_terminal_events(tid)] == ["completed"], ( + "a reported send failure must rewind the claim, not silently drop the event" + ) + + def test_notifier_redelivers_same_kind_on_dispatch_cycle(tmp_path, monkeypatch): """A retry cycle (crashed → reclaimed → crashed) notifies the user twice. @@ -288,6 +327,36 @@ def test_notifier_redelivers_same_kind_on_dispatch_cycle(tmp_path, monkeypatch): assert "crashed" in adapter.sent[1]["text"].lower() +def test_notifier_delivers_subscription_owned_by_active_profile(tmp_path, monkeypatch): + """A single-profile gateway stamps active profile but keeps adapters primary.""" + db_path = tmp_path / "active-profile-owner.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="owned by active profile", assignee="worker") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat-1", + notifier_profile="dev", + ) + kb.complete_task(conn, tid, summary="done") + finally: + conn.close() + + adapter = RecordingAdapter() + runner = _make_runner(adapter) + runner._active_profile_name = lambda: "dev" + + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert len(adapter.sent) == 1 + assert tid in adapter.sent[0]["text"] + + def test_notifier_owning_profile_adapter_no_default_fallback(tmp_path, monkeypatch): """A subscription owned by a secondary profile whose profile-adapter registry entry EXISTS but lacks this platform must NOT fall back to the diff --git a/tests/gateway/test_multiplex_profile_authz.py b/tests/gateway/test_multiplex_profile_authz.py index 0620b14e0cc26..2a43492e64585 100644 --- a/tests/gateway/test_multiplex_profile_authz.py +++ b/tests/gateway/test_multiplex_profile_authz.py @@ -101,6 +101,14 @@ def test_secondary_allowlist_still_authorized(monkeypatch): assert runner._is_user_authorized(source) is True +def test_active_profile_stamp_resolves_primary_adapter(monkeypatch): + """A single-profile gateway stamps its active profile but stores adapters as primary.""" + runner, default_adapter, _secondary_adapter = _make_multiplex_runner(monkeypatch) + runner._active_profile_name = lambda: "dev" + + assert runner._authorization_adapter(Platform.WECOM, profile="dev") is default_adapter + + def test_adapter_for_source_resolves_secondary_profile_adapter(monkeypatch): """Ingress adapter lookup must use the stamped profile's adapter map.""" runner, default_adapter, secondary_adapter = _make_multiplex_runner(monkeypatch)