fix(cli): discard owned notification backlog on first Ctrl+C

cli-config.yaml.example

@@ -1907,6 +1907,9 @@ display:
   #   error:   Only the final raw-output message when exit code != 0
   #   all:     Running output updates + final raw-output message
   background_process_notifications: concise
+  # First Ctrl+C in the classic CLI discards this session's pending background reports.
+  # Running work and stored output/results survive. false restores interrupt-only behavior.
+  ctrl_c_purge_notifications: true
 
 
   # Play terminal bell when agent finishes a response.

hermes_cli/cli_info_mixin.py

@@ -462,9 +462,9 @@ def _should_handle_steer_command_inline(self, text: str, has_images: bool = Fals
 
     def _should_handle_background_command_inline(
         self, text: str, has_images: bool = False) -> bool:
-        """Return True when /bg or /btw should be dispatched while the agent runs (their
+        """Return True when /bg, /btw or /purge should dispatch while the agent runs (their
         ``CommandDef`` entries declare ``busy_policy="dispatch"``; the classic CLI honours it here)."""
-        return self._busy_inline_command(text, has_images, ("bg", "btw"))
+        return self._busy_inline_command(text, has_images, ("bg", "btw", "purge"))
 
     def handle_bang_shell(self, text: str) -> bool:
         """Run a ``!<command>`` submission. Returns True when it was handled.

hermes_cli/cli_process_notifications.py

@@ -19,6 +19,40 @@ def _owns_process_notification(self, event: dict) -> bool:
             resolved_key = event_key
         return str(resolved_key) == current_key
 
+    def _purge_interrupt_notifications(self) -> None:
+        """A first Ctrl+C explicitly discards this session's pending reports only."""
+        from cli import CLI_CONFIG
+
+        if not (CLI_CONFIG.get("display") or {}).get("ctrl_c_purge_notifications", True):
+            return
+        self._purge_owned_notifications()
+
+    def _handle_purge_command(self, command: str) -> None:
+        """Discard pending reports without interrupting work or clearing user input."""
+        if len(command.split()) != 1:
+            print("Usage: /purge — discard this session's pending registry notifications; "
+                  "all/kill are not supported.")
+            return
+        self._purge_owned_notifications(report_empty=True)
+
+    def _purge_owned_notifications(self, *, report_empty: bool = False) -> None:
+        from tools.process_registry import process_registry
+
+        try:
+            count = process_registry.purge_notifications(
+                session_key=getattr(self, "session_id", "") or "",
+                owns_event=self._owns_process_notification,
+            )
+        except Exception:
+            # A failed ledger operation must never prevent the requested interrupt.
+            import logging
+            logging.getLogger(__name__).warning("Could not discard pending notifications", exc_info=True)
+            if report_empty:
+                print("Could not discard all pending notifications; see the log for details.")
+            return
+        if count or report_empty:
+            print(f"Discarded {count} pending background notification(s); output remains available.")
+
     def _drain_process_notifications(self, consumer: str) -> None:
         from tools.process_registry import process_registry
         from tools.async_delegation import claim_event_delivery, complete_event_delivery

hermes_cli/cli_tui_mixin.py

@@ -1029,6 +1029,7 @@ def _tui_handle_ctrl_c(self, event):
                 return
             self._last_ctrl_c_time = now
             print("\n⚡ Interrupting agent... (press Ctrl+C again to force exit)")
+            self._purge_interrupt_notifications()
             request_hard_interrupt(self.agent)
         else:
             self._tui_clear_or_exit(event)

hermes_cli/commands.py

@@ -91,6 +91,8 @@ class CommandDef:
                cli_only=True, args_hint="<archive.tar.gz> [--name <name>]"),
     CommandDef("stop", "Kill all running background processes", "Session",
                busy_policy="interrupt_then_dispatch", busy_handler="stop"),
