diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 1a34feb12b9f..b05217a6e963 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -25,6 +25,33 @@ logger = logging.getLogger("gateway.run") +def _resolve_notification_delivery_mode() -> str: + """Return the configured Kanban notification delivery mode.""" + from hermes_cli.config import load_config + + try: + config = load_config() + kanban = config.get("kanban", {}) if isinstance(config, dict) else {} + value = kanban.get("notification_delivery_mode", "text_and_agent") + except Exception as exc: + logger.warning( + "Could not read kanban.notification_delivery_mode (%s); using " + "text_and_agent", + exc, + ) + return "text_and_agent" + if isinstance(value, str): + mode = value.strip().lower() + if mode in {"text_and_agent", "agent_only"}: + return mode + logger.warning( + "Invalid kanban.notification_delivery_mode=%r; expected " + "text_and_agent or agent_only; using text_and_agent", + value, + ) + return "text_and_agent" + + def _resolve_auto_decompose_settings( load_config: Callable[[], Any], ) -> "tuple[bool, int]": @@ -202,6 +229,7 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: # A genuinely dead chat still drops, just ~60s later — a fine trade # for an unattended gate where a false drop means silent work pileup. MAX_SEND_FAILURES = 12 + notification_delivery_mode = _resolve_notification_delivery_mode() sub_fail_counts: dict[tuple, int] = getattr( self, "_kanban_sub_fail_counts", {} ) @@ -416,6 +444,9 @@ def _collect(): board_slug, ) continue + from gateway.wake import adapter_supports_push + + _is_push_adapter = adapter_supports_push(adapter) title = (task.title if task else sub["task_id"])[:120] board_tag = f"[{board_slug}] " if board_slug else "" # Per-subscription failure-counter key. Hoisted out of the @@ -545,9 +576,7 @@ def _collect(): # non-push adapters, skip the doomed send attempt # entirely: there is nothing to text-notify, the # creator is woken via the self-post below instead. - from gateway.wake import adapter_supports_push - - if not adapter_supports_push(adapter): + if not _is_push_adapter: logger.debug( "kanban notifier: adapter %s has no push " "channel; skipping text ping for %s, relying " @@ -559,6 +588,10 @@ def _collect(): # so the counter is resolved (reset or bumped) by # the self-post outcome, not by skipping the send. continue + if notification_delivery_mode == "agent_only": + # The wake below is the delivery in this mode. Do + # not send native text or upload native artifacts. + continue try: _send_res = await adapter.send( sub["chat_id"], msg, metadata=metadata, @@ -634,33 +667,28 @@ def _collect(): # dropping the subscription is the terminal action. break else: - # All text pings delivered (or intentionally skipped - # for non-push adapters, whose delivery is the wake - # self-post below). Whether the cursor may advance now - # depends on the adapter class: + # All text pings delivered or intentionally skipped. + # Whether the cursor may advance now depends on the + # delivery path: # - # * push-capable: the text send WAS the delivery, so - # advance immediately (pre-existing behavior); the - # wake injection below stays best-effort. - # * non-push (api_server): the wake self-post IS the - # delivery. Advancing first would let a failed / - # retry-exhausted self-post (swallowed by the - # best-effort except) permanently lose the event. - # So the self-post runs FIRST and the cursor only - # advances after it succeeds — a failure rewinds the - # claim exactly like a failed send() above, so the - # next tick retries. + # * text_and_agent push: native text was the delivery; + # advance before the best-effort wake. + # * non-push or agent_only push: wake is the delivery; + # run it first and rewind the claim on failure. task_terminal = task and task.status in {"done", "archived"} - _WAKE_KINDS = ("completed", "gave_up", "crashed", "timed_out", "blocked") + _WAKE_KINDS = { + "completed", "gave_up", "crashed", "timed_out", "blocked", + } + if notification_delivery_mode == "agent_only" and _is_push_adapter: + _WAKE_KINDS.update({ + "status", "review_requested", "block_loop_detected", + }) _wake_kinds = {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} - from gateway.wake import adapter_supports_push as _adapter_push_ok - - _is_push_adapter = _adapter_push_ok(adapter) _session_key = "" _synth = "" if _wake_kinds: _session_key = getattr(task, "session_id", None) or "" - if _wake_kinds and _session_key: + if _wake_kinds and (_session_key or _is_push_adapter): _title = (task.title if task else sub["task_id"])[:120] _assignee = task.assignee if task else "" _parts = [] @@ -679,9 +707,37 @@ def _collect(): board=board_slug, ) - if not _is_push_adapter and _wake_kinds and _session_key: - # Wake self-post IS the delivery on this path — - # it must succeed BEFORE the cursor advances. + _source = None + if _is_push_adapter and _wake_kinds: + from gateway.session import SessionSource + + _chat_type = str(sub.get("chat_type") or "").strip() + if not _chat_type: + _delivery_meta = sub.get("delivery_metadata") + if isinstance(_delivery_meta, dict): + _chat_type = str( + _delivery_meta.get("chat_type") or "" + ).strip() + _source = SessionSource( + platform=plat, + chat_id=sub["chat_id"], + chat_type=_chat_type or "group", + thread_id=sub.get("thread_id") or None, + user_id=sub.get("user_id"), + profile=sub_profile or None, + ) + + _wake_is_delivery = ( + not _is_push_adapter + or notification_delivery_mode == "agent_only" + ) + _can_wake = bool( + _wake_kinds + and (_session_key or _is_push_adapter) + ) + if _wake_is_delivery and _can_wake: + # Wake IS the delivery on this path, so it must + # succeed before the claimed cursor advances. from gateway.wake import deliver_wake try: @@ -689,6 +745,7 @@ def _collect(): adapter, text=_synth, session_id=_session_key, + source=_source, ) logger.info( "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", @@ -699,7 +756,7 @@ def _collect(): fails = sub_fail_counts.get(sub_key, 0) + 1 sub_fail_counts[sub_key] = fails logger.warning( - "kanban notifier: wake self-post failed " + "kanban notifier: wake delivery failed " "for %s (attempt %d/%d): %s", sub["task_id"], fails, MAX_SEND_FAILURES, _wk_err, exc_info=True, @@ -713,8 +770,8 @@ def _collect(): await asyncio.to_thread(self._kanban_unsub, sub, board_slug) sub_fail_counts.pop(sub_key, None) else: - # Rewind the pre-send claim so the next - # tick retries the self-post — the event + # Rewind the claim so the next tick retries + # the wake delivery — the event # is NOT lost. await asyncio.to_thread( self._kanban_rewind, @@ -725,10 +782,8 @@ def _collect(): ) continue - # Delivery complete (text ping for push adapters, wake - # self-post for non-push): advance cursor. The cursor - # is the dedup mechanism — it prevents re-delivery - # of the same event on subsequent ticks. + # Delivery complete: advance the dedup cursor so this + # event is not delivered again on subsequent ticks. await asyncio.to_thread( self._kanban_advance, sub, d["cursor"], board_slug, ) @@ -743,40 +798,14 @@ def _collect(): # dispatcher respawns the task and it cycles into the # same state. See the longer comment on TERMINAL_KINDS # above for the failure mode this prevents. - if _is_push_adapter and _wake_kinds and _session_key: + if ( + _is_push_adapter + and notification_delivery_mode == "text_and_agent" + and _wake_kinds + and _session_key + ): try: - from gateway.session import SessionSource from gateway.wake import deliver_wake - # Rebuild the creator's real session scope from - # the chat_type persisted on the subscription - # row (#56580). build_session_key() keys DMs - # (":dm:") on a wholly different shape - # from group/thread, so the old hardcoded - # "group" mis-routed DM/thread creators into a - # fresh session. Legacy rows written before the - # column existed may still carry chat_type in - # delivery_metadata (#60600 rows) — fall back - # to that, then to "group" (the historical - # default that suits the dashboard/group flows). - # handle_message() get_or_create_session's the - # target, so a mismatch only ever degrades to a - # fresh session, never an exception. - _chat_type = str(sub.get("chat_type") or "").strip() - if not _chat_type: - _delivery_meta = sub.get("delivery_metadata") - if isinstance(_delivery_meta, dict): - _chat_type = str( - _delivery_meta.get("chat_type") or "" - ).strip() - _chat_type = _chat_type or "group" - _source = SessionSource( - platform=plat, - chat_id=sub["chat_id"], - chat_type=_chat_type, - thread_id=sub.get("thread_id") or None, - user_id=sub.get("user_id"), - profile=sub_profile or None, - ) # deliver_wake preserves the synthetic # MessageEvent/handle_message path for # push-capable adapters (the non-push / diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index f7ee770719cf..ffaa6c161582 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2340,6 +2340,10 @@ # behaviour — e.g. for a profile that prefers explicit # ``kanban_notify-subscribe`` calls per task. "auto_subscribe_on_create": True, + # Push adapters normally send native Kanban text/artifacts and then + # wake the originating agent. Set to agent_only to make the wake turn + # the sole delivery and let the agent respond in the originating chat. + "notification_delivery_mode": "text_and_agent", # Run the dispatcher inside the gateway process. On by default — # the cost is ~300µs every `dispatch_interval_seconds` when idle, # and gateway is the supervisor users already have. Set to false diff --git a/tests/gateway/test_kanban_notifier_delivery_mode.py b/tests/gateway/test_kanban_notifier_delivery_mode.py new file mode 100644 index 000000000000..72171b45f625 --- /dev/null +++ b/tests/gateway/test_kanban_notifier_delivery_mode.py @@ -0,0 +1,331 @@ +import asyncio + +import pytest + +from gateway.config import Platform +from gateway.run import GatewayRunner +from hermes_cli import kanban_db as kb + + +class PushAdapter: + def __init__(self, wake_failures=0): + self.sent = [] + self.handled = [] + self.wake_failures = wake_failures + + async def send(self, chat_id, text, metadata=None): + self.sent.append((chat_id, text, metadata)) + + async def handle_message(self, event): + self.handled.append(event) + if self.wake_failures: + self.wake_failures -= 1 + raise RuntimeError("wake failed") + + +class NonPushAdapter: + supports_async_delivery = False + + def __init__(self): + self.send_calls = 0 + + async def send(self, chat_id, text, metadata=None): + self.send_calls += 1 + + +async def _run_tick(monkeypatch, runner): + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + if delay == 5: + return + runner._running = False + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + await runner._kanban_notifier_watcher(interval=1) + + +def _runner(adapter, platform=Platform.TELEGRAM): + runner = GatewayRunner.__new__(GatewayRunner) + runner._running = True + runner.adapters = {platform: adapter} + runner._profile_adapters = {} + runner._kanban_sub_fail_counts = {} + runner._kanban_dispatcher_lock_handle = object() + return runner + + +def _completed_sub( + *, platform="telegram", chat_id="chat-1", thread_id="", profile="", chat_type="group" +): + conn = kb.connect() + try: + task_id = kb.create_task( + conn, + title="delivery mode", + assignee="worker", + session_id="origin-session", + ) + kb.add_notify_sub( + conn, + task_id=task_id, + platform=platform, + chat_id=chat_id, + thread_id=thread_id, + notifier_profile=profile, + chat_type=chat_type, + ) + kb.complete_task(conn, task_id, summary="done") + return task_id + finally: + conn.close() + + +def _unblocked_sub(): + conn = kb.connect() + try: + task_id = kb.create_task(conn, title="silent delivery", assignee="worker") + assert kb.block_task(conn, task_id, reason="waiting") + kb.add_notify_sub( + conn, + task_id=task_id, + platform="telegram", + chat_id="chat-1", + ) + before = kb.list_notify_subs(conn, task_id)[0]["last_event_id"] + assert kb.unblock_task(conn, task_id) + return task_id, before + finally: + conn.close() + + +def _event_sub(kind, *, platform="telegram", chat_id="chat-1"): + conn = kb.connect() + try: + task_id = kb.create_task( + conn, + title="agent-only event", + assignee="worker", + session_id="origin-session", + ) + kb.add_notify_sub( + conn, + task_id=task_id, + platform=platform, + chat_id=chat_id, + ) + before = kb.list_notify_subs(conn, task_id)[0]["last_event_id"] + payload = { + "status": {"status": "running"}, + "review_requested": {"summary": "ready"}, + "block_loop_detected": {"reason": "needs input", "recurrences": 2}, + }[kind] + kb._append_event(conn, task_id, kind, payload) + return task_id, before + finally: + conn.close() + + +def _configure(monkeypatch, value=...): + import hermes_cli.config as config + + kanban = {} if value is ... else {"notification_delivery_mode": value} + monkeypatch.setattr(config, "load_config", lambda: {"kanban": kanban}) + + +def _subs(task_id): + conn = kb.connect() + try: + return kb.list_notify_subs(conn, task_id) + finally: + conn.close() + + +@pytest.mark.parametrize("configured", [..., "text_and_agent", "unknown", None]) +def test_default_and_invalid_modes_send_native_text(tmp_path, monkeypatch, configured): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, configured) + _completed_sub() + adapter = PushAdapter() + + asyncio.run(_run_tick(monkeypatch, _runner(adapter))) + + assert len(adapter.sent) == 1 + assert len(adapter.handled) == 1 + + +def test_agent_only_uses_wake_as_delivery_and_skips_native_artifacts(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + task_id = _completed_sub() + adapter = PushAdapter() + runner = _runner(adapter) + artifact_calls = [] + + async def record_artifacts(**kwargs): + artifact_calls.append(kwargs) + + runner._deliver_kanban_artifacts = record_artifacts + asyncio.run(_run_tick(monkeypatch, runner)) + + assert adapter.sent == [] + assert len(adapter.handled) == 1 + assert artifact_calls == [] + assert _subs(task_id) == [] + + +def test_agent_only_failed_wake_rewinds_and_retries_without_loss(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + task_id = _completed_sub() + adapter = PushAdapter(wake_failures=1) + runner = _runner(adapter) + + asyncio.run(_run_tick(monkeypatch, runner)) + assert adapter.sent == [] + assert len(adapter.handled) == 1 + assert len(_subs(task_id)) == 1 + + runner._running = True + asyncio.run(_run_tick(monkeypatch, runner)) + assert adapter.sent == [] + assert len(adapter.handled) == 2 + assert _subs(task_id) == [] + + +def test_agent_only_drops_subscription_after_twelfth_failed_wake(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + task_id = _completed_sub() + adapter = PushAdapter(wake_failures=12) + runner = _runner(adapter) + + for _ in range(11): + asyncio.run(_run_tick(monkeypatch, runner)) + runner._running = True + + assert len(adapter.handled) == 11 + assert len(_subs(task_id)) == 1 + + asyncio.run(_run_tick(monkeypatch, runner)) + + assert adapter.sent == [] + assert len(adapter.handled) == 12 + assert _subs(task_id) == [] + + +def test_agent_only_preserves_discord_group_thread_profile_source(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + _completed_sub( + platform="discord", + chat_id="guild-channel", + thread_id="thread-42", + profile="writer", + chat_type="group", + ) + adapter = PushAdapter() + runner = _runner(PushAdapter()) + runner._profile_adapters = {"writer": {Platform.DISCORD: adapter}} + runner._active_profile_name = lambda: "default" + + asyncio.run(_run_tick(monkeypatch, runner)) + + assert adapter.sent == [] + assert len(adapter.handled) == 1 + source = adapter.handled[0].source + assert source.platform is Platform.DISCORD + assert source.chat_id == "guild-channel" + assert source.chat_type == "group" + assert source.thread_id == "thread-42" + assert source.profile == "writer" + + +def test_agent_only_does_not_change_non_push_delivery(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + _completed_sub(platform="api_server", chat_id="origin-session") + adapter = NonPushAdapter() + runner = _runner(adapter, Platform.API_SERVER) + wakes = [] + + async def record_wake(adapter, *, text, session_id="", source=None): + wakes.append((session_id, source)) + + import gateway.wake as wake + + monkeypatch.setattr(wake, "deliver_wake", record_wake) + asyncio.run(_run_tick(monkeypatch, runner)) + + assert adapter.send_calls == 0 + assert wakes == [("origin-session", None)] + + +def test_agent_only_non_push_status_advances_without_wake(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + task_id, before = _event_sub( + "status", platform="api_server", chat_id="origin-session", + ) + adapter = NonPushAdapter() + wakes = [] + + async def record_wake(adapter, *, text, session_id="", source=None): + wakes.append((session_id, source)) + + import gateway.wake as wake + + monkeypatch.setattr(wake, "deliver_wake", record_wake) + asyncio.run(_run_tick(monkeypatch, _runner(adapter, Platform.API_SERVER))) + + assert adapter.send_calls == 0 + assert wakes == [] + assert _subs(task_id)[0]["last_event_id"] > before + + +@pytest.mark.parametrize( + "kind", ["status", "review_requested", "block_loop_detected"], +) +def test_agent_only_push_wakes_for_agent_delivery_events( + tmp_path, monkeypatch, kind, +): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, "agent_only") + task_id, before = _event_sub(kind) + adapter = PushAdapter() + + asyncio.run(_run_tick(monkeypatch, _runner(adapter))) + + assert adapter.sent == [] + assert len(adapter.handled) == 1 + assert _subs(task_id)[0]["last_event_id"] > before + + +@pytest.mark.parametrize("mode", ["text_and_agent", "agent_only"]) +def test_silent_claim_advances_without_send_or_wake( + tmp_path, monkeypatch, caplog, mode, +): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "kanban.db")) + kb.init_db() + _configure(monkeypatch, mode) + task_id, before = _unblocked_sub() + adapter = PushAdapter() + + asyncio.run(_run_tick(monkeypatch, _runner(adapter))) + + assert _subs(task_id)[0]["last_event_id"] > before + assert adapter.sent == [] + assert adapter.handled == [] + assert not any( + "kanban notifier tick failed" in record.getMessage() + for record in caplog.records + ) diff --git a/tests/hermes_cli/test_config.py b/tests/hermes_cli/test_config.py index c2db04cd9661..f2e4f565e7ab 100644 --- a/tests/hermes_cli/test_config.py +++ b/tests/hermes_cli/test_config.py @@ -1422,6 +1422,7 @@ def test_default_config_kanban_block_not_dropped_by_duplicate_key(): kanban = DEFAULT_CONFIG["kanban"] # From the first (dropped) block: assert kanban.get("auto_subscribe_on_create") is True + assert kanban.get("notification_delivery_mode") == "text_and_agent" # From the second block: assert "dispatch_in_gateway" in kanban assert "auto_decompose" in kanban diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 7ee2942d2b2f..ac4b3b006628 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -612,6 +612,7 @@ Config knobs (all under `kanban:` in `~/.hermes/config.yaml`): | `orchestrator_profile` | `""` | Profile assigned to the root/orchestration task after decomposition. Empty = fall back to active default profile. | | `default_assignee` | `""` | Where a child task lands when the LLM picks an unknown profile. Empty = fall back to active default. | | `auto_subscribe_on_create` | `true` | When a worker calls `kanban_create` from inside a session with a persistent delivery channel (messaging gateway or TUI), the originating session is auto-subscribed to the new task's completion/block events. The dispatcher still drives the delivery — this only changes whether the caller's chat/key shows up in the notify-sub table. Set to `false` to require explicit `kanban_notify-subscribe` calls per task. | +| `notification_delivery_mode` | `text_and_agent` | `text_and_agent` sends native Kanban text/artifacts and wakes the originating agent. `agent_only` skips native delivery and uses the agent wake response as the delivery. Invalid values fall back to `text_and_agent`. | And the two auxiliary LLM slots: