feat(kanban): auto-subscribe on create/assign, SIGUSR1 dispatcher wake, claimed event

gateway/run.py

@@ -3353,12 +3353,13 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None:
             return
 
         TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out")
+        NOTIFY_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "claimed")
         # Terminal event kinds trigger automatic unsubscription — the task
         # is done, blocked, or in a retry-needed state that the human
         # shouldn't keep pinging a stale chat for. Previously we only
         # unsubbed when task.status in ('done', 'archived'), which left
         # subscriptions on 'blocked' / 'gave_up' / 'crashed' / 'timed_out'
-        # tasks stranded forever.
+        # tasks stranded forever. Claimed is NOT terminal.
         TERMINAL_EVENT_KINDS = TERMINAL_KINDS
         # Per-subscription send-failure counter. Adapter.send raising
         # means the chat is dead (deleted, bot kicked, etc.) — after N
@@ -3403,7 +3404,7 @@ def _collect():
                                     platform=sub["platform"],
                                     chat_id=sub["chat_id"],
                                     thread_id=sub.get("thread_id") or "",
-                                    kinds=TERMINAL_KINDS,
+                                    kinds=NOTIFY_KINDS,
                                 )
                                 if not events:
                                     continue
@@ -3491,6 +3492,14 @@ def _collect():
                                 f"⏱ {tag}Kanban {sub['task_id']} timed out "
                                 f"(max_runtime={limit}s); will retry"
                             )
+                        elif kind == "claimed":
+                            who_claimed = ""
+                            if ev.payload and ev.payload.get("claimer"):
+                                who_claimed = f" by @{ev.payload['claimer']}"
+                            msg = (
+                                f"▶ {tag}Kanban {sub['task_id']} started{who_claimed}"
+                                f" — {title}"
+                            )
                         else:
                             continue
                         metadata: dict[str, Any] = {}
@@ -3737,6 +3746,27 @@ def _ready_nonempty() -> bool:
         logger.info(
             "kanban dispatcher: embedded in gateway (interval=%.1fs)", interval
         )
+
+        # SIGUSR1 = immediate dispatch wake. Helper scripts send this after
+        # creating a task so the worker spawns in <1s instead of waiting up
+        # to dispatch_interval_seconds for the next tick.
+        _dispatch_wake = asyncio.Event()
+
+        def _handle_sigusr1(*_):
+            """Schedule an event set on the running loop (signal-safe)."""
+            try:
+                loop = asyncio.get_running_loop()
+                loop.call_soon_threadsafe(_dispatch_wake.set)
+            except Exception:
+                pass
+
+        import signal as _signal
+        try:
+            loop = asyncio.get_running_loop()
+            loop.add_signal_handler(_signal.SIGUSR1, _handle_sigusr1)
+        except (NotImplementedError, OSError):
+            pass  # Windows or unsupported platform — degrade gracefully
+
         while self._running:
             try:
                 results = await asyncio.to_thread(_tick_once)
@@ -3780,12 +3810,17 @@ def _ready_nonempty() -> bool:
             except Exception:
                 logger.exception("kanban dispatcher: unexpected watcher error")
 
-            # Sleep in 1s slices so shutdown is snappy — otherwise a stop()
-            # waits up to `interval` seconds for the current sleep to finish.
-            slept = 0.0
-            while slept < interval and self._running:
-                await asyncio.sleep(min(1.0, interval - slept))
-                slept += 1.0
+            # Wait for the interval OR a SIGUSR1 wake — whichever comes first.
+            _dispatch_wake.clear()
+            try:
+                await asyncio.wait_for(
+                    _dispatch_wake.wait(),
+                    timeout=interval,
+                )
+            except asyncio.TimeoutError:
+                pass
+            if not self._running:
+                return
 
     async def _platform_reconnect_watcher(self) -> None:
         """Background task that periodically retries connecting failed platforms.

hermes_cli/kanban.py

@@ -149,6 +149,75 @@ def _check_dispatcher_presence() -> tuple[bool, str]:
     )
 
 
+# ---------------------------------------------------------------------------
+# Notify-routes auto-subscribe helper
+# ---------------------------------------------------------------------------
+
+def _load_notify_routes() -> dict:
+    """Load profile→platform routes from ~/.hermes/kanban/notify-routes.yaml.
+
+    Returns empty dict on any error (missing file, bad YAML, import failure).
+    Schema:
+        routes:
+          <profile>:
+            platform: telegram
+            chat_id: "<id>"
+            thread_id: null   # optional
+    """
+    import os
+    routes_path = Path(os.path.expanduser("~/.hermes/kanban/notify-routes.yaml"))
+    if not routes_path.exists():
+        return {}
+    try:
+        import yaml  # type: ignore[import]
+        data = yaml.safe_load(routes_path.read_text()) or {}
+    except Exception:
+        return {}
+    return data.get("routes", {}) or {}
+
+
+def _auto_subscribe_from_routes(
+    task_id: str,
+    assignee: Optional[str],
+    board: Optional[str] = None,
+) -> None:
+    """Auto-subscribe assignee's notify route (if any) to task_id.
+
+    Silently skips when:
+    - assignee is None
+    - no route for assignee in notify-routes.yaml
+    - route is missing platform or chat_id
+    Logs a warning on subscribe failure but does NOT raise.
+    """
+    if not assignee:
+        return
+    routes = _load_notify_routes()
+    route = routes.get(assignee)
+    if not route:
+        return
+    platform = route.get("platform")
+    chat_id = route.get("chat_id")
+    thread_id = route.get("thread_id") or None
+    if not platform or not chat_id:
+        print(
+            f"kanban: warn: route for '{assignee}' missing platform/chat_id; skipping auto-subscribe",
+            file=sys.stderr,
+        )
+        return
+    try:
+        with kb.connect(board=board) as conn:
+            kb.add_notify_sub(
+                conn,
+                task_id=task_id,
+                platform=platform,
+                chat_id=str(chat_id),
+                thread_id=str(thread_id) if thread_id else None,
+                user_id=None,
+            )
+    except Exception as exc:
+        print(f"kanban: warn: auto-subscribe failed for {task_id}: {exc}", file=sys.stderr)
+
+
 # ---------------------------------------------------------------------------
 # Argparse builder
 # ---------------------------------------------------------------------------
@@ -1000,6 +1069,8 @@ def _cmd_create(args: argparse.Namespace) -> int:
             skills=getattr(args, "skills", None) or None,
         )
         task = kb.get_task(conn, task_id)
+    # Auto-subscribe assignee's notify route (if any).
+    _auto_subscribe_from_routes(task_id, args.assignee, board=getattr(args, "board", None))
     if getattr(args, "json", False):
         print(json.dumps(_task_to_dict(task), indent=2, ensure_ascii=False))
     else:
@@ -1212,6 +1283,8 @@ def _cmd_assign(args: argparse.Namespace) -> int:
         print(f"no such task: {args.task_id}", file=sys.stderr)
         return 1
     print(f"Assigned {args.task_id} to {profile or '(unassigned)'}")
+    # Auto-subscribe new assignee's notify route (if any).
+    _auto_subscribe_from_routes(args.task_id, profile, board=getattr(args, "board", None))
     return 0
 
 

tests/hermes_cli/test_kanban_notify_routes.py

@@ -0,0 +1,68 @@
+"""Tests for CLI auto-subscribe via notify-routes.yaml (P4 from EVE audit)."""
+from __future__ import annotations
+
+import sqlite3
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from hermes_cli.kanban import _load_notify_routes, _auto_subscribe_from_routes
+from hermes_cli import kanban_db as kb
+
+
+@pytest.fixture
+def routes_yaml(tmp_path, monkeypatch):
+    """Write a notify-routes.yaml and point the helper at it."""
+    monkeypatch.setenv("HOME", str(tmp_path))
+    routes_dir = tmp_path / ".hermes" / "kanban"
+    routes_dir.mkdir(parents=True)
+    yaml_file = routes_dir / "notify-routes.yaml"
+    return yaml_file
+
+
+def test_load_routes_missing_file(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    assert _load_notify_routes() == {}
+
+
+def test_load_routes_valid(routes_yaml):
+    routes_yaml.write_text(
+        "routes:\n"
+        "  default:\n"
+        "    platform: telegram\n"
+        "    chat_id: '12345'\n"
+        "    thread_id: null\n"
+    )
+    routes = _load_notify_routes()
+    assert routes["default"]["platform"] == "telegram"
+    assert routes["default"]["chat_id"] == "12345"
+
+
+def test_missing_route_no_subscribe(routes_yaml):
+    """Missing route = no subscribe attempt (not an error)."""
+    routes_yaml.write_text("routes:\n  other:\n    platform: telegram\n    chat_id: '999'\n")
+    called = []
+    with patch("hermes_cli.kanban.kb.add_notify_sub", side_effect=lambda *a, **k: called.append(1)):
+        _auto_subscribe_from_routes("t_test123", "default")
+    assert called == []
+
+
+def test_bad_route_missing_chat_id_no_crash(routes_yaml):
+    """Bad route (missing chat_id) = warning, not crash."""
+    routes_yaml.write_text(
+        "routes:\n  default:\n    platform: telegram\n"
+    )
+    # Should not raise
+    _auto_subscribe_from_routes("t_test123", "default")
+
+
+def test_subscribe_failure_no_raise(routes_yaml):
+    """Subscribe failure = warning logged, no exception propagated."""
+    routes_yaml.write_text(
+        "routes:\n  default:\n    platform: telegram\n    chat_id: '12345'\n"
+    )
+    with patch("hermes_cli.kanban.kb.add_notify_sub", side_effect=Exception("db error")):
+        # Should not raise
+        _auto_subscribe_from_routes("t_test123", "default")