Skip to content

fix: kanban notifier ownership #70091

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
duu261 wants to merge 4 commits into from
Closed

fix: kanban notifier ownership #70091

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
89546b6
fix session profile fallback
duu261 Jul 23, 2026
72c88ca
fix(gateway): decouple notifier polling
duu261 Jul 23, 2026
1420c2b
fix kanban notifier ownership
duu261 Jul 23, 2026
32a9225
fix(kanban): route profile-owned notices
duu261 Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 50 additions & 69 deletions gateway/kanban_watchers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,30 +131,10 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None:
cross boards, so delivery semantics are unchanged — this is
purely a fan-out of the single-DB poll.
"""
# Gate: only the dispatch-owning gateway opens kanban DBs for notifier polling.
# Non-dispatch gateways have no subscriptions to deliver — all kanban state lives
# in the dispatch owner's per-board DBs. This prevents N-gateway -shm contention.
# TODO: gate per-board when per-board dispatcher_owner tracking lands.
try:
from hermes_cli.config import load_config as _load_config
except Exception:
logger.warning("kanban notifier: config loader unavailable; disabled")
return
env_override = os.environ.get("HERMES_KANBAN_DISPATCH_IN_GATEWAY", "").strip().lower()
if env_override in {"0", "false", "no", "off"}:
logger.info("kanban notifier: disabled via HERMES_KANBAN_DISPATCH_IN_GATEWAY env")
return
try:
cfg = _load_config()
except Exception as exc:
logger.warning("kanban notifier: cannot load config (%s); disabled", exc)
return
kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {}
if not kanban_cfg.get("dispatch_in_gateway", True):
logger.info(
"kanban notifier: disabled via config kanban.dispatch_in_gateway=false"
)
return
# A standalone gateway claims only its active profile. A multiplex
# gateway also claims rows for registered secondary profiles and routes
# them through those profiles' adapters. Only the active default
# profile handles legacy ownerless rows.
from gateway.config import Platform as _Platform
try:
from hermes_cli import kanban_db as _kb
Expand Down Expand Up @@ -198,11 +178,18 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None:
try:
def _collect():
deliveries: list[dict] = []
active_platforms = {
getattr(platform, "value", str(platform)).lower()
for platform in self.adapters.keys()
include_ownerless = (notifier_profile == "default")
owned_adapter_maps = {
notifier_profile: getattr(self, "adapters", {})
}
if not active_platforms:
owned_adapter_maps.update({
profile: adapters
for profile, adapters in getattr(
self, "_profile_adapters", {}
).items()
if profile != notifier_profile
})
if not any(owned_adapter_maps.values()):
logger.debug("kanban notifier: no connected adapters; skipping tick")
return deliveries

Expand Down Expand Up @@ -248,20 +235,25 @@ def _collect():
# a legacy DB. `_add_column_if_missing` now
# tolerates that race, but we still skip the
# redundant call to avoid the wasted work.
subs = _kb.list_notify_subs(conn)
subs = []
for owner in owned_adapter_maps:
subs.extend(_kb.list_notify_subs(
conn,
notifier_profile=owner,
include_ownerless=(
include_ownerless
and owner == notifier_profile
),
))
if not subs:
logger.debug("kanban notifier: board %s has no subscriptions", slug)
for sub in subs:
owner_profile = sub.get("notifier_profile") or None
if owner_profile and owner_profile != notifier_profile:
_owner_adapters = getattr(self, "_profile_adapters", {}).get(owner_profile)
if not _owner_adapters:
logger.debug(
"kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, skipping",
sub.get("task_id"), owner_profile, notifier_profile,
)
continue
platform = (sub.get("platform") or "").lower()
sub_profile = sub.get("notifier_profile") or notifier_profile
active_platforms = {
getattr(p, "value", str(p)).lower()
for p in owned_adapter_maps[sub_profile].keys()
}
if platform not in active_platforms:
logger.debug(
"kanban notifier: subscription for %s on %s skipped; adapter not connected",
Expand All @@ -275,6 +267,11 @@ def _collect():
chat_id=sub["chat_id"],
thread_id=sub.get("thread_id") or "",
kinds=TERMINAL_KINDS,
notifier_profile=sub_profile,
include_ownerless=(
include_ownerless
and not sub.get("notifier_profile")
),
)
if not events:
continue
Expand Down Expand Up @@ -311,20 +308,18 @@ def _collect():
)
continue
sub_profile = sub.get("notifier_profile") or ""
# Route via the SAME chokepoint the authorization path uses
# (gateway/authz_mixin.py::_authorization_adapter): a stamped
# profile with its own adapter-registry entry must be served
# by THAT profile's same-platform adapter and must NOT silently
# fall back to the default profile's adapter — otherwise a
# secondary profile's task notification is delivered by the
# wrong bot (the cross-profile mis-delivery this whole change
# exists to fix). The helper returns None only when the profile
# (or default) genuinely has no adapter for the platform.
adapter = self._authorization_adapter(plat, sub_profile or None)
if not sub_profile or sub_profile == notifier_profile:
adapter = getattr(self, "adapters", {}).get(plat)
else:
adapter = (
getattr(self, "_profile_adapters", {})
.get(sub_profile, {})
.get(plat)
)
if adapter is None:
logger.debug(
"kanban notifier: adapter %s disconnected before delivery for %s; rewinding claim",
platform_str, sub["task_id"],
"kanban notifier: adapter %s disconnected before delivery for %s profile=%s; rewinding claim",
platform_str, sub["task_id"], sub_profile or notifier_profile,
)
await asyncio.to_thread(
self._kanban_rewind,
Expand Down Expand Up @@ -514,27 +509,14 @@ def _collect():
)
from gateway.session import SessionSource
from gateway.platforms.base import MessageEvent, MessageType
# KNOWN LIMITATION (tracked follow-up): the
# subscription row does not persist the
# creator's chat_type, and it is not carried
# on the session-context bridge, so we cannot
# faithfully reconstruct the creator's real
# session key here. build_session_key() keys
# DMs (":dm:<chat_id>") on a wholly different
# shape from group/thread, so any hardcoded
# value mis-routes some creators. "group" is
# the least-surprising default for the
# dashboard/group flows this wake primarily
# serves; DM-originated creators are handled
# by the follow-up that stamps + persists
# chat_type end-to-end. handle_message()
# get_or_create_session's the target, so a
# mismatch degrades to "wake lands in a fresh
# group session" — never an exception.
_chat_type = sub.get("chat_type") or "group"
# Legacy rows predate chat_type persistence.
# Default them to group so existing non-DM
# subscriptions keep their old wake shape.
_source = SessionSource(
platform=plat,
chat_id=sub["chat_id"],
chat_type="group",
chat_type=_chat_type,
thread_id=sub.get("thread_id") or None,
user_id=sub.get("user_id"),
profile=sub_profile or None,
Expand Down Expand Up @@ -815,7 +797,6 @@ async def _kanban_dispatcher_watcher(self) -> None:
"kanban dispatcher: advisory lock unavailable at %s; proceeding "
"on config control alone.", _lock_path,
)

try:
interval = float(kanban_cfg.get("dispatch_interval_seconds", 60) or 60)
except (ValueError, TypeError):
Expand Down
3 changes: 2 additions & 1 deletion gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16746,12 +16746,13 @@ def _set_session_env(self, context: SessionContext) -> list:
platform=context.source.platform.value,
chat_id=context.source.chat_id,
chat_name=context.source.chat_name or "",
chat_type=context.source.chat_type or "",
thread_id=str(context.source.thread_id) if context.source.thread_id else "",
user_id=str(context.source.user_id) if context.source.user_id else "",
user_name=str(context.source.user_name) if context.source.user_name else "",
session_key=context.session_key,
message_id=str(context.source.message_id) if context.source.message_id else "",
profile=getattr(context.source, "profile", "") or "",
profile=(getattr(context.source, "profile", "") or self._active_profile_name()),
async_delivery=_async_delivery,
)

Expand Down
5 changes: 5 additions & 0 deletions gateway/session_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ def session_context_engaged() -> bool:
_SESSION_SOURCE: ContextVar = ContextVar("HERMES_SESSION_SOURCE", default=_UNSET)
_SESSION_CHAT_ID: ContextVar = ContextVar("HERMES_SESSION_CHAT_ID", default=_UNSET)
_SESSION_CHAT_NAME: ContextVar = ContextVar("HERMES_SESSION_CHAT_NAME", default=_UNSET)
_SESSION_CHAT_TYPE: ContextVar = ContextVar("HERMES_SESSION_CHAT_TYPE", default=_UNSET)
_SESSION_THREAD_ID: ContextVar = ContextVar("HERMES_SESSION_THREAD_ID", default=_UNSET)
_SESSION_USER_ID: ContextVar = ContextVar("HERMES_SESSION_USER_ID", default=_UNSET)
_SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET)
Expand Down Expand Up @@ -125,6 +126,7 @@ def session_context_engaged() -> bool:
"HERMES_SESSION_SOURCE": _SESSION_SOURCE,
"HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID,
"HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME,
"HERMES_SESSION_CHAT_TYPE": _SESSION_CHAT_TYPE,
"HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID,
"HERMES_SESSION_USER_ID": _SESSION_USER_ID,
"HERMES_SESSION_USER_NAME": _SESSION_USER_NAME,
Expand Down Expand Up @@ -159,6 +161,7 @@ def set_session_vars(
source: str = "",
chat_id: str = "",
chat_name: str = "",
chat_type: str = "",
thread_id: str = "",
user_id: str = "",
user_name: str = "",
Expand Down Expand Up @@ -195,6 +198,7 @@ def set_session_vars(
_SESSION_SOURCE.set(source),
_SESSION_CHAT_ID.set(chat_id),
_SESSION_CHAT_NAME.set(chat_name),
_SESSION_CHAT_TYPE.set(chat_type),
_SESSION_THREAD_ID.set(thread_id),
_SESSION_USER_ID.set(user_id),
_SESSION_USER_NAME.set(user_name),
Expand Down Expand Up @@ -230,6 +234,7 @@ def clear_session_vars(tokens: list) -> None:
_SESSION_SOURCE,
_SESSION_CHAT_ID,
_SESSION_CHAT_NAME,
_SESSION_CHAT_TYPE,
_SESSION_THREAD_ID,
_SESSION_USER_ID,
_SESSION_USER_NAME,
Expand Down
2 changes: 2 additions & 0 deletions gateway/slash_commands.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,6 +479,7 @@ async def _handle_kanban_command(self, event: MessageEvent) -> str:
).lower()
chat_id = str(getattr(source, "chat_id", "") or "")
thread_id = str(getattr(source, "thread_id", "") or "")
chat_type = str(getattr(source, "chat_type", "") or "")
user_id = str(getattr(source, "user_id", "") or "") or None
if platform_str and chat_id:
def _sub():
Expand All @@ -490,6 +491,7 @@ def _sub():
platform=platform_str, chat_id=chat_id,
thread_id=thread_id or None,
user_id=user_id,
chat_type=chat_type or None,
notifier_profile=getattr(self, "_kanban_notifier_profile", None) or self._active_profile_name(),
)
finally:
Expand Down
3 changes: 2 additions & 1 deletion hermes_cli/kanban.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2687,8 +2687,9 @@ def _cmd_notify_list(args: argparse.Namespace) -> int:
return 0
for s in subs:
thr = f":{s['thread_id']}" if s.get("thread_id") else ""
ctype = f":{s['chat_type']}" if s.get("chat_type") else ""
owner = f" owner={s['notifier_profile']}" if s.get("notifier_profile") else ""
print(f" {s['task_id']:10s} {s['platform']}:{s['chat_id']}{thr}"
print(f" {s['task_id']:10s} {s['platform']}:{s['chat_id']}{ctype}{thr}"
f" (since event {s['last_event_id']}){owner}")
return 0

Expand Down
Loading