feat(kanban): auto-subscribe gateway chat on tool-driven kanban_create (#19718)

Closes #19479.

When an orchestrator agent calls kanban_create from a gateway session
(e.g. a Telegram user delegating to an orchestrator profile), auto-
subscribe the originating (platform, chat, thread, user) to the new
task's terminal events. Mirrors the behavior of the /kanban create
slash command in gateway/run.py so tool-driven creation is at parity
with human-driven creation.

Without this, a user who interacts with an orchestrator exclusively
via the gateway never receives blocked / completed / gave_up
notifications for tasks the orchestrator created on their behalf —
silently breaking the gateway-first multi-agent flow the reporter
describes.

Reads the context-local HERMES_SESSION_* vars via get_session_env()
(not os.environ — those are contextvars for asyncio concurrency
safety). Falls through cleanly in CLI / cron contexts with no
session active (subscribed=False in the response). Best-effort: if
the gateway module isn't importable (test rigs stubbing gateway.*),
the task still creates, we just skip the subscription.

Response gains a 'subscribed' bool so the orchestrator knows whether
terminal events will land back in the originating chat or whether it
needs to poll / unblock manually.

Tests: 4 new in tests/tools/test_kanban_tools.py covering
CLI/no-subscribe, telegram/gateway-auto-subscribe, discord-DM/no-
thread subscribe, and partial-ctx/no-chat_id no-subscribe. 40/40
kanban tool tests pass.

tests/tools/test_kanban_tools.py

@@ -610,3 +610,103 @@ def test_orchestrator_complete_any_task_allowed(monkeypatch, tmp_path):
     out = kt._handle_complete({"task_id": tid, "summary": "orchestrator close"})
     d = json.loads(out)
     assert d.get("ok") is True and d.get("task_id") == tid
+
+
+# ---------------------------------------------------------------------------
+# kanban_create auto-subscribe to gateway notifications (#19479)
+# ---------------------------------------------------------------------------
+#
+# When an orchestrator agent (running under the gateway) calls kanban_create,
+# the originating (platform, chat, thread) should be auto-subscribed to the
+# new task's terminal events — matching the /kanban create slash-command
+# behavior in gateway/run.py. In CLI / cron contexts (no session vars set),
+# no subscription row is written.
+
+
+def test_create_no_subscribe_in_cli_context(worker_env):
+    """Classic CLI: no gateway session vars -> no notify subscription."""
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+    out = kt._handle_create({"title": "cli task", "assignee": "peer"})
+    d = json.loads(out)
+    assert d.get("ok") is True
+    assert d.get("subscribed") is False
+    conn = kb.connect()
+    try:
+        assert kb.list_notify_subs(conn, d["task_id"]) == []
+    finally:
+        conn.close()
+
+
+def test_create_auto_subscribes_in_gateway_context(worker_env):
+    """Gateway session vars set -> auto-subscribe the originating source."""
+    from gateway.session_context import set_session_vars, clear_session_vars
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+
+    tokens = set_session_vars(
+        platform="telegram",
+        chat_id="1234567",
+        thread_id="42",
+        user_id="u_alice",
+    )
+    try:
+        out = kt._handle_create({"title": "gateway task", "assignee": "peer"})
+        d = json.loads(out)
+        assert d.get("ok") is True
+        assert d.get("subscribed") is True
+        conn = kb.connect()
+        try:
+            subs = kb.list_notify_subs(conn, d["task_id"])
+        finally:
+            conn.close()
+        assert len(subs) == 1
+        assert subs[0]["platform"] == "telegram"
+        assert subs[0]["chat_id"] == "1234567"
+        assert subs[0]["thread_id"] == "42"
+        assert subs[0]["user_id"] == "u_alice"
+    finally:
+        clear_session_vars(tokens)
+
+
+def test_create_subscribe_without_thread_id(worker_env):
+    """DM / no-thread platforms subscribe without a thread_id."""
+    from gateway.session_context import set_session_vars, clear_session_vars
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+
+    tokens = set_session_vars(platform="discord", chat_id="ch_dm_789")
+    try:
+        out = kt._handle_create({"title": "dm task", "assignee": "peer"})
+        d = json.loads(out)
+        assert d.get("subscribed") is True
+        conn = kb.connect()
+        try:
+            subs = kb.list_notify_subs(conn, d["task_id"])
+        finally:
+            conn.close()
+        assert len(subs) == 1
+        assert subs[0]["thread_id"] == ""
+        assert subs[0]["user_id"] is None
+    finally:
+        clear_session_vars(tokens)
+
+
+def test_create_no_subscribe_when_chat_id_missing(worker_env):
+    """Partial gateway context (platform but no chat_id) -> no subscription."""
+    from gateway.session_context import set_session_vars, clear_session_vars
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+
+    tokens = set_session_vars(platform="telegram", chat_id="")
+    try:
+        out = kt._handle_create({"title": "partial ctx", "assignee": "peer"})
+        d = json.loads(out)
+        assert d.get("subscribed") is False
+        conn = kb.connect()
+        try:
+            assert kb.list_notify_subs(conn, d["task_id"]) == []
+        finally:
+            conn.close()
+    finally:
+        clear_session_vars(tokens)