Gateway background process watcher notification modes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #840 feat(gateway): background process notification modes + fix spinner line spam

  merged to main

  files: AGENTS.md, agent/display.py, cli-config.yaml.example, gateway/run.py, tests/gateway/test_background_process_notifications.py

  Show the diff

  • AGENTS.md

    @@ -300,6 +300,17 @@ Cache-breaking forces dramatically higher costs. The ONLY time we alter context
     - **CLI**: Uses current directory (`.` → `os.getcwd()`)
     - **Messaging**: Uses `MESSAGING_CWD` env var (default: home directory)
     
    +### Background Process Notifications (Gateway)
    +
    +When `terminal(background=true, check_interval=...)` is used, the gateway runs a watcher that
    +pushes status updates to the user's chat. Control verbosity with `display.background_process_notifications`
    +in config.yaml (or `HERMES_BACKGROUND_NOTIFICATIONS` env var):
    +
    +- `all` — running-output updates + final message (default)
    +- `result` — only the final completion message
    +- `error` — only the final message when exit code != 0
    +- `off` — no watcher messages at all
    +
     ---
     
     ## Known Pitfalls

  • agent/display.py

    @@ -206,6 +206,7 @@ def __init__(self, message: str = "", spinner_type: str = 'dots'):
             self.frame_idx = 0
             self.start_time = None
             self.last_line_len = 0
    +        self._last_flush_time = 0.0  # Rate-limit flushes for patch_stdout compat
             # Capture stdout NOW, before any redirect_stdout(devnull) from
             # child agents can replace sys.stdout with a black hole.
             self._out = sys.stdout
    @@ -236,7 +237,18 @@ def _animate(self):
                 else:
                     line = f"  {frame} {self.message} ({elapsed:.1f}s)"
                 pad = max(self.last_line_len - len(line), 0)
    -            self._write(f"\r{line}{' ' * pad}", end='', flush=True)
    +            # Rate-limit flush() calls to avoid spinner spam under
    +            # prompt_toolkit's patch_stdout.  Each flush() pushes a queue
    +            # item that may trigger a separate run_in_terminal() call; if
    +            # items are processed one-at-a-time the \r overwrite is lost
    +            # and every frame appears on its own line.  By flushing at
    +            # most every 0.4s we guarantee multiple \r-frames are batched
    +            # into a single write, so the terminal collapses them correctly.
    +            now = time.time()
    +            should_flush = (now - self._last_flush_time) >= 0.4
    +            self._write(f"\r{line}{' ' * pad}", end='', flush=should_flush)
    +            if should_flush:
    +                self._last_flush_time = now
                 self.last_line_len = len(line)
                 self.frame_idx += 1
                 time.sleep(0.12)

  • cli-config.yaml.example

    @@ -655,6 +655,15 @@ display:
       # Toggle at runtime with /verbose in the CLI
       tool_progress: all
     
    +  # Background process notifications (gateway/messaging only).
    +  # Controls how chatty the process watcher is when you use
    +  # terminal(background=true, check_interval=...) from Telegram/Discord/etc.
    +  #   off:     No watcher messages at all
    +  #   result:  Only the final completion message
    +  #   error:   Only the final message when exit code != 0
    +  #   all:     Running output updates + final message (default)
    +  background_process_notifications: all
    +
       # Play terminal bell when agent finishes a response.
       # Useful for long-running tasks — your terminal will ding when the agent is done.
       # Works over SSH. Most terminals can be configured to flash the taskbar or play a sound.

  • gateway/run.py

    @@ -391,6 +391,41 @@ def _load_reasoning_config() -> dict | None:
             logger.warning("Unknown reasoning_effort '%s', using default (medium)", effort)
             return None
     
    +    @staticmethod
    +    def _load_background_notifications_mode() -> str:
    +        """Load background process notification mode from config or env var.
    +
    +        Modes:
    +          - ``all``    — push running-output updates *and* the final message (default)
    +          - ``result`` — only the final completion message (regardless of exit code)
    +          - ``error``  — only the final message when exit code is non-zero
    +          - ``off``    — no watcher messages at all
    +        """
    +        mode = os.getenv("HERMES_BACKGROUND_NOTIFICATIONS", "")
    +        if not mode:
    +            try:
    +                import yaml as _y
    +                cfg_path = _hermes_home / "config.yaml"
    +                if cfg_path.exists():
    +                    with open(cfg_path, encoding="utf-8") as _f:
    +                        cfg = _y.safe_load(_f) or {}
    +                    raw = cfg.get("display", {}).get("background_process_notifications")
    +                    if raw is False:
    +                        mode = "off"
    +                    elif raw not in (None, ""):
    +                        mode = str(raw)
    +            except Exception:
    +                pass
    +        mode = (mode or "all").strip().lower()
    +        valid = {"all", "result", "error", "off"}
    +        if mode not in valid:
    +            logger.warning(
    +                "Unknown background_process_notifications '%s', defaulting to 'all'",
    +                mode,
    +            )
    +            return "all"
    +        return mode
    +
         @staticmethod
         def _load_provider_routing() -> dict:
             """Load OpenRouter provider routing preferences from config.yaml."""
    @@ -2370,6 +2405,12 @@ async def _run_process_watcher(self, watcher: dict) -> None:
     
             Runs as an asyncio task. Stays silent when nothing changed.
             Auto-removes when the process exits or is killed.
    +
    +        Notification mode (from ``display.background_process_notifications``):
    +          - ``all``    — running-output updates + final message
    +          - ``result`` — final completion message only
    +          - ``error``  — final message only when exit code != 0
    +          - ``off``    — no messages at all
             """
             from tools.process_registry import process_registry
     
    @@ -2378,8 +2419,21 @@ async def _run_process_watcher(self, watcher: dict) -> None:
             session_key = watcher.get("session_key", "")
             platform_name = watcher.get("platform", "")
             chat_id = watcher.get("chat_id", "")
    +        notify_mode = self._load_background_notifications_mode()
     
    -        logger.debug("Process watcher started: %s (every %ss)", session_id, interval)
    +        logger.debug("Process watcher started: %s (every %ss, notify=%s)",
    +                      session_id, interval, notify_mode)
    +
    +        if notify_mode == "off":
    +            # Still wait for the process to exit so we can log it, but don't
    +            # push any messages to the user.
    +            while True:
    +                await asyncio.sleep(interval)
    +                session = process_registry.get(session_id)
    +                if session is None or session.exited:
    +                    break
    +            logger.debug("Process watcher ended (silent): %s", session_id)
    +            return
     
             last_output_len = 0
             while True:
    @@ -2394,27 +2448,31 @@ async def _run_process_watcher(self, watcher: dict) -> None:
                 last_output_len = current_output_len
     
                 if session.exited:
    -                # Process finished -- deliver final update
    -                new_output = session.output_buffer[-1000:] if session.output_buffer else ""
    -                message_text = (
    -                    f"[Background process {session_id} finished with exit code {session.exit_code}~ "
    -                    f"Here's the final output:\n{new_output}]"
    +                # Decide whether to notify based on mode
    +                should_notify = (
    +                    notify_mode in ("all", "result")
    +                    or (notify_mode == "error" and session.exit_code not in (0, None))
                     )
    -                # Try to deliver to the originating platform
    -                adapter = None
    -                for p, a in self.adapters.items():
    -                    if p.value == platform_name:
    -                        adapter = a
    -                        break
    -                if adapter and chat_id:
    -                    try:
    -                        await adapter.send(chat_id, message_text)
    -                    except Exception as e:
    -                        logger.error("Watcher delivery error: %s", e)
    +                if should_notify:
    +                    new_output = session.output_buffer[-1000:] if session.output_buffer else ""
    +                    message_text = (
    +                        f"[Background process {session_id} finished with exit code {session.exit_code}~ "
    +                        f"Here's the final output:\n{new_output}]"
    +                    )
    +                    adapter = None
    +                    for p, a in self.adapters.items():
    +                        if p.value == platform_name:
    +                            adapter = a
    +                            break
    +                    if adapter and chat_id:
    +                        try:
    +                            await adapter.send(chat_id, message_text)
    +                        except Exception as e:
    +                            logger.error("Watcher delivery error: %s", e)
                     break
     
    -            elif has_new_output:
    -                # New output available -- deliver status update
    +            elif has_new_output and notify_mode == "all":
    +                # New output available -- deliver status update (only in "all" mode)
                     new_output = session.output_buffer[-500:] if session.output_buffer else ""
                     message_text = (
                         f"[Background process {session_id} is still running~ "

  • tests/gateway/test_background_process_notifications.py

    @@ -0,0 +1,198 @@
    +"""Tests for configurable background process notification modes.
    +
    +The gateway process watcher pushes status updates to users' chats when
    +background terminal commands run.  ``display.background_process_notifications``
    +controls verbosity: off | result | error | all (default).
    +
    +Contributed by @PeterFile (PR #593), reimplemented on current main.
    +"""
    +
    +import asyncio
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, patch
    +
    +import pytest
    +
    +from gateway.config import GatewayConfig, Platform
    +from gateway.run import GatewayRunner
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +class _FakeRegistry:
    +    """Return pre-canned sessions, then None once exhausted."""
    +
    +    def __init__(self, sessions):
    +        self._sessions = list(sessions)
    +
    +    def get(self, session_id):
    +        if self._sessions:
    +            return self._sessions.pop(0)
    +        return None
    +
    +
    +def _build_runner(monkeypatch, tmp_path, mode: str) -> GatewayRunner:
    +    """Create a GatewayRunner with a fake config for the given mode."""
    +    (tmp_path / "config.yaml").write_text(
    +        f"display:\n  background_process_notifications: {mode}\n",
    +        encoding="utf-8",
    +    )
    +
    +    import gateway.run as gateway_run
    +
    +    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
    +
    +    runner = GatewayRunner(GatewayConfig())
    +    adapter = SimpleNamespace(send=AsyncMock())
    +    runner.adapters[Platform.TELEGRAM] = adapter
    +    return runner
    +
    +
    +def _watcher_dict(session_id="proc_test"):
    +    return {
    +        "session_id": session_id,
    +        "check_interval": 0,
    +        "platform": "telegram",
    +        "chat_id": "123",
    +    }
    +
    +
    +# ---------------------------------------------------------------------------
    +# _load_background_notifications_mode unit tests
    +# ---------------------------------------------------------------------------
    +
    +class TestLoadBackgroundNotificationsMode:
    +
    +    def test_defaults_to_all(self, monkeypatch, tmp_path):
    +        import gateway.run as gw
    +        monkeypatch.setattr(gw, "_hermes_home", tmp_path)
    +        monkeypatch.delenv("HERMES_BACKGROUND_NOTIFICATIONS", raising=False)
    +        assert GatewayRunner._load_background_notifications_mode() == "all"
    +
    +    def test_reads_config_yaml(self, monkeypatch, tmp_path):
    +        (tmp_path / "config.yaml").write_text(
    +            "display:\n  background_process_notifications: error\n"
    +        )
    +        import gateway.run as gw
    +        monkeypatch.setattr(gw, "_hermes_home", tmp_path)
    +        monkeypatch.delenv("HERMES_BACKGROUND_NOTIFICATIONS", raising=False)
    +        assert GatewayRunner._load_background_notifications_mode() == "error"
    +
    +    def test_env_var_overrides_config(self, monkeypatch, tmp_path):
    +        (tmp_path / "config.yaml").write_text(
    +            "display:\n  background_process_notifications: error\n"
    +        )
    +        import gateway.run as gw
    +        monkeypatch.setattr(gw, "_hermes_home", tmp_path)
    +        monkeypatch.setenv("HERMES_BACKGROUND_NOTIFICATIONS", "off")
    +        assert GatewayRunner._load_background_notifications_mode() == "off"
    +
    +    def test_false_value_maps_to_off(self, monkeypatch, tmp_path):
    +        (tmp_path / "config.yaml").write_text(
    +            "display:\n  background_process_notifications: false\n"
    +        )
    +        import gateway.run as gw
    +        monkeypatch.setattr(gw, "_hermes_home", tmp_path)
    +        monkeypatch.delenv("HERMES_BACKGROUND_NOTIFICATIONS", raising=False)
    +        assert GatewayRunner._load_background_notifications_mode() == "off"
    +
    +    def test_invalid_value_defaults_to_all(self, monkeypatch, tmp_path):
    +        (tmp_path / "config.yaml").write_text(
    +            "display:\n  background_process_notifications: banana\n"
    +        )
    +        import gateway.run as gw
    +        monkeypatch.setattr(gw, "_hermes_home", tmp_path)
    +        monkeypatch.delenv("HERMES_BACKGROUND_NOTIFICATIONS", raising=False)
    +        assert GatewayRunner._load_background_notifications_mode() == "all"
    +
    +
    +# ---------------------------------------------------------------------------
    +# _run_process_watcher integration tests
    +# ---------------------------------------------------------------------------
    +
    [redacted:email]
    [redacted:email](
    +    ("mode", "sessions", "expected_calls", "expected_fragment"),
    +    [
    +        # all mode: running output → sends update
    +        (
    +            "all",
    +            [
    +                SimpleNamespace(output_buffer="building...\n", exited=False, exit_code=None),
    +                None,  # process disappears → watcher exits
    +            ],
    +            1,
    +            "is still running",
    +        ),
    +        # result mode: running output → no update
    +        (
    +            "result",
    +            [
    +                SimpleNamespace(output_buffer="building...\n", exited=False, exit_code=None),
    +                None,
    +            ],
    +            0,
    +            None,
    +        ),
    +        # off mode: exited process → no notification
    +        (
    +            "off",
    +            [SimpleNamespace(output_buffer="done\n", exited=True, exit_code=0)],
    +            0,
    +            None,
    +        ),
    +        # result mode: exited → notifies
    +        (
    +            "result",
    +            [SimpleNamespace(output_buffer="done\n", exited=True, exit_code=0)],
    +            1,
    +            "finished with exit code 0",
    +        ),
    +        # error mode: exit 0 → no notification
    +        (
    +            "error",
    +            [SimpleNamespace(output_buffer="done\n", exited=True, exit_code=0)],
    +            0,
    +            None,
    +        ),
    +        # error mode: exit 1 → notifies
    +        (
    +            "error",
    +            [SimpleNamespace(output_buffer="traceback\n", exited=True, exit_code=1)],
    +            1,
    +            "finished with exit code 1",
    +        ),
    +        # all mode: exited → notifies
    +        (
    +            "all",
    +            [SimpleNamespace(output_buffer="ok\n", exited=True, exit_code=0)],
    +            1,
    +            "finished with exit code 0",
    +        ),
    +    ],
    +)
    +async def test_run_process_watcher_respects_notification_mode(
    +    monkeypatch, tmp_path, mode, sessions, expected_calls, expected_fragment
    +):
    +    import tools.process_registry as pr_module
    +
    +    monkeypatch.setattr(pr_module, "process_registry", _FakeRegistry(sessions))
    +
    +    # Patch asyncio.sleep to avoid real delays
    +    async def _instant_sleep(*_a, **_kw):
    +        pass
    +    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
    +
    +    runner = _build_runner(monkeypatch, tmp_path, mode)
    +    adapter = runner.adapters[Platform.TELEGRAM]
    +
    +    await runner._run_process_watcher(_watcher_dict())
    +
    +    assert adapter.send.await_count == expected_calls, (
    +        f"mode={mode}: expected {expected_calls} sends, got {adapter.send.await_count}"
    +    )
    +    if expected_fragment is not None:
    +        sent_message = adapter.send.await_args.args[1]
    +        assert expected_fragment in sent_message