-
Notifications
You must be signed in to change notification settings - Fork 0
fix(kanban): route notifications via owning profile + wake creator agent (salvage #54872) #447
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.
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
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,8 @@ | |
| from pathlib import Path | ||
| from typing import Any, Callable, Optional | ||
|
|
||
| from agent.i18n import t | ||
|
|
||
| # Match the logger run.py uses (logging.getLogger(__name__) where __name__ == | ||
| # "gateway.run") so extracted log records keep their original logger name. | ||
| logger = logging.getLogger("gateway.run") | ||
|
|
@@ -160,7 +162,9 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: | |
| logger.warning("kanban notifier: kanban_db not importable; notifier disabled") | ||
| return | ||
|
|
||
| TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out") | ||
| # "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") | ||
| # 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 | ||
|
|
@@ -250,11 +254,13 @@ def _collect(): | |
| for sub in subs: | ||
| owner_profile = sub.get("notifier_profile") or None | ||
| if owner_profile and owner_profile != notifier_profile: | ||
| logger.debug( | ||
| "kanban notifier: subscription for %s owned by profile %s; current profile %s skipping", | ||
| sub.get("task_id"), owner_profile, notifier_profile, | ||
| ) | ||
| continue | ||
| _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() | ||
| if platform not in active_platforms: | ||
| logger.debug( | ||
|
|
@@ -304,7 +310,14 @@ def _collect(): | |
| self._kanban_advance, sub, d["cursor"], board_slug, | ||
| ) | ||
| continue | ||
| adapter = self.adapters.get(plat) | ||
| sub_profile = sub.get("notifier_profile") or "" | ||
| adapter = None | ||
| if sub_profile: | ||
| _profile_map = getattr(self, "_profile_adapters", {}).get(sub_profile) | ||
| if _profile_map: | ||
| adapter = _profile_map.get(plat) | ||
| if adapter is None: | ||
| adapter = self.adapters.get(plat) | ||
| if adapter is None: | ||
| logger.debug( | ||
| "kanban notifier: adapter %s disconnected before delivery for %s; rewinding claim", | ||
|
|
@@ -319,6 +332,7 @@ def _collect(): | |
| ) | ||
| continue | ||
| title = (task.title if task else sub["task_id"])[:120] | ||
| board_tag = f"[{board_slug}] " if board_slug else "" | ||
| for ev in d["events"]: | ||
| kind = ev.kind | ||
| # Identity prefix: attribute terminal pings to the | ||
|
|
@@ -345,35 +359,40 @@ def _collect(): | |
| r = lines[0][:160] if lines else task.result[:160] | ||
| handoff = f"\n{r}" | ||
| msg = ( | ||
| f"✔ {tag}Kanban {sub['task_id']} done" | ||
| f"✔ {board_tag}{tag}Kanban {sub['task_id']} done" | ||
| f" — {title}{handoff}" | ||
| ) | ||
| elif kind == "blocked": | ||
| reason = "" | ||
| if ev.payload and ev.payload.get("reason"): | ||
| reason = f": {str(ev.payload['reason'])[:160]}" | ||
| msg = f"⏸ {tag}Kanban {sub['task_id']} blocked{reason}" | ||
| msg = f"⏸ {board_tag}{tag}Kanban {sub['task_id']} blocked{reason}" | ||
| elif kind == "gave_up": | ||
| err = "" | ||
| if ev.payload and ev.payload.get("error"): | ||
| err = f"\n{str(ev.payload['error'])[:200]}" | ||
| msg = ( | ||
| f"✖ {tag}Kanban {sub['task_id']} gave up " | ||
| f"✖ {board_tag}{tag}Kanban {sub['task_id']} gave up " | ||
| f"after repeated spawn failures{err}" | ||
| ) | ||
| elif kind == "crashed": | ||
| msg = ( | ||
| f"✖ {tag}Kanban {sub['task_id']} worker crashed " | ||
| f"✖ {board_tag}{tag}Kanban {sub['task_id']} worker crashed " | ||
| f"(pid gone); dispatcher will retry" | ||
| ) | ||
| elif kind == "timed_out": | ||
| limit = 0 | ||
| if ev.payload and ev.payload.get("limit_seconds"): | ||
| limit = int(ev.payload["limit_seconds"]) | ||
| msg = ( | ||
| f"⏱ {tag}Kanban {sub['task_id']} timed out " | ||
| f"⏱ {board_tag}{tag}Kanban {sub['task_id']} timed out " | ||
| f"(max_runtime={limit}s); will retry" | ||
| ) | ||
| elif kind == "status": | ||
| new_status = "" | ||
| 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}" | ||
|
Comment on lines
361
to
+395
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 Markdown injection in kanban notification messages via unescaped user-controlled fields (security) The
These values are interpolated into f-strings at lines 361-395 and sent to messaging platforms via Example attack: Setting a task title to Data flow: 💡 Suggestion: Escape Markdown-special characters in user-controlled values ( 📋 Prompt for AI AgentsIn |
||
| else: | ||
| continue | ||
|
Comment on lines
167
to
397
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 Missing message handlers for 'archived' and 'unblocked' event kinds in kanban notification dispatch (bug) The After the loop completes (via the Evidence from independent scanners: both the correctness scanner and the domain-kanban-workflow scanner independently identified this gap. 💡 Suggestion: Add message formatting handlers for 📋 Prompt for AI AgentsIn elif kind == "archived":
msg = f"📦 {board_tag}{tag}Kanban {sub['task_id']} archived"
elif kind == "unblocked":
msg = f"▶ {board_tag}{tag}Kanban {sub['task_id']} unblocked"If these event kinds should NOT produce user-visible notifications, instead remove |
||
| metadata: dict[str, Any] = {} | ||
|
|
@@ -460,6 +479,55 @@ def _collect(): | |
| # same state. See the longer comment on TERMINAL_KINDS | ||
| # above for the failure mode this prevents. | ||
| task_terminal = task and task.status in {"done", "archived"} | ||
| _WAKE_KINDS = ("completed", "gave_up", "crashed", "timed_out", "blocked") | ||
| _wake_kinds = {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} | ||
| if _wake_kinds: | ||
| try: | ||
| _session_key = getattr(task, "session_id", None) or "" | ||
| if _session_key: | ||
| _title = (task.title if task else sub["task_id"])[:120] | ||
| _assignee = task.assignee if task else "" | ||
| _parts = [] | ||
| if "completed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.completed")) | ||
| if "gave_up" in _wake_kinds: _parts.append(t("gateway.kanban.wake.gave_up")) | ||
| if "crashed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.crashed")) | ||
| if "timed_out" in _wake_kinds: _parts.append(t("gateway.kanban.wake.timed_out")) | ||
| if "blocked" in _wake_kinds: _parts.append(t("gateway.kanban.wake.blocked")) | ||
| _status = t("gateway.kanban.wake.status_joiner").join(_parts) or t("gateway.kanban.wake.status_default") | ||
| _synth = t( | ||
| "gateway.kanban.wake.message", | ||
| task_id=sub["task_id"], | ||
| status=_status, | ||
| title=_title, | ||
| assignee=_assignee, | ||
| board=board_slug, | ||
| ) | ||
| from gateway.session import SessionSource | ||
| from gateway.platforms.base import MessageEvent, MessageType | ||
| _source = SessionSource( | ||
| platform=plat, | ||
| chat_id=sub["chat_id"], | ||
| chat_type="group", | ||
| thread_id=sub.get("thread_id") or None, | ||
| user_id=sub.get("user_id"), | ||
| profile=sub_profile or None, | ||
| ) | ||
| _synth_event = MessageEvent( | ||
| text=_synth, | ||
| message_type=MessageType.TEXT, | ||
| source=_source, | ||
| internal=True, | ||
| ) | ||
| await adapter.handle_message(_synth_event) | ||
| logger.info( | ||
| "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", | ||
| sub["task_id"], platform_str, sub["chat_id"], sub_profile or "default", _wake_kinds, | ||
| ) | ||
| except Exception as _wk_err: | ||
| logger.debug( | ||
| "kanban notifier: wakeup injection failed for %s: %s", | ||
| sub["task_id"], _wk_err, | ||
| ) | ||
| if task_terminal: | ||
| await asyncio.to_thread( | ||
| self._kanban_unsub, sub, board_slug, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟠 Kanban notification adapter fallback leaks profile isolation in multiplexed gateway (security)
In
gateway/kanban_watchers.py, the_kanban_notifier_watchermethod's_collectclosure resolves which adapter delivers kanban notifications. For cross-profile subscriptions, the code at lines 255-263 guards against attempting delivery when the notifier profile has no adapters for the owner profile:The bug: The guard only checks that
_owner_adaptersis truthy (has ANY adapters), not that the specific platform's adapter exists within it. If profile B has a Discord adapter connected but NOT Telegram, and a subscription wants Telegram delivery, the guard passes (profile B's adapter map is truthy), and the delivery proceeds.Later, in the delivery loop (lines 313-320), the adapter is resolved:
Since
_profile_map.get(plat)returns None (profile B has no Telegram adapter), the fallback at line 320 uses the default profile's Telegram adapter. The notification — including task title, status, assignee, result summary, and any artifacts — is delivered through the wrong profile's adapter, leaking task information across profile boundaries. The subsequent wakeup injection (lines 505-521) further creates aSessionSourcewith the subscription's profile and callsadapter.handle_message()on the wrong adapter.Impact: In a multiplexed gateway serving multiple profiles, kanban task notifications and results leak from one profile into another profile's messaging channels.
💡 Suggestion: Tighten the early-skip guard in
_collect()to also verify the specific platform adapter exists in the owner's profile map. Move the platform string extraction above the guard, parse it into a Platform enum, and checkplat in _owner_adapters. This prevents cross-profile subscriptions from proceeding to delivery when the required platform adapter isn't available in the owner profile's adapter set.📋 Prompt for AI Agents
In
gateway/kanban_watchers.py, in the_kanban_notifier_watchermethod, the_collectclosure around lines 255-264, restructure the cross-profile guard to also check for the specific platform adapter:Move
platform = (sub.get("platform") or "").lower()(currently line 264) ABOVE the profile guard block (before line 255).In the profile guard block (lines 256-263), after verifying
_owner_adaptersexists, parse the platform string into a Platform enum and check it's present:This ensures cross-profile subscriptions are only processed when the notifier profile has the SPECIFIC platform adapter for the subscription owner.