fix(kanban): profile-correct notification ownership and routing (salvage #57995, #56589, #62380, #69181)

gateway/kanban_watchers.py

@@ -208,6 +208,24 @@ def _collect():
                         getattr(platform, "value", str(platform)).lower()
                         for platform in self.adapters.keys()
                     }
+                    # Widen to every platform any secondary profile has live,
+                    # not just the default profile's. This is only a coarse
+                    # pre-filter to skip claiming events for subs nobody can
+                    # possibly deliver — the precise per-profile check (via
+                    # gateway/authz_mixin.py::_authorization_adapter, which
+                    # forbids default-profile fallback) still runs at delivery
+                    # time below, rewinding the claim if it resolves to None.
+                    # Without this, a subscription owned by a secondary
+                    # profile on a platform the DEFAULT profile never
+                    # connected (e.g. beta owns discord, default doesn't) was
+                    # dropped here before ever being claimed — no rewind
+                    # applies to an unclaimed event, so it silently never
+                    # retries.
+                    for _profile_adapter_map in getattr(self, "_profile_adapters", {}).values():
+                        active_platforms.update(
+                            getattr(platform, "value", str(platform)).lower()
+                            for platform in _profile_adapter_map.keys()
+                        )
                     if not active_platforms:
                         logger.debug("kanban notifier: no connected adapters; skipping tick")
                         return deliveries

hermes_cli/kanban_db.py

@@ -8815,6 +8815,12 @@ def _default_spawn(
 
     prompt = f"work kanban task {task.id}"
     env = dict(os.environ)
+    # The dispatcher is detached from every conversation. Its worker must never
+    # inherit routing mirrored by a previous gateway turn, even before the first
+    # session binds ContextVars in this process.
+    from gateway.session_context import _VAR_MAP
+    for key in _VAR_MAP:
+        env.pop(key, None)
 
     # Inject HERMES_HOME so the worker reads the profile-scoped config.yaml
     # (fallback_providers, toolsets, agent settings, etc.) instead of the root

tests/gateway/test_kanban_notifier.py

@@ -174,6 +174,44 @@ def test_kanban_notifier_rewinds_claim_if_adapter_disconnects(tmp_path, monkeypa
     assert [ev.kind for ev in _unseen_terminal_events(tid)] == ["completed"]
 
 
+def test_active_named_profile_subscription_is_delivered(tmp_path, monkeypatch):
+    """A sub stamped with the gateway's own named profile uses self.adapters.
+
+    Regression for #71340: on a standalone (non-multiplex) gateway running a
+    named profile, _authorization_adapter() used to treat the active name as a
+    multiplex secondary, find no _profile_adapters entry, fail closed, and
+    rewind the claim forever — silent zero-delivery.
+    """
+    db_path = tmp_path / "actionable-block.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    reason = "AGE-39 — https://linear.example/AGE-39 — publishing verified."
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="approval", assignee="publisher")
+        kb.add_notify_sub(
+            conn,
+            task_id=tid,
+            platform="telegram",
+            chat_id="chat-1",
+            notifier_profile="main",
+        )
+        kb.block_task(conn, tid, reason=reason, kind="needs_input")
+    finally:
+        conn.close()
+
+    adapter = RecordingAdapter()
+    runner = _make_runner(adapter)
+    runner._active_profile_name = lambda: "main"
+
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    assert len(adapter.sent) == 1
+    message = adapter.sent[0]["text"]
+    assert tid in message
+    assert "blocked" in message
+
+
 def test_kanban_db_path_is_test_isolated_from_real_home():
     hermes_home = Path(kb.kanban_home())
     production_db = Path.home() / ".hermes" / "kanban.db"
@@ -416,6 +454,55 @@ def test_notifier_owning_profile_adapter_no_default_fallback(tmp_path, monkeypat
     assert [ev.kind for ev in _unseen_terminal_events_for(tid, "chat-beta")] == ["completed"]
 
 
+def test_notifier_claims_platform_only_a_secondary_profile_owns(tmp_path, monkeypatch):
+    """A subscription owned by a secondary profile on a platform the DEFAULT
+    profile never connected must still be claimed and delivered.
+
+    Regression: the ``_collect()`` pre-filter built ``active_platforms``
+    solely from ``self.adapters`` (the default profile). A sub owned by
+    profile "beta" on "discord", where beta genuinely has a live discord
+    adapter but the default profile has no discord adapter at all, was
+    dropped by that pre-filter (``platform not in active_platforms``)
+    before ``claim_unseen_events_for_sub`` ever ran — unlike the
+    disconnected-adapter path, an unclaimed event is never rewound, so this
+    was a permanent, silent notification loss, not a retryable one. This
+    directly contradicts the feature's own purpose (routing notifications
+    via the owning profile), and is the same cross-profile-adapter-lookup
+    class the delivery-side chokepoint in
+    ``test_notifier_owning_profile_adapter_no_default_fallback`` already
+    guards — just one gate earlier.
+    """
+    db_path = tmp_path / "secondary-only-platform.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="owned by beta on discord", assignee="worker")
+        kb.add_notify_sub(
+            conn, task_id=tid, platform="discord", chat_id="chat-beta",
+            notifier_profile="beta",
+        )
+        kb.complete_task(conn, tid, summary="done")
+    finally:
+        conn.close()
+
+    beta_adapter = RecordingAdapter()
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner._running = True
+    # Default profile has NO discord adapter at all.
+    runner.adapters = {Platform.TELEGRAM: RecordingAdapter()}
+    # Secondary profile "beta" has a live discord adapter.
+    runner._profile_adapters = {"beta": {Platform.DISCORD: beta_adapter}}
+    runner._kanban_sub_fail_counts = {}
+
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    assert len(beta_adapter.sent) == 1, (
+        f"beta's discord adapter should have received the notification; got {beta_adapter.sent!r}"
+    )
+
+
 def test_notifier_wakeup_uses_subscription_chat_type(tmp_path, monkeypatch):
     db_path = tmp_path / "chat-type-wakeup.db"
     monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))

