feat(gateway): configurable background process watcher notifications

Add display.background_process_notifications config option to control
how chatty the gateway process watcher is when using
terminal(background=true, check_interval=...) from messaging platforms.

Modes:
  - all:    running-output updates + final message (default, current behavior)
  - result: only the final completion message
  - error:  only the final message when exit code != 0
  - off:    no watcher messages at all

Also supports HERMES_BACKGROUND_NOTIFICATIONS env var override.

Includes 12 tests (5 config loading + 7 watcher behavior).

Inspired by @PeterFile's PR #593. Closes #592.

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

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 @@ class GatewayRunner:
         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 @@ class GatewayRunner:
 
         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 @@ class GatewayRunner:
         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 @@ class GatewayRunner:
             last_output_len = current_output_len
 
             if session.exited:
-                # Process finished -- deliver final update
+                # Decide whether to notify based on mode
-                new_output = session.output_buffer[-1000:] if session.output_buffer else ""
+                should_notify = (
-                message_text = (
+                    notify_mode in ("all", "result")
-                    f"[Background process {session_id} finished with exit code {session.exit_code}~ "
+                    or (notify_mode == "error" and session.exit_code not in (0, None))
-                    f"Here's the final output:\n{new_output}]"
                 )
-                # Try to deliver to the originating platform
+                if should_notify:
-                adapter = None
+                    new_output = session.output_buffer[-1000:] if session.output_buffer else ""
-                for p, a in self.adapters.items():
+                    message_text = (
-                    if p.value == platform_name:
+                        f"[Background process {session_id} finished with exit code {session.exit_code}~ "
-                        adapter = a
+                        f"Here's the final output:\n{new_output}]"
-                        break
+                    )
-                if adapter and chat_id:
+                    adapter = None
-                    try:
+                    for p, a in self.adapters.items():
-                        await adapter.send(chat_id, message_text)
+                        if p.value == platform_name:
-                    except Exception as e:
+                            adapter = a
-                        logger.error("Watcher delivery error: %s", e)
+                            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:
+            elif has_new_output and notify_mode == "all":
-                # New output available -- deliver status update
+                # 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
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("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