+    CommandDef("purge", "Discard this session's pending background reports (keep running work)", "Session",
+               cli_only=True, desktop="terminal", busy_policy="dispatch"),
     CommandDef("pause", "Pause new work globally (emergency stop); '/pause off' resumes", "Session",
                gateway_only=True, args_hint="[reason | off]", busy_policy="dispatch"),
     CommandDef("approve", "Approve a pending dangerous command", "Session",

hermes_cli/config_defaults.py

@@ -839,6 +839,7 @@ def _aux(timeout, *, reasoning_effort=True, **extra):
         # failures append an output tail) | "all" (running updates + final raw output) | "result"
         # (final raw only) | "error" (raw only on non-zero exit) | "off".
         "background_process_notifications": "concise",
+        "ctrl_c_purge_notifications": True,  # First Ctrl+C discards pending reports, not running work/output.
         "streaming": False,
         "timestamps": False,      # message timestamps (CLI labels, TUI rows, desktop transcript)
         "timestamp_format": "%H:%M",  # strftime format, e.g. "%b-%d %H:%M"

tests/hermes_cli/test_ctrl_c_notification_backlog.py

@@ -0,0 +1,197 @@
+"""A first Ctrl+C must not turn a pending completion into another CLI turn."""
+import queue
+import threading
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from hermes_cli.cli_process_notifications import CLIProcessNotificationsMixin
+from hermes_cli.cli_tui_mixin import CLITuiMixin
+from tools.process_registry import ProcessRegistry
+
+
+class Harness(CLITuiMixin, CLIProcessNotificationsMixin):
+    pass
+
+
+def cli_fixture(monkeypatch):
+    cli = Harness()
+    cli.session_id = "owner"
+    cli._pending_input = queue.Queue()
+    cli._session_db = None
+    cli._voice_lock = threading.Lock()
+    cli._voice_recording = False
+    for name in ("_slash_confirm_state", "_model_picker_state", "_command_palette_state",
+                 "_sudo_state", "_secret_state", "_approval_state", "_clarify_state", "_connection_state"):
+        setattr(cli, name, None)
+    cli._close_model_picker = Mock()
+    cli._close_command_palette = Mock()
+    cli.agent = SimpleNamespace(hard_interrupt=Mock())
+    cli._agent_running = True
+    cli._last_ctrl_c_time = 0
+    cli._should_exit = False
+    registry = ProcessRegistry()
+    monkeypatch.setattr("tools.process_registry.process_registry", registry)
+    import cli as facade
+    monkeypatch.setattr(facade, "CLI_CONFIG", {"display": {}})
+    return cli, registry, SimpleNamespace(app=Mock())
+
+
+def completion(sid="proc_owned", owner="owner"):
+    return {"type": "completion", "session_id": sid, "session_key": owner,
+            "command": "fixture", "exit_code": 0, "output": "done"}
+
+
+def test_ctrl_c_discards_owned_backlog_and_suppresses_requeued_copy(monkeypatch):
+    cli, registry, event = cli_fixture(monkeypatch)
+    own, foreign = completion(), completion("proc_foreign", "other")
+    registry.completion_queue.put(own)
+    registry.completion_queue.put(foreign)
+    cli._tui_handle_ctrl_c(event)
+    cli.agent.hard_interrupt.assert_called_once()
+    cli._drain_process_notifications("cli-post-turn")
+    assert cli._pending_input.empty(), "Ctrl+C must not restart the agent from its completion backlog"
+    assert registry.is_completion_consumed("proc_owned")
+    registry.completion_queue.put(dict(own))
+    assert registry.drain_notifications(session_key="owner") == []
+    remaining = registry.drain_notifications(session_key="other")
+    assert [e for e, _ in remaining] == [foreign]
+    cli._tui_handle_ctrl_c(event)
+    assert cli._should_exit
+    event.app.exit.assert_called_once()
+
+
+def test_ctrl_c_opt_out_preserves_notification(monkeypatch):
+    cli, registry, event = cli_fixture(monkeypatch)
+    monkeypatch.setattr("cli.CLI_CONFIG", {"display": {"ctrl_c_purge_notifications": False}})
+    registry.completion_queue.put(completion())
+    cli._tui_handle_ctrl_c(event)
+    cli._drain_process_notifications("cli-post-turn")
+    assert not cli._pending_input.empty()
+    cli.agent.hard_interrupt.assert_called_once()
+
+
+@pytest.mark.parametrize("owner", ["other", ""])
+def test_ctrl_c_does_not_adopt_unproven_delegation(monkeypatch, owner):
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = {"type": "async_delegation", "delegation_id": "foreign", "session_key": owner}
+    registry.completion_queue.put(notice)
+    cli._tui_handle_ctrl_c(event)
+    assert registry.completion_queue.get_nowait() == notice
+    cli.agent.hard_interrupt.assert_called_once()
+
+
+def durable_notice(owner="owner", deleg_id="deleg_fixture", interim=False):
+    import time
+    from tools import async_delegation as durable
+    notice = {"type": "async_delegation", "delegation_id": deleg_id,
+              "session_key": owner, "dispatched_at": time.time(),
+              "status": "completed", "summary": "preserved result"}
+    durable._persist_dispatch(notice)
+    if interim:
+        notice["task_failure_notice"] = True
+    else:
+        durable._persist_completion(notice, {"summary": "preserved result"})
+    return notice
+
+
+def test_ctrl_c_drops_durable_completion_without_deleting_result(monkeypatch):
+    from tools import async_delegation as durable
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = durable_notice()
+    foreign = durable_notice("other", "deleg_foreign")
+    registry.completion_queue.put(notice)
+    registry.completion_queue.put(foreign)
+    cli._tui_handle_ctrl_c(event)
+    row = durable.get_durable_delegation("deleg_fixture")
+    assert row["delivery_state"] == "dropped"
+    assert row["result"]["summary"] == "preserved result"
+    assert durable.get_durable_delegation("deleg_foreign")["delivery_state"] == "pending"
+    restored = queue.Queue()
+    durable.restore_undelivered_completions(restored)
+    assert restored.get_nowait()["delegation_id"] == "deleg_foreign"
+    assert restored.empty()
+    registry.completion_queue.put(dict(notice))
+    cli._drain_process_notifications("cli-post-turn")
+    assert cli._pending_input.empty()
+
+
+def test_ctrl_c_interim_notice_does_not_drop_future_final(monkeypatch):
+    from tools import async_delegation as durable
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = durable_notice(interim=True)
+    registry.completion_queue.put(notice)
+    cli._tui_handle_ctrl_c(event)
+    assert registry.completion_queue.empty()
+    assert durable.get_durable_delegation("deleg_fixture")["delivery_state"] == "pending"
+
+
+def test_ctrl_c_preserves_claim_owned_by_another_consumer(monkeypatch):
+    from tools import async_delegation as durable
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = durable_notice()
+    assert durable.claim_completion_delivery("deleg_fixture", "other-consumer")
+    registry.completion_queue.put(notice)
+    cli._tui_handle_ctrl_c(event)
+    assert registry.completion_queue.get_nowait() == notice
+    assert durable.get_durable_delegation("deleg_fixture")["delivery_state"] == "pending"
+
+
+def test_failed_purge_preserves_event_and_still_interrupts(monkeypatch):
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = completion()
+    registry.completion_queue.put(notice)
+    monkeypatch.setattr("tools.async_delegation.claim_event_delivery", Mock(side_effect=OSError("fixture")))
+    cli._tui_handle_ctrl_c(event)
+    cli.agent.hard_interrupt.assert_called_once()
+    assert registry.completion_queue.get_nowait() == notice
+
+
+def test_empty_purge_still_interrupts(monkeypatch):
+    cli, registry, event = cli_fixture(monkeypatch)
+    cli._tui_handle_ctrl_c(event)
+    cli.agent.hard_interrupt.assert_called_once()
+    assert registry.completion_queue.empty()
+
+
+def test_ctrl_c_purges_legacy_delegation_without_ledger_row(monkeypatch):
+    cli, registry, event = cli_fixture(monkeypatch)
+    registry.completion_queue.put({"type": "async_delegation", "delegation_id": "legacy",
+                                   "session_key": "owner", "summary": "legacy result"})
+    cli._tui_handle_ctrl_c(event)
+    cli._drain_process_notifications("cli-post-turn")
+    assert cli._pending_input.empty()
+
+
+def test_completed_real_process_keeps_output_after_ctrl_c(monkeypatch, tmp_path):
+    import subprocess
+    import sys
+    cli, registry, event = cli_fixture(monkeypatch)
+    proc = subprocess.Popen([sys.executable, "-c", "print('actual-background-output')"],
+                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                            text=True, encoding="utf-8", cwd=tmp_path)
+    session = registry.adopt_local(proc, command="fixture python", cwd=str(tmp_path),
+                                   task_id="owner", session_key="owner", notify_on_complete=True)
+    assert session._completion_event.wait(10), "real child must complete"
+    cli._tui_handle_ctrl_c(event)
+    cli._drain_process_notifications("cli-post-turn")
+    assert cli._pending_input.empty()
+    assert registry.is_completion_consumed(session.id)
+    assert registry.get(session.id).exited
+    assert "actual-background-output" in registry.read_log(session.id)["output"]
+
+
+@pytest.mark.parametrize("failure", [False, OSError("drop failed")])
+def test_failed_durable_drop_releases_claim_for_normal_delivery(monkeypatch, failure):
+    from tools import async_delegation as durable
+    cli, registry, event = cli_fixture(monkeypatch)
+    notice = durable_notice()
+    registry.completion_queue.put(notice)
+    drop = Mock(side_effect=failure) if isinstance(failure, Exception) else Mock(return_value=False)
+    monkeypatch.setattr(durable, "drop_completion_delivery", drop)
+    cli._tui_handle_ctrl_c(event)
+    cli.agent.hard_interrupt.assert_called_once()
+    cli._drain_process_notifications("cli-post-turn")
+    assert not cli._pending_input.empty(), "failed purge must leave a normally deliverable result"
+    assert durable.get_durable_delegation("deleg_fixture")["delivery_state"] == "delivered"

tests/hermes_cli/test_purge_notification_command.py

@@ -0,0 +1,85 @@
+"""Explicit backlog disposal must work even while a foreground turn runs."""
+import queue
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from cli import HermesCLI
+from tools.process_registry import ProcessRegistry
+
+
+def fixture(monkeypatch):
+    cli = HermesCLI.__new__(HermesCLI)
+    cli.session_id = "owner"
+    cli._session_db = None
+    cli._pending_input = queue.Queue()
+    cli._agent_running = True
+    cli._should_handle_model_command_inline = lambda *a, **kw: False
+    cli._process_unregistered_slash = Mock(return_value=True)
+    registry = ProcessRegistry()
+    monkeypatch.setattr("tools.process_registry.process_registry", registry)
+    monkeypatch.setattr("cli.CLI_CONFIG", {"display": {"ctrl_c_purge_notifications": False}})
+    return cli, registry
+
+
+def notice(sid="proc_owned", owner="owner"):
+    return {"type": "completion", "session_id": sid, "session_key": owner,
+            "command": "fixture", "exit_code": 0, "output": "retained result"}
+
+
+def test_purge_dispatch_discards_only_owned_reports_even_with_ctrl_c_opt_out(monkeypatch):
+    cli, registry = fixture(monkeypatch)
+    own, foreign = notice(), notice("proc_foreign", "other")
+    registry.completion_queue.put(own)
+    registry.completion_queue.put(foreign)
+    cli._pending_input.put("user prompt must survive")
+    assert cli.process_command("/purge")
+    assert registry.drain_notifications(session_key="owner") == []
+    assert registry.is_completion_consumed("proc_owned")
+    assert [e for e, _ in registry.drain_notifications(session_key="other")] == [foreign]
+    assert cli._pending_input.get_nowait() == "user prompt must survive"
+    assert cli._agent_running
+    cli._process_unregistered_slash.assert_not_called()
+
+
+def test_purge_enter_bypasses_busy_queue(monkeypatch):
+    cli, registry = fixture(monkeypatch)
+    registry.completion_queue.put(notice())
+    event = SimpleNamespace(app=Mock())
+    assert cli._tui_enter_inline_command(event, "/purge", False)
+    assert registry.completion_queue.empty()
+    assert cli._pending_input.empty()
+    event.app.current_buffer.reset.assert_called_once_with(append_to_history=True)
+    event.app.invalidate.assert_called_once()
+
+
+@pytest.mark.parametrize("text", ["/purge kill", "/purge all", "/purge unknown", "/purge all extra"])
+def test_unsupported_purge_modes_are_not_silently_destructive(monkeypatch, capsys, text):
+    cli, registry = fixture(monkeypatch)
+    own = notice()
+    registry.completion_queue.put(own)
+    assert cli.process_command(text)
+    assert registry.completion_queue.get_nowait() == own
+    assert "Usage:" in capsys.readouterr().out
+
+
+def test_explicit_empty_purge_reports_zero(monkeypatch, capsys):
+    cli, registry = fixture(monkeypatch)
+    assert cli.process_command("/purge")
+    assert "Discarded 0" in capsys.readouterr().out
+
+
+def test_purge_failure_keeps_event_and_reports_failure(monkeypatch, capsys):
+    cli, registry = fixture(monkeypatch)
+    registry.completion_queue.put(notice())
+    monkeypatch.setattr(registry, "purge_notifications", Mock(side_effect=RuntimeError("fixture")))
+    assert cli.process_command("/purge")
+    assert not registry.completion_queue.empty()
+    assert "Could not discard" in capsys.readouterr().out
+
+
+def test_purge_is_not_inline_when_idle_or_attached(monkeypatch):
+    cli, _ = fixture(monkeypatch)
+    assert not cli._tui_enter_inline_command(SimpleNamespace(app=Mock()), "/purge", True)
+    cli._agent_running = False
+    assert not cli._tui_enter_inline_command(SimpleNamespace(app=Mock()), "/purge", False)