feat: add Discord Kanban approval gates

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)

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",

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(

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}")