fix(gateway): kanban notifier delivery reliability

- 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 <noreply@anthropic.com>

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)

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

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

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)