diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 884a60948c9c9..0c86caea465ab 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -63,6 +63,30 @@ def _authorization_adapter( return None profile_name = (profile or "").strip() or None if profile_name and profile_name != "default": + # A named profile can run in two shapes: + # + # 1. Multiplex secondary profile: adapter lives in + # ``_profile_adapters[profile_name]``. + # 2. Standalone profile gateway (``hermes --profile profile-pmo + # gateway run``): that profile is the active/primary runtime and + # its adapter lives in ``self.adapters``. + # + # The original fail-closed branch handled (1) but broke (2): Kanban + # notify rows stamped ``notifier_profile='profile-pmo'`` could not + # resolve the live Discord adapter, so blocked/completed events were + # never delivered from standalone role services. + active_profile = None + try: + active_profile_fn = getattr(self, "_active_profile_name", None) + if callable(active_profile_fn): + active_profile = str(active_profile_fn() or "").strip() or None + except Exception: + active_profile = None + if active_profile == profile_name: + adapters = getattr(self, "adapters", None) or {} + adapter = adapters.get(platform) + if adapter is not None: + return adapter 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 eb1c68ffd66b9..a6839aeb285ca 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -11,6 +11,7 @@ from __future__ import annotations import asyncio +import inspect import logging import os import sqlite3 @@ -200,8 +201,22 @@ def _collect(): deliveries: list[dict] = [] active_platforms = { getattr(platform, "value", str(platform)).lower() - for platform in self.adapters.keys() + for platform in getattr(self, "adapters", {}).keys() } + # Multiplexed role gateways keep secondary-profile adapters + # under ``_profile_adapters`` while subscriptions are stamped + # with ``notifier_profile`` (for example ``profile-pmo``). + # The delivery path below correctly resolves those adapters + # with ``_authorization_adapter()``, so the pre-claim + # platform gate must consider them too. Otherwise PMO-owned + # subscriptions stay at cursor 0 forever: the notifier sees + # only the default adapter map, decides ``discord`` is not + # active, and never claims/delivers blocked/completed events. + for profile_adapters in (getattr(self, "_profile_adapters", {}) or {}).values(): + active_platforms.update( + getattr(platform, "value", str(platform)).lower() + for platform in profile_adapters.keys() + ) if not active_platforms: logger.debug("kanban notifier: no connected adapters; skipping tick") return deliveries @@ -343,6 +358,11 @@ def _collect(): # chat subscribes to many tasks) legible at a glance. who = (task.assignee if task and task.assignee else None) tag = f"@{who} " if who else "" + rich_approval: dict[str, Any] | None = None + rich_protocol_violation: dict[str, Any] | None = None + metadata: dict[str, Any] = {} + if sub.get("thread_id"): + metadata["thread_id"] = sub["thread_id"] if kind == "completed": # Prefer the run's summary (the worker's # intentional human-facing handoff, carried @@ -367,9 +387,40 @@ def _collect(): ) elif kind == "blocked": reason = "" + block_kind = "" if ev.payload and ev.payload.get("reason"): reason = f": {str(ev.payload['reason'])[:160]}" + if ev.payload and ev.payload.get("kind"): + block_kind = ev.payload.get("kind") msg = f"āø {board_tag}{tag}Kanban {sub['task_id']} blocked{reason}" + if block_kind == "needs_input": + msg += "\nšŸ”” **Human approval required**" + reason_text = "" + if ev.payload and ev.payload.get("reason"): + reason_text = str(ev.payload["reason"]) + approval_context = { + "why": reason_text or "The worker blocked with `kind=needs_input` and needs a human decision before it can safely continue.", + "risk": "Approving lets the assigned worker continue from the current blocked state. Rejecting preserves the block and asks for remediation instead of continuing down the rejected path.", + "approve_means": "Approve means: append `HUMAN_DECISION: APPROVED`, unblock the task, and allow the worker to continue.", + "reject_means": "Reject means: append `HUMAN_DECISION: REJECTED`, keep the task blocked, and create a remediation child task for follow-up.", + } + rich_approval = { + "chat_id": sub["chat_id"], + "task_id": sub["task_id"], + "board": board_slug or _kb.DEFAULT_BOARD, + "title": title, + "summary": reason_text or "Human approval required.", + "progress": ( + f"Task `{sub['task_id']}` is blocked and waiting for a human decision." + ), + "next_steps": ( + "Approve to append `HUMAN_DECISION: APPROVED` and unblock the task; " + "reject to append `HUMAN_DECISION: REJECTED`, keep it blocked, and create a remediation child task." + ), + "approval_context": approval_context, + "session_key": getattr(task, "session_id", None) or "", + "metadata": metadata, + } elif kind == "gave_up": err = "" if ev.payload and ev.payload.get("error"): @@ -378,6 +429,24 @@ def _collect(): f"āœ– {board_tag}{tag}Kanban {sub['task_id']} gave up " f"after repeated spawn failures{err}" ) + payload = ev.payload or {} + is_protocol_violation = bool( + payload.get("protocol_violation") + or payload.get("protocol_violations") + or "protocol violation" in str(payload.get("error") or "").lower() + ) + if is_protocol_violation: + rich_protocol_violation = { + "chat_id": sub["chat_id"], + "task_id": sub["task_id"], + "board": board_slug or _kb.DEFAULT_BOARD, + "title": title, + "assignee": task.assignee if task and task.assignee else "", + "error": str(payload.get("error") or "Worker exited without a terminal Kanban lifecycle call."), + "violation_count": int(payload.get("protocol_violations") or 0), + "violation_limit": int(payload.get("protocol_violation_limit") or 0), + "metadata": metadata, + } elif kind == "crashed": msg = ( f"āœ– {board_tag}{tag}Kanban {sub['task_id']} worker crashed " @@ -405,17 +474,37 @@ def _collect(): # internal transition. They are also excluded from # _WAKE_KINDS below, so they never wake the creator. continue - metadata: dict[str, Any] = {} - if sub.get("thread_id"): - metadata["thread_id"] = sub["thread_id"] sub_key = ( sub["task_id"], sub["platform"], sub["chat_id"], sub.get("thread_id") or "", ) try: - await adapter.send( - sub["chat_id"], msg, metadata=metadata, - ) + send_kanban_approval = getattr(adapter, "send_kanban_approval", None) + send_kanban_protocol_violation = getattr(adapter, "send_kanban_protocol_violation", None) + if rich_protocol_violation and callable(send_kanban_protocol_violation) and inspect.iscoroutinefunction(send_kanban_protocol_violation): + result = await send_kanban_protocol_violation(**rich_protocol_violation) + if not getattr(result, "success", False): + logger.warning( + "kanban notifier: protocol violation decision card send failed for %s, falling back to text: %s", + sub["task_id"], getattr(result, "error", "unknown error"), + ) + await adapter.send( + sub["chat_id"], msg, metadata=metadata, + ) + elif rich_approval and callable(send_kanban_approval) and inspect.iscoroutinefunction(send_kanban_approval): + result = await send_kanban_approval(**rich_approval) + if not getattr(result, "success", False): + logger.warning( + "kanban notifier: rich approval send failed for %s, falling back to text: %s", + sub["task_id"], getattr(result, "error", "unknown error"), + ) + await adapter.send( + sub["chat_id"], msg, metadata=metadata, + ) + else: + await adapter.send( + sub["chat_id"], msg, metadata=metadata, + ) logger.debug( "kanban notifier: delivered %s event for %s to %s/%s on board %s", kind, sub["task_id"], platform_str, sub["chat_id"], board_slug, @@ -1172,6 +1261,28 @@ def _auto_decompose_tick(auto_decompose_per_tick: int) -> int: if attempted >= auto_decompose_per_tick: break attempted += 1 + # Skip auto-decompose for worker-created tasks (created_by=profile-pmo). + # Inspection failures should not block the normal decomposer path. + try: + conn = _kb.connect(board=slug) + try: + task = _kb.get_task(conn, tid) + except Exception: + task = None + finally: + conn.close() + if getattr(task, "created_by", None) == "profile-pmo": + logger.info( + "kanban auto-decompose: skipping %s because created_by=profile-pmo", + tid, + ) + continue + except Exception: + logger.debug( + "kanban auto-decompose: created_by inspection failed for %s", + tid, + exc_info=True, + ) try: outcome = _decomp.decompose_task( tid, author="auto-decomposer", diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index de332f36ee44e..da8c909b6d768 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -2667,6 +2667,12 @@ def create_task( "INSERT OR IGNORE INTO task_links (parent_id, child_id) VALUES (?, ?)", (pid, task_id), ) + _copy_parent_subscriptions_to_children_in_txn( + conn, + parent_task_id=pid, + child_ids=[task_id], + created_at=now, + ) _append_event( conn, task_id, @@ -2822,10 +2828,16 @@ def link_tasks(conn: sqlite3.Connection, parent_id: str, child_id: str) -> None: raise ValueError( f"linking {parent_id} -> {child_id} would create a cycle" ) - conn.execute( + cur = conn.execute( "INSERT OR IGNORE INTO task_links (parent_id, child_id) VALUES (?, ?)", (parent_id, child_id), ) + if cur.rowcount: + _copy_parent_subscriptions_to_children_in_txn( + conn, + parent_task_id=parent_id, + child_ids=[child_id], + ) # If child was ready but parent is not yet done, demote child to todo. parent_status = conn.execute( "SELECT status FROM tasks WHERE id = ?", (parent_id,) @@ -5324,6 +5336,7 @@ def decompose_triage_task( children: list[dict], author: Optional[str] = None, auto_promote: bool = True, + auto_subscribe: bool = True, ) -> Optional[list[str]]: """Fan a triage task out into child tasks and promote the root to ``todo``. @@ -5529,6 +5542,14 @@ def decompose_triage_task( }, ) + if auto_subscribe: + _copy_parent_subscriptions_to_children_in_txn( + conn, + parent_task_id=task_id, + child_ids=child_ids, + created_at=now, + ) + # Outside the write_txn: promote parent-free children to 'ready' # so the dispatcher picks them up on its next tick. Same pattern # specify_triage_task uses. When auto_promote is False children @@ -5539,6 +5560,78 @@ def decompose_triage_task( return child_ids +def _copy_parent_subscriptions_to_children_in_txn( + conn: sqlite3.Connection, + *, + parent_task_id: str, + child_ids: list[str], + created_at: Optional[int] = None, +) -> int: + """Copy notification subscriptions from a parent task to child tasks. + + Caller must already hold the Kanban write transaction. This keeps + decomposition atomic and avoids calling ``add_notify_sub``, which opens its + own transaction. + """ + if not child_ids: + return 0 + rows = conn.execute( + """ + SELECT platform, chat_id, thread_id, user_id, notifier_profile, last_event_id + FROM kanban_notify_subs + WHERE task_id = ? + """, + (parent_task_id,), + ).fetchall() + if not rows: + return 0 + + now = int(created_at if created_at is not None else time.time()) + created = 0 + for child_id in child_ids: + for row in rows: + cur = conn.execute( + """ + INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, + notifier_profile, created_at, last_event_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + child_id, + row["platform"], + row["chat_id"], + row["thread_id"] or "", + row["user_id"], + row["notifier_profile"], + now, + row["last_event_id"] or 0, + ), + ) + created += int(cur.rowcount or 0) + return created + + +def _copy_parent_subscriptions_to_children( + conn: sqlite3.Connection, + *, + parent_task_id: str, + child_ids: list[str], +) -> int: + """Copy a task's notification subscriptions to children. + + Exposed for tests and maintenance scripts. Idempotent because the notify + table primary key is ``(task_id, platform, chat_id, thread_id)``. + """ + with write_txn(conn): + return _copy_parent_subscriptions_to_children_in_txn( + conn, + parent_task_id=parent_task_id, + child_ids=child_ids, + ) + + + def archive_task(conn: sqlite3.Connection, task_id: str) -> bool: with write_txn(conn): cur = conn.execute( diff --git a/hermes_cli/kanban_decompose.py b/hermes_cli/kanban_decompose.py index d6c248c658a6d..aa72354ccd595 100644 --- a/hermes_cli/kanban_decompose.py +++ b/hermes_cli/kanban_decompose.py @@ -45,6 +45,7 @@ from hermes_cli import kanban_db as kb from hermes_cli import profiles as profiles_mod +from hermes_cli.skill_alias_registry import normalize_skill_name logger = logging.getLogger(__name__) @@ -295,6 +296,7 @@ def decompose_task( default_assignee = _resolve_default_assignee(cfg) kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} auto_promote = bool(kanban_cfg.get("auto_promote_children", True)) + auto_subscribe = bool(kanban_cfg.get("auto_subscribe_on_decompose", True)) roster, valid_names = _build_roster() try: @@ -422,11 +424,24 @@ def decompose_task( parents = [] # Clean parent indices: drop non-int and out-of-range. clean_parents = [p for p in parents if isinstance(p, int) and 0 <= p < len(raw_tasks) and p != idx] + + # Normalize skills if provided. + skills = entry.get("skills") + if skills is not None: + if not isinstance(skills, list): + logger.warning("decompose: task %s child %d skills is not a list, ignoring", task_id, idx) + skills = None + else: + skills = [normalize_skill_name(s) for s in skills if s] + if not skills: + skills = None + children.append({ "title": title.strip()[:200], "body": body.strip(), "assignee": chosen, "parents": clean_parents, + "skills": skills, }) try: @@ -438,6 +453,7 @@ def decompose_task( children=children, author=audit_author, auto_promote=auto_promote, + auto_subscribe=auto_subscribe, ) except ValueError as exc: return DecomposeOutcome(task_id, False, f"DB rejected graph: {exc}") diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 215ae0d46d18b..ba0164f33c98c 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -18,6 +18,7 @@ import math import os import re +import shlex import struct import subprocess import tempfile @@ -5150,9 +5151,22 @@ async def _handler(interaction: discord.Interaction): config_overrides = _resolve_config_gates() + # Only PMO profile should have /kanban slash command + kanban_allowed = self.config.extra.get("kanban_slash_command", False) + + # Force re-sync: remove stale /kanban command if not allowed for this profile + if not kanban_allowed: + try: + tree.remove_command("kanban") + except Exception: + pass + for cmd_def in COMMAND_REGISTRY: if not _is_gateway_available(cmd_def, config_overrides): continue + # Skip kanban command for non-PMO profiles + if cmd_def.name == "kanban" and not kanban_allowed: + continue # Discord command names: lowercase, hyphens OK, max 32 chars. discord_name = cmd_def.name.lower()[:32] if discord_name in already_registered: @@ -6593,6 +6607,177 @@ async def send_slash_confirm( except Exception as e: return SendResult(success=False, error=str(e)) + async def send_kanban_approval( + self, + chat_id: str, + task_id: str, + board: str, + title: str, + summary: str, + progress: str, + next_steps: str, + session_key: str, + metadata: Optional[dict] = None, + approval_context: Optional[dict] = None, + ) -> SendResult: + """Send a Kanban approval gate message with Approve/Reject buttons. + + This is used when a Kanban task is blocked with `kind=needs_input` + (human approval gate). The buttons allow an authorized human to approve + or reject the task. Approve records `HUMAN_DECISION: APPROVED` and + unblocks the task; reject records `HUMAN_DECISION: REJECTED` and keeps + it blocked for remediation/follow-up. + """ + if not self._client or not DISCORD_AVAILABLE: + return SendResult(success=False, error="Not connected") + + try: + target_id = chat_id + if metadata and metadata.get("thread_id"): + target_id = metadata["thread_id"] + + channel = self._client.get_channel(int(target_id)) + if not channel: + channel = await self._client.fetch_channel(int(target_id)) + + ctx = approval_context or {} + why = str(ctx.get("why") or summary or "Human approval required.") + risk = str(ctx.get("risk") or "Approving allows the worker to continue; rejecting keeps the task blocked for remediation.") + approve_means = str(ctx.get("approve_means") or "Approve means: record approval and unblock the task.") + reject_means = str(ctx.get("reject_means") or "Reject means: record rejection, keep the task blocked, and create a remediation child task.") + + # Build the approval message with full context + content = ( + f"šŸ”” **Kanban Approval Gate Required**\n\n" + f"**Task:** {task_id}\n" + f"**Title:** {title}\n\n" + f"**Why approval is needed:**\n{why}\n\n" + f"**Risk / consequence:**\n{risk}\n\n" + f"**Progress:**\n{progress}\n\n" + f"**Decision effects:**\n{approve_means}\n{reject_means}\n\n" + f"**Next Steps (if approved):**\n{next_steps}\n\n" + f"**Summary:** {summary}" + ) + + embed = discord.Embed( + title="šŸ”” Kanban Approval Gate Required", + description=f"**Task:** `{task_id}` \n**Title:** {title}", + color=discord.Color.orange(), + ) + embed.add_field(name="ā“ Why approval is needed", value=why[:1024] or "ā€”", inline=False) + embed.add_field(name="āš ļø Risk / consequence", value=risk[:1024] or "ā€”", inline=False) + embed.add_field(name="šŸ“‹ Progress So Far", value=progress[:1024] or "ā€”", inline=False) + embed.add_field(name="āœ… Approve means", value=approve_means[:1024] or "ā€”", inline=False) + embed.add_field(name="āŒ Reject means", value=reject_means[:1024] or "ā€”", inline=False) + embed.add_field(name="āž”ļø Next Steps (if approved)", value=next_steps[:1024] or "ā€”", inline=False) + embed.add_field(name="šŸ“ Summary", value=summary[:1024] or "ā€”", inline=False) + embed.set_footer(text=f"Task ID: {task_id} | Board: {board}") + + # Mirror in plain content for clients where embeds don't render. + # Keep the same decision context as the embed so mobile / compact + # clients still show the consequence of each button. + + view = KanbanApprovalView( + task_id=task_id, + board=board, + channel_id=target_id, + allowed_user_ids=self._allowed_user_ids, + allowed_role_ids=self._allowed_role_ids, + ) + + msg = await channel.send(content=content, embed=embed, view=view) + view._message = msg + return SendResult(success=True, message_id=str(msg.id)) + + except Exception as e: + return SendResult(success=False, error=str(e)) + + async def send_kanban_protocol_violation( + self, + chat_id: str, + task_id: str, + board: str, + title: str, + assignee: str, + error: str, + violation_count: int = 0, + violation_limit: int = 0, + metadata: Optional[dict] = None, + ) -> SendResult: + """Send a production decision card for Kanban protocol violations. + + This is distinct from `send_kanban_approval`: a protocol violation is + not a simple approve/reject gate. The human needs operational choices: + retry, mark done after external verification, create remediation, or + keep blocked for investigation. + """ + if not self._client or not DISCORD_AVAILABLE: + return SendResult(success=False, error="Not connected") + + try: + target_id = chat_id + if metadata and metadata.get("thread_id"): + target_id = metadata["thread_id"] + + channel = self._client.get_channel(int(target_id)) + if not channel: + channel = await self._client.fetch_channel(int(target_id)) + + error_text = str(error or "Protocol violation: worker exited without kanban_complete or kanban_block.").strip() + count_label = ( + f"{violation_count}/{violation_limit}" + if violation_count and violation_limit + else "threshold reached" + ) + content = ( + "āš ļø **Kanban Protocol Violation ā€” Human Decision Needed**\n\n" + f"**Task:** {task_id}\n" + f"**Title:** {title}\n" + f"**Assignee:** {assignee or 'unknown'}\n" + f"**Violation count:** {count_label}\n\n" + "**What happened:**\n" + f"{error_text}\n\n" + "**Choose one:**\n" + "šŸ” Retry worker ā€” promote the blocked task back to ready.\n" + "āœ… Mark verified done ā€” only if you already checked the artifact/output manually.\n" + "šŸ›  Create remediation ā€” create an idempotent triage child task.\n" + "ā›” Keep blocked ā€” leave it stopped and record the decision." + ) + + embed = discord.Embed( + title="āš ļø Kanban Protocol Violation", + description=f"**Task:** `{task_id}` \n**Title:** {title}", + color=discord.Color.orange(), + ) + embed.add_field(name="Assignee", value=assignee or "unknown", inline=True) + embed.add_field(name="Violation count", value=count_label, inline=True) + embed.add_field(name="What happened", value=error_text[:1024] or "ā€”", inline=False) + embed.add_field( + name="Decision options", + value=( + "šŸ” Retry worker ā€” promote task to ready.\n" + "āœ… Mark verified done ā€” human has checked the output.\n" + "šŸ›  Create remediation ā€” idempotent triage child.\n" + "ā›” Keep blocked ā€” audit only." + ), + inline=False, + ) + embed.set_footer(text=f"Task ID: {task_id} | Board: {board}") + + view = KanbanProtocolViolationView( + task_id=task_id, + board=board, + channel_id=target_id, + allowed_user_ids=self._allowed_user_ids, + allowed_role_ids=self._allowed_role_ids, + error=error_text, + ) + msg = await channel.send(content=content, embed=embed, view=view) + view._message = msg + return SendResult(success=True, message_id=str(msg.id)) + except Exception as e: + return SendResult(success=False, error=str(e)) + async def send_clarify( self, chat_id: str, @@ -7773,7 +7958,7 @@ def _define_discord_view_classes() -> None: lazy install sets DISCORD_AVAILABLE=True but leaves the classes undefined, causing NameError on the first button interaction. """ - global ExecApprovalView, SlashConfirmView, UpdatePromptView, ModelPickerView, ClarifyChoiceView, ChoicePickerView + global ExecApprovalView, SlashConfirmView, UpdatePromptView, ModelPickerView, ClarifyChoiceView, ChoicePickerView, KanbanApprovalView, KanbanProtocolViolationView class ExecApprovalView(discord.ui.View): """ @@ -7929,6 +8114,424 @@ async def on_timeout(self): except Exception: pass # message deleted or too old to edit + if all( + hasattr(discord.ui, attr) for attr in ("Modal", "TextInput") + ) and hasattr(discord, "TextStyle"): + class KanbanDecisionCommentModal(discord.ui.Modal): + """Collect an optional human reason before resolving a Kanban gate.""" + + def __init__(self, view, decision: str, color: discord.Color, label: str): + super().__init__(title=f"{label} with comment") + self._approval_view = view + self._decision = decision + self._color = color + self._label = label + self.reason = discord.ui.TextInput( + label="Reason / instruction", + style=discord.TextStyle.paragraph, + required=False, + max_length=1000, + placeholder="Optional: explain constraints, rollback request, or what must change next.", + ) + self.add_item(self.reason) + + async def on_submit(self, interaction: discord.Interaction): + await self._approval_view._resolve( + interaction, + self._decision, + self._color, + self._label, + note=str(self.reason.value or "").strip(), + ) + else: + KanbanDecisionCommentModal = None + + + class KanbanApprovalView(discord.ui.View): + """Two-button view for Kanban human approval gates. + + Used when a Kanban task is blocked with `kind=needs_input` (human approval gate). + Buttons map to the gateway's two choices: + + * "āœ… Approve" ā†’ complete the task with HUMAN_DECISION: APPROVED + * "āŒ Reject" ā†’ complete the task with HUMAN_DECISION: REJECTED + + Clicking calls the internal handler to complete the task via kanban_complete. + Only users in the adapter's allowlist can click. Times out after 10 minutes. + """ + + def __init__( + self, + task_id: str, + board: str, + channel_id: str, + allowed_user_ids: set, + allowed_role_ids: Optional[set] = None, + ): + super().__init__(timeout=600) # 10 minutes + self.task_id = task_id + self.board = board + self.channel_id = channel_id + self.allowed_user_ids = allowed_user_ids + self.allowed_role_ids = allowed_role_ids or set() + self.resolved = False + + def _check_auth(self, interaction: discord.Interaction) -> bool: + return _component_check_auth( + interaction, self.allowed_user_ids, self.allowed_role_ids + ) + + async def _resolve( + self, + interaction: discord.Interaction, + decision: str, + color: discord.Color, + label: str, + note: Optional[str] = None, + ): + if self.resolved: + await interaction.response.send_message( + "This approval gate has already been resolved~", ephemeral=True + ) + return + if not self._check_auth(interaction): + await interaction.response.send_message( + "You're not authorized to approve/reject this task~", ephemeral=True + ) + return + + self.resolved = True + + # Update the embed with the decision + embed = interaction.message.embeds[0] if interaction.message.embeds else None + if embed: + embed.color = color + embed.set_footer(text=f"{label} by {interaction.user.display_name}") + + # Disable all buttons + for child in self.children: + child.disabled = True + + try: + await interaction.response.edit_message(embed=embed, view=self) + except Exception as exc: + # Discord interactions expire quickly and duplicate clicks can + # produce 10062/Unknown interaction. The Kanban decision is the + # important durable side effect, so do not let a stale UI ack + # prevent comment/unblock. Valid fresh clicks still update the + # original message normally. + logger.debug("Kanban approval gate UI update failed: %s", exc) + + # Record the human decision in comments. APPROVED additionally + # unblocks the task so the assigned worker can continue; REJECTED + # deliberately keeps the task blocked so the agent cannot proceed + # down a rejected path without a human follow-up/change request. + try: + from hermes_cli.kanban import run_slash + + decision_text = "HUMAN_DECISION: APPROVED" if decision == "approve" else "HUMAN_DECISION: REJECTED" + actor = str(getattr(interaction.user, "display_name", "") or getattr(interaction.user, "id", "user")) + note_text = str(note or "").strip() + comment = f"{decision_text} by {actor}" + if note_text: + comment = f"{comment}\nReason: {note_text}" + commands = [ + f"--board {shlex.quote(self.board)} comment {self.task_id} {shlex.quote(comment)} --author {shlex.quote(actor)}", + ] + if decision == "approve": + commands.append( + f"--board {shlex.quote(self.board)} unblock --reason {shlex.quote(comment)} {self.task_id}" + ) + else: + remediation_title = f"Address rejection for {self.task_id}" + remediation_body = ( + f"{comment}\n\n" + f"Original task `{self.task_id}` was rejected by a human approval gate. " + "Revise the proposal or provide an alternative path before requesting approval again." + ) + commands.append( + f"--board {shlex.quote(self.board)} create {shlex.quote(remediation_title)} " + f"--body {shlex.quote(remediation_body)} " + f"--parent {self.task_id} --triage " + f"--idempotency-key {shlex.quote('kanban-approval-rejection:' + self.task_id)} " + f"--created-by {shlex.quote(actor)}" + ) + for command in commands: + result = await asyncio.to_thread(run_slash, command) + if result and ("usage error" in result.lower() or "invalid choice" in result.lower()): + raise RuntimeError(result[:500]) + logger.debug( + "Kanban approval gate command for task %s by %s: %s", + self.task_id, + actor, + result[:200] if result else "OK", + ) + logger.info( + "Kanban approval gate %s for task %s by %s", + decision.upper(), + self.task_id, + actor, + ) + except Exception as exc: + logger.error("Failed to resolve Kanban approval gate task: %s", exc) + + @discord.ui.button(label="āœ… Approve", style=discord.ButtonStyle.green) + async def approve( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve(interaction, "approve", discord.Color.green(), "āœ… Approved") + + @discord.ui.button(label="šŸ“ Approve + comment", style=discord.ButtonStyle.blurple) + async def approve_with_comment( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + if self.resolved: + await interaction.response.send_message( + "This approval gate has already been resolved~", ephemeral=True + ) + return + if not self._check_auth(interaction): + await interaction.response.send_message( + "You're not authorized to approve/reject this task~", ephemeral=True + ) + return + modal_cls = KanbanDecisionCommentModal + send_modal = getattr(interaction.response, "send_modal", None) + if modal_cls is None or not callable(send_modal): + await interaction.response.send_message( + "Comment modal is unavailable in this Discord runtime; use the plain Approve button or add a Kanban comment manually.", + ephemeral=True, + ) + return + await send_modal( + modal_cls(self, "approve", discord.Color.green(), "āœ… Approved") + ) + + @discord.ui.button(label="āŒ Reject", style=discord.ButtonStyle.red) + async def reject( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve(interaction, "reject", discord.Color.red(), "āŒ Rejected") + + @discord.ui.button(label="šŸ“ Reject + comment", style=discord.ButtonStyle.grey) + async def reject_with_comment( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + if self.resolved: + await interaction.response.send_message( + "This approval gate has already been resolved~", ephemeral=True + ) + return + if not self._check_auth(interaction): + await interaction.response.send_message( + "You're not authorized to approve/reject this task~", ephemeral=True + ) + return + modal_cls = KanbanDecisionCommentModal + send_modal = getattr(interaction.response, "send_modal", None) + if modal_cls is None or not callable(send_modal): + await interaction.response.send_message( + "Comment modal is unavailable in this Discord runtime; use the plain Reject button or add a Kanban comment manually.", + ephemeral=True, + ) + return + await send_modal( + modal_cls(self, "reject", discord.Color.red(), "āŒ Rejected") + ) + + async def on_timeout(self): + """Handle view timeout -- disable buttons and mark as expired.""" + self.resolved = True + for child in self.children: + child.disabled = True + # Visually update the Discord message so buttons appear disabled. + msg = getattr(self, '_message', None) + if msg: + try: + embed = msg.embeds[0] if msg.embeds else None + if embed: + embed.color = discord.Color.greyple() + embed.set_footer(text="ā± Approval gate expired ā€” no action taken") + await msg.edit(embed=embed, view=self) + except Exception: + pass # message deleted or too old to edit + + + class KanbanProtocolViolationView(discord.ui.View): + """Decision buttons for Kanban workers that violated lifecycle protocol.""" + + def __init__( + self, + task_id: str, + board: str, + channel_id: str, + allowed_user_ids: set, + allowed_role_ids: Optional[set] = None, + error: str = "", + ): + super().__init__(timeout=900) # 15 minutes: ops decisions take longer + self.task_id = task_id + self.board = board + self.channel_id = channel_id + self.allowed_user_ids = allowed_user_ids + self.allowed_role_ids = allowed_role_ids or set() + self.error = str(error or "").strip() + self.resolved = False + + def _check_auth(self, interaction: discord.Interaction) -> bool: + return _component_check_auth( + interaction, self.allowed_user_ids, self.allowed_role_ids + ) + + async def _resolve( + self, + interaction: discord.Interaction, + decision: str, + color: discord.Color, + label: str, + ): + if self.resolved: + await interaction.response.send_message( + "This protocol-violation decision has already been resolved~", + ephemeral=True, + ) + return + if not self._check_auth(interaction): + await interaction.response.send_message( + "You're not authorized to resolve this Kanban protocol violation~", + ephemeral=True, + ) + return + + self.resolved = True + embed = interaction.message.embeds[0] if interaction.message.embeds else None + if embed: + embed.color = color + embed.set_footer(text=f"{label} by {interaction.user.display_name}") + for child in self.children: + child.disabled = True + try: + await interaction.response.edit_message(embed=embed, view=self) + except Exception as exc: + logger.debug("Kanban protocol violation UI update failed: %s", exc) + + try: + from hermes_cli.kanban import run_slash + + actor = str(getattr(interaction.user, "display_name", "") or getattr(interaction.user, "id", "user")) + decision_text = f"PROTOCOL_VIOLATION_DECISION: {decision} by {actor}" + commands = [ + f"--board {shlex.quote(self.board)} comment {self.task_id} {shlex.quote(decision_text)} --author {shlex.quote(actor)}", + ] + if decision == "RETRY": + commands.append( + f"--board {shlex.quote(self.board)} promote --force {self.task_id} {shlex.quote(decision_text)}" + ) + elif decision == "REMEDIATE": + remediation_title = f"Remediate protocol violation for {self.task_id}" + remediation_body = ( + f"{decision_text}\n\n" + f"Original task `{self.task_id}` stopped because the worker exited without a terminal Kanban lifecycle call.\n\n" + f"Observed error:\n{self.error or 'Protocol violation: worker exited without kanban_complete or kanban_block.'}\n\n" + "Required next action: inspect worker output/artifacts, then either finish the task with kanban_complete or update the task/worker so it follows the Kanban protocol." + ) + commands.append( + f"--board {shlex.quote(self.board)} create {shlex.quote(remediation_title)} " + f"--body {shlex.quote(remediation_body)} " + f"--parent {self.task_id} --triage " + f"--idempotency-key {shlex.quote('kanban-protocol-violation:' + self.task_id)} " + f"--created-by {shlex.quote(actor)}" + ) + elif decision == "HUMAN_VERIFIED_DONE": + result = ( + f"Human verified completion after protocol violation. {decision_text}. " + "Use this only when artifacts/output have already been checked outside the worker." + ) + commands.append( + f"--board {shlex.quote(self.board)} complete {self.task_id} " + f"--result {shlex.quote(result)} --summary {shlex.quote(result)}" + ) + elif decision == "KEEP_BLOCKED": + # Durable side effect is the audit comment above. The task + # is already blocked by the gave_up/protocol-violation path. + pass + else: + raise RuntimeError(f"unknown protocol violation decision: {decision}") + + for command in commands: + result = await asyncio.to_thread(run_slash, command) + if result and ("usage error" in result.lower() or "invalid choice" in result.lower()): + raise RuntimeError(result[:500]) + logger.debug( + "Kanban protocol violation command for task %s by %s: %s", + self.task_id, + actor, + result[:200] if result else "OK", + ) + logger.info( + "Kanban protocol violation decision %s for task %s by %s", + decision, + self.task_id, + actor, + ) + except Exception as exc: + logger.error("Failed to resolve Kanban protocol violation decision: %s", exc) + + @discord.ui.button(label="šŸ” Retry worker", style=discord.ButtonStyle.blurple) + async def retry( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve(interaction, "RETRY", discord.Color.blue(), "šŸ” Retry") + + @discord.ui.button(label="āœ… Mark verified done", style=discord.ButtonStyle.green) + async def mark_verified_done( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve( + interaction, + "HUMAN_VERIFIED_DONE", + discord.Color.green(), + "āœ… Human verified done", + ) + + @discord.ui.button(label="šŸ›  Create remediation", style=discord.ButtonStyle.grey) + async def create_remediation( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve( + interaction, + "REMEDIATE", + discord.Color.orange(), + "šŸ›  Remediation created", + ) + + @discord.ui.button(label="ā›” Keep blocked", style=discord.ButtonStyle.red) + async def keep_blocked( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._resolve( + interaction, + "KEEP_BLOCKED", + discord.Color.red(), + "ā›” Kept blocked", + ) + + async def on_timeout(self): + self.resolved = True + for child in self.children: + child.disabled = True + msg = getattr(self, '_message', None) + if msg: + try: + embed = msg.embeds[0] if msg.embeds else None + if embed: + embed.color = discord.Color.greyple() + embed.set_footer(text="ā± Protocol violation decision expired ā€” no action taken") + await msg.edit(embed=embed, view=self) + except Exception: + pass + + class SlashConfirmView(discord.ui.View): """Three-button view for generic slash-command confirmations. diff --git a/tests/gateway/test_discord_kanban_approval.py b/tests/gateway/test_discord_kanban_approval.py new file mode 100644 index 0000000000000..9165288524c61 --- /dev/null +++ b/tests/gateway/test_discord_kanban_approval.py @@ -0,0 +1,198 @@ +"""Discord rich Kanban approval gate tests.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +# Trigger shared discord mock before importing adapter module. +from plugins.platforms.discord.adapter import ( # noqa: E402 + KanbanApprovalView, + KanbanProtocolViolationView, +) + + +def _interaction(user_id="458519787346198530", display_name="capt.america"): + embed = MagicMock() + embed.set_footer = MagicMock() + return SimpleNamespace( + user=SimpleNamespace(id=user_id, display_name=display_name, roles=[]), + response=SimpleNamespace(edit_message=AsyncMock(), send_message=AsyncMock()), + message=SimpleNamespace(embeds=[embed]), + ) + + +def test_kanban_approval_approve_comments_and_unblocks_not_completes(): + async def _run(): + view = KanbanApprovalView( + task_id="t_gate", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.approve(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert len(commands) == 2 + assert commands[0].startswith("--board default comment t_gate ") + assert "HUMAN_DECISION: APPROVED" in commands[0] + assert "capt.america" in commands[0] + assert commands[1] == "--board default unblock --reason 'HUMAN_DECISION: APPROVED by capt.america' t_gate" + assert not any("kanban complete" in cmd for cmd in commands) + interaction.response.edit_message.assert_awaited_once() + + asyncio.run(_run()) + + +def test_kanban_approval_approve_continues_when_discord_ack_expired(): + async def _run(): + view = KanbanApprovalView( + task_id="t_gate", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + interaction.response.edit_message.side_effect = RuntimeError("Unknown interaction") + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.approve(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert any(cmd.startswith("--board default comment t_gate ") for cmd in commands) + assert any(cmd.startswith("--board default unblock ") for cmd in commands) + + asyncio.run(_run()) + + +def test_kanban_approval_reject_comments_and_keeps_task_blocked(): + async def _run(): + view = KanbanApprovalView( + task_id="t_gate", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.reject(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert len(commands) == 2 + assert commands[0].startswith("--board default comment t_gate ") + assert "HUMAN_DECISION: REJECTED" in commands[0] + assert "capt.america" in commands[0] + assert commands[1].startswith("--board default create ") + assert "Address rejection for t_gate" in commands[1] + assert "--parent t_gate" in commands[1] + assert "--triage" in commands[1] + assert "--idempotency-key kanban-approval-rejection:t_gate" in commands[1] + assert "HUMAN_DECISION: REJECTED by capt.america" in commands[1] + assert not any(" unblock " in cmd for cmd in commands) + assert not any("kanban complete" in cmd for cmd in commands) + interaction.response.edit_message.assert_awaited_once() + + asyncio.run(_run()) + + +def test_kanban_approval_comment_reason_is_recorded(): + async def _run(): + view = KanbanApprovalView( + task_id="t_gate", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view._resolve( + interaction, + "reject", + MagicMock(), + "āŒ Rejected", + note="Need safer rollback plan first.", + ) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert "Reason: Need safer rollback plan first." in commands[0] + assert "Reason: Need safer rollback plan first." in commands[1] + + asyncio.run(_run()) + + +def test_protocol_violation_retry_comments_and_promotes(): + async def _run(): + view = KanbanProtocolViolationView( + task_id="t_proto", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.retry(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert len(commands) == 2 + assert commands[0].startswith("--board default comment t_proto ") + assert "PROTOCOL_VIOLATION_DECISION: RETRY" in commands[0] + assert "capt.america" in commands[0] + assert commands[1] == "--board default promote --force t_proto 'PROTOCOL_VIOLATION_DECISION: RETRY by capt.america'" + interaction.response.edit_message.assert_awaited_once() + + asyncio.run(_run()) + + +def test_protocol_violation_remediation_creates_idempotent_child(): + async def _run(): + view = KanbanProtocolViolationView( + task_id="t_proto", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + error="worker exited cleanly without terminal call", + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.create_remediation(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert len(commands) == 2 + assert "PROTOCOL_VIOLATION_DECISION: REMEDIATE" in commands[0] + assert commands[1].startswith("--board default create ") + assert "Remediate protocol violation for t_proto" in commands[1] + assert "--parent t_proto" in commands[1] + assert "--triage" in commands[1] + assert "--idempotency-key kanban-protocol-violation:t_proto" in commands[1] + assert "worker exited cleanly without terminal call" in commands[1] + assert not any(" promote " in cmd for cmd in commands) + + asyncio.run(_run()) + + +def test_protocol_violation_mark_verified_done_completes_with_human_summary(): + async def _run(): + view = KanbanProtocolViolationView( + task_id="t_proto", + board="default", + channel_id="1523615301965447298", + allowed_user_ids={"458519787346198530"}, + ) + interaction = _interaction() + + with patch("hermes_cli.kanban.run_slash", return_value="OK") as run_slash: + await view.mark_verified_done(interaction, MagicMock()) + + commands = [call.args[0] for call in run_slash.call_args_list] + assert len(commands) == 2 + assert "PROTOCOL_VIOLATION_DECISION: HUMAN_VERIFIED_DONE" in commands[0] + assert commands[1].startswith("--board default complete t_proto ") + assert "Human verified completion after protocol violation" in commands[1] + + asyncio.run(_run()) diff --git a/tests/hermes_cli/test_kanban_decompose_subscribe.py b/tests/hermes_cli/test_kanban_decompose_subscribe.py new file mode 100644 index 0000000000000..090bff06db526 --- /dev/null +++ b/tests/hermes_cli/test_kanban_decompose_subscribe.py @@ -0,0 +1,152 @@ +"""Kanban decomposition notification subscription inheritance tests.""" + +from __future__ import annotations + +import time + +from hermes_cli import kanban_db as kb + + +def _insert_triage_task(conn, task_id: str = "t_parent") -> str: + now = int(time.time()) + conn.execute( + """ + INSERT INTO tasks + (id, title, body, assignee, status, created_by, created_at, + workspace_kind, tenant) + VALUES (?, ?, ?, ?, 'triage', ?, ?, 'scratch', ?) + """, + (task_id, "Parent", "Root body", "profile-pmo", "tester", now, "demo"), + ) + conn.commit() + return task_id + + +def _subscribe(conn, task_id: str) -> None: + kb.add_notify_sub( + conn, + task_id=task_id, + platform="discord", + chat_id="1523615301965447298", + thread_id="1527382896757964971", + user_id="458519787346198530", + notifier_profile="profile-pmo", + ) + + +def test_decompose_copies_parent_notify_subscriptions_to_children(tmp_path): + conn = kb.connect(tmp_path / "kanban.db") + try: + parent_id = _insert_triage_task(conn) + _subscribe(conn, parent_id) + + child_ids = kb.decompose_triage_task( + conn, + parent_id, + root_assignee="profile-pmo", + children=[ + {"title": "Child one", "body": "one", "assignee": "profile-dev", "parents": []}, + {"title": "Child two", "body": "two", "assignee": "profile-qa", "parents": [0]}, + ], + author="profile-pmo", + auto_promote=False, + ) + + assert child_ids is not None + assert len(child_ids) == 2 + for child_id in child_ids: + rows = kb.list_notify_subs(conn, child_id) + assert rows == [ + { + "task_id": child_id, + "platform": "discord", + "chat_id": "1523615301965447298", + "thread_id": "1527382896757964971", + "user_id": "458519787346198530", + "notifier_profile": "profile-pmo", + "created_at": rows[0]["created_at"], + "last_event_id": 0, + } + ] + finally: + conn.close() + + +def test_decompose_respects_auto_subscribe_false(tmp_path): + conn = kb.connect(tmp_path / "kanban.db") + try: + parent_id = _insert_triage_task(conn) + _subscribe(conn, parent_id) + + child_ids = kb.decompose_triage_task( + conn, + parent_id, + root_assignee="profile-pmo", + children=[{"title": "Child", "body": "one", "assignee": "profile-dev", "parents": []}], + author="profile-pmo", + auto_promote=False, + auto_subscribe=False, + ) + + assert child_ids is not None + assert kb.list_notify_subs(conn, child_ids[0]) == [] + finally: + conn.close() + + +def test_copy_parent_subscriptions_to_children_is_idempotent(tmp_path): + conn = kb.connect(tmp_path / "kanban.db") + try: + parent_id = _insert_triage_task(conn) + _subscribe(conn, parent_id) + child_id = "t_child" + _insert_triage_task(conn, child_id) + + assert kb._copy_parent_subscriptions_to_children( + conn, parent_task_id=parent_id, child_ids=[child_id] + ) == 1 + assert kb._copy_parent_subscriptions_to_children( + conn, parent_task_id=parent_id, child_ids=[child_id] + ) == 0 + assert len(kb.list_notify_subs(conn, child_id)) == 1 + finally: + conn.close() + + +def test_create_task_with_parent_inherits_notify_subscription(tmp_path): + conn = kb.connect(tmp_path / "kanban.db") + try: + parent_id = _insert_triage_task(conn) + _subscribe(conn, parent_id) + + child_id = kb.create_task( + conn, + title="Child from worker", + body="Needs human gate", + assignee="profile-pmo", + created_by="profile-pmo", + parents=[parent_id], + ) + + rows = kb.list_notify_subs(conn, child_id) + assert len(rows) == 1 + assert rows[0]["platform"] == "discord" + assert rows[0]["chat_id"] == "1523615301965447298" + assert rows[0]["notifier_profile"] == "profile-pmo" + finally: + conn.close() + + +def test_link_tasks_copies_notify_subscription_once(tmp_path): + conn = kb.connect(tmp_path / "kanban.db") + try: + parent_id = _insert_triage_task(conn, "t_parent") + child_id = _insert_triage_task(conn, "t_linked_child") + _subscribe(conn, parent_id) + + kb.link_tasks(conn, parent_id, child_id) + kb.link_tasks(conn, parent_id, child_id) + + assert len(kb.list_notify_subs(conn, child_id)) == 1 + finally: + conn.close() diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index 07b1fa201bac1..2af9e713c2495 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -227,6 +227,287 @@ async def _fast_sleep(_): ) +def test_notifier_sends_rich_approval_for_needs_input_block(kanban_home): + """needs_input blocks use the adapter's rich approval surface when present.""" + + async def _run(): + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="needs rich approval", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="chat1", + notifier_profile="profile-pmo", + ) + kb.block_task(conn, tid, reason="needs approval", kind="needs_input") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._active_profile_name = lambda: "profile-pmo" + + fake_adapter = MagicMock() + + async def _rich_and_stop(*args, **kwargs): + runner._running = False + return SimpleNamespace(success=True, message_id="m1") + + fake_adapter.send_kanban_approval = AsyncMock(side_effect=_rich_and_stop) + fake_adapter.send = AsyncMock() + runner.adapters = {Platform.DISCORD: fake_adapter} + runner._profile_adapters = {} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send_kanban_approval.assert_awaited_once() + fake_adapter.send.assert_not_called() + _, kwargs = fake_adapter.send_kanban_approval.call_args + assert kwargs["chat_id"] == "chat1" + assert kwargs["task_id"] == tid + assert kwargs["board"] == "default" + assert kwargs["title"] == "needs rich approval" + assert "needs approval" in kwargs["summary"] + assert "why" in kwargs["approval_context"] + assert "Approve means" in kwargs["approval_context"]["approve_means"] + assert "Reject means" in kwargs["approval_context"]["reject_means"] + + asyncio.run(_run()) + + +def test_notifier_sends_protocol_violation_decision_card_for_gave_up(kanban_home): + """Protocol-violation give-up events use the dedicated decision card.""" + + async def _run(): + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="quiet worker", assignee="profile-developer") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="chat1", + thread_id="thread1", + notifier_profile="profile-pmo", + ) + kb._append_event( + conn, + tid, + kind="gave_up", + payload={ + "error": "worker exited cleanly without terminal Kanban call", + "protocol_violations": 3, + "protocol_violation_limit": 3, + "protocol_violation": True, + }, + ) + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._active_profile_name = lambda: "profile-pmo" + + fake_adapter = MagicMock() + + async def _decision_and_stop(*args, **kwargs): + runner._running = False + return SimpleNamespace(success=True, message_id="m_proto") + + fake_adapter.send_kanban_protocol_violation = AsyncMock(side_effect=_decision_and_stop) + fake_adapter.send = AsyncMock() + runner.adapters = {Platform.DISCORD: fake_adapter} + runner._profile_adapters = {} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send_kanban_protocol_violation.assert_awaited_once() + fake_adapter.send.assert_not_called() + _, kwargs = fake_adapter.send_kanban_protocol_violation.call_args + assert kwargs["chat_id"] == "chat1" + assert kwargs["task_id"] == tid + assert kwargs["board"] == "default" + assert kwargs["title"] == "quiet worker" + assert kwargs["assignee"] == "profile-developer" + assert "without terminal Kanban call" in kwargs["error"] + assert kwargs["violation_count"] == 3 + assert kwargs["violation_limit"] == 3 + assert kwargs["metadata"]["thread_id"] == "thread1" + + asyncio.run(_run()) + + +# NOTE: SimpleNamespace is imported later in this module for slash-command tests; +# keep this local import near the test that needs it to avoid changing older tests. +from types import SimpleNamespace + + +def test_notifier_standalone_named_profile_adapter_delivers(kanban_home): + """Standalone named-profile gateways resolve their primary adapter. + + Team-gondrong systemd services run like ``hermes --profile profile-pmo + gateway run``. In that shape the profile's Discord adapter is stored in + ``runner.adapters`` while notify rows are stamped with + ``notifier_profile='profile-pmo'``. Delivery must treat that as the active + profile instead of fail-closing as if it were an unavailable secondary + multiplex profile. + """ + + async def _run(): + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="standalone profile notify", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="chat1", + notifier_profile="profile-pmo", + ) + kb.block_task(conn, tid, reason="needs approval", kind="needs_input") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._active_profile_name = lambda: "profile-pmo" + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner.adapters = {Platform.DISCORD: fake_adapter} + runner._profile_adapters = {} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + assert "Human approval required" in fake_adapter.send.call_args[0][1] + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + assert len(subs) == 1 + assert int(subs[0]["last_event_id"]) > 0 + + asyncio.run(_run()) + + +def test_notifier_multiplex_profile_adapter_delivers(kanban_home): + """Notifier must see adapters stored under _profile_adapters. + + In multi-profile Discord gateways, a subscription row is stamped with + notifier_profile='profile-pmo' while the live PMO Discord adapter may live + only in runner._profile_adapters['profile-pmo'], not runner.adapters. The + notifier's pre-claim platform gate must include that registry or it skips + the row forever and last_event_id remains 0. + """ + + async def _run(): + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="multiplex notify", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="chat1", + notifier_profile="profile-pmo", + ) + kb.block_task(conn, tid, reason="needs approval", kind="needs_input") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner.adapters = {} + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner._profile_adapters = {"profile-pmo": {Platform.DISCORD: fake_adapter}} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + assert "Human approval required" in fake_adapter.send.call_args[0][1] + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + assert len(subs) == 1 + assert int(subs[0]["last_event_id"]) > 0 + + asyncio.run(_run()) + + # --------------------------------------------------------------------------- # Regression: gateway watchers must not double-init the kanban DB. #