tests/gateway/test_multiplex_profile_authz.py

@@ -209,6 +209,15 @@ def test_adapter_for_direct_source_keeps_native_platform_adapter(monkeypatch):
     assert runner._adapter_for_source(source) is slack_adapter
 
 
+def test_explicit_active_profile_stamp_uses_default_adapter_map(monkeypatch):
+    """A named active profile is not misclassified as multiplex secondary."""
+    runner, default_adapter, _secondary_adapter = _make_multiplex_runner(monkeypatch)
+    runner._active_profile_name = lambda: "main"
+
+    assert runner._authorization_adapter(Platform.WECOM, profile="main") is default_adapter
+
+
+
 def test_secondary_allowlist_dm_behavior_ignores_unauthorized(monkeypatch):
     """Unauthorized-DM behavior must read the secondary adapter's dm_policy."""
     runner, _default_adapter, secondary_adapter = _make_multiplex_runner(monkeypatch)

tests/hermes_cli/test_kanban_db.py

@@ -3096,17 +3096,23 @@ def test_empty_per_path_overrides_fall_through(
         assert kb.kanban_db_path() == default_home / "kanban.db"
         assert kb.workspaces_root() == default_home / "kanban" / "workspaces"
 
-    def test_dispatcher_spawn_injects_kanban_db_and_workspaces_root(
+    def test_dispatcher_spawn_injects_kanban_paths_without_stale_session(
         self, tmp_path, monkeypatch
     ):
-        # The dispatcher's `_default_spawn` must inject HERMES_KANBAN_DB
-        # and HERMES_KANBAN_WORKSPACES_ROOT into the worker env so the
-        # worker converges on the dispatcher's paths even when the
-        # `-p <profile>` flag rewrites HERMES_HOME.
+        # The dispatcher must pin board paths while stripping any unrelated
+        # HERMES_SESSION_* identity inherited from the long-lived gateway.
         default_home = tmp_path / ".hermes"
         default_home.mkdir()
         self._set_home(monkeypatch, tmp_path, default_home)
 
+        from gateway import session_context as sc
+
+        # A dispatcher can launch before the gateway binds its first session.
+        monkeypatch.setattr(sc, "_session_context_engaged", False)
+        sc.reset_session_vars()
+        for key in sc._VAR_MAP:
+            monkeypatch.setenv(key, "stale-routing-value")
+
         captured = {}
 
         class _FakePopen:
@@ -3144,6 +3150,8 @@ def __init__(self, cmd, **kwargs):
         )
         assert env["HERMES_KANBAN_TASK"] == "t_dispatch_env"
         assert env["HERMES_KANBAN_BRANCH"] == "wt/t_dispatch_env"
+        for key in sc._VAR_MAP:
+            assert key not in env
 
 
 # ---------------------------------------------------------------------------

tests/tools/test_kanban_tools.py

@@ -2416,6 +2416,7 @@ def _sub_index(subs):
                 "thread_id": getattr(s, "thread_id", None),
                 "user_id": getattr(s, "user_id", None),
                 "delivery_metadata": getattr(s, "delivery_metadata", None),
+                "notifier_profile": getattr(s, "notifier_profile", None),
             })
     return out
 
@@ -2457,6 +2458,30 @@ def test_create_subscribes_gateway_session(monkeypatch, worker_env):
     }
 
 
+def test_create_subscribes_gateway_session_with_active_profile_when_env_missing(monkeypatch, worker_env):
+    """Gateway auto-subscribe rows must be owned by the active profile even
+    when session/env profile markers are missing. Otherwise every Telegram
+    gateway with the same chat_id can deliver another bot's Kanban event."""
+    from tools import kanban_tools as kt
+    monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram")
+    monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "chat-42")
+    monkeypatch.delenv("HERMES_SESSION_PROFILE", raising=False)
+    monkeypatch.delenv("HERMES_PROFILE", raising=False)
+    monkeypatch.setattr("hermes_cli.profiles.get_active_profile_name", lambda: "spanorama")
+
+    out = kt._handle_create({
+        "title": "auto-sub active profile",
+        "assignee": "peer",
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    assert d["subscribed"] is True, d
+
+    subs = _sub_index(_list_subs_for_task(d["task_id"]))
+    assert len(subs) == 1
+    assert subs[0]["notifier_profile"] == "spanorama"
+
+
 def test_create_subscribes_tui_session_via_session_key(monkeypatch, worker_env):
     """TUI / desktop sessions don't have a platform/chat_id (single
     local channel), but the parent process exports HERMES_SESSION_KEY.

tools/kanban_tools.py

@@ -1339,6 +1339,12 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool:
             get_session_env("HERMES_SESSION_PROFILE", "")
             or os.environ.get("HERMES_PROFILE")
         )
+        if not notifier_profile:
+            try:
+                from hermes_cli.profiles import get_active_profile_name
+                notifier_profile = get_active_profile_name() or "default"
+            except Exception:
+                notifier_profile = "default"
         delivery_metadata: dict[str, Any] = {}
         if thread_id:
             delivery_metadata["thread_id"] = thread_id