From 01f1ef96ae9d94d9481b775f56f189f8886d7725 Mon Sep 17 00:00:00 2001
From: funky-xamarin <30426178+Wenfengcheng@users.noreply.github.com>
Date: Sat, 26 Sep 2026 06:06:36 +0800
Subject: [PATCH 1/4] fix(cli): discard owned notification backlog on first
 interrupt

---
 cli-config.yaml.example                       |   3 +
 hermes_cli/cli_process_notifications.py       |  20 +++
 hermes_cli/cli_tui_mixin.py                   |   1 +
 hermes_cli/config_defaults.py                 |   1 +
 .../test_ctrl_c_notification_backlog.py       | 155 ++++++++++++++++++
 tools/process_registry.py                     |  43 +++++
 6 files changed, 223 insertions(+)
 create mode 100644 tests/hermes_cli/test_ctrl_c_notification_backlog.py

diff --git a/cli-config.yaml.example b/cli-config.yaml.example
index 77baaf2cfaed9..3e310e8c70cc8 100644
--- a/cli-config.yaml.example
+++ b/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.
diff --git a/hermes_cli/cli_process_notifications.py b/hermes_cli/cli_process_notifications.py
index 6f30310c56511..534bddf86e7a8 100644
--- a/hermes_cli/cli_process_notifications.py
+++ b/hermes_cli/cli_process_notifications.py
@@ -19,6 +19,26 @@ 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
+        from tools.process_registry import process_registry
+
+        if not (CLI_CONFIG.get("display") or {}).get("ctrl_c_purge_notifications", True):
+            return
+        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)
+            return
+        if count:
+            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
diff --git a/hermes_cli/cli_tui_mixin.py b/hermes_cli/cli_tui_mixin.py
index 7bc9c1b533b8f..df8f86c8769de 100644
--- a/hermes_cli/cli_tui_mixin.py
+++ b/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)
diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py
index 6a22cf6a780d5..ae855a33c5f31 100644
--- a/hermes_cli/config_defaults.py
+++ b/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"
diff --git a/tests/hermes_cli/test_ctrl_c_notification_backlog.py b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
new file mode 100644
index 0000000000000..146bf0ba2c8bc
--- /dev/null
+++ b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
@@ -0,0 +1,155 @@
+"""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()
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 89a2f360efc1b..995814a59d926 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -1832,6 +1832,49 @@ def _owns_event(evt: dict, session_key: str, owns_event, is_async_delegation: bo
         # ownership, so leave them for the owner.
         return not (is_async_delegation and evt.get("restored"))
 
+    def purge_notifications(self, session_key: str = "", owns_event=None) -> int:
+        """Discard this owner's current notification backlog, not running work or output.
+
+        Reuse drain routing. Terminal process consumption suppresses requeued copies;
+        durable delegation results are claim-scoped drops, never successful deliveries.
+        Only inspect the queue's starting size so new completions remain notifications.
+        """
+        from tools.async_delegation import claim_event_delivery, drop_completion_delivery
+
+        requeue = []
+        purged = 0
+        try:
+            for _ in range(self.completion_queue.qsize()):
+                try:
+                    event = self.completion_queue.get_nowait()
+                except queue.Empty:
+                    break
+                is_delegation = event.get("type") == "async_delegation"
+                if not self._owns_event(event, session_key, owns_event, is_delegation):
+                    requeue.append(event)
+                    continue
+                try:
+                    claim = claim_event_delivery(event, "notification-purge")
+                    if claim is None:
+                        requeue.append(event)
+                        continue
+                    if claim and not drop_completion_delivery(str(event["delegation_id"]), claim):
+                        requeue.append(event)
+                        continue
+                except Exception:
+                    requeue.append(event)
+                    raise
+                if not is_delegation and event.get("type", "completion") == "completion":
+                    if sid := event.get("session_id"):
+                        with self._lock:
+                            self._completion_consumed.add(sid)
+                            self._poll_observed.add(sid)
+                purged += 1
+        finally:
+            for event in requeue:
+                self.completion_queue.put(event)
+        return purged
+
     def drain_notifications(
         self, session_key: str = "", owns_event=None, *, skip_poll_observed: bool = True,
     ) -> "list[tuple[dict, str]]":

From d9789fa60cee9050094950d553a9e364419f51cc Mon Sep 17 00:00:00 2001
From: funky-xamarin <30426178+Wenfengcheng@users.noreply.github.com>
Date: Sat, 26 Sep 2026 06:12:22 +0800
Subject: [PATCH 2/4] fix(cli): release failed purge claims and verify real
 process output

---
 .../test_ctrl_c_notification_backlog.py       | 33 +++++++++++++++++++
 tools/process_registry.py                     |  9 ++++-
 2 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/tests/hermes_cli/test_ctrl_c_notification_backlog.py b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
index 146bf0ba2c8bc..4703a5403cd04 100644
--- a/tests/hermes_cli/test_ctrl_c_notification_backlog.py
+++ b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
@@ -153,3 +153,36 @@ def test_empty_purge_still_interrupts(monkeypatch):
     cli._tui_handle_ctrl_c(event)
     cli.agent.hard_interrupt.assert_called_once()
     assert registry.completion_queue.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"
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 995814a59d926..cf4a8203572c7 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -1839,7 +1839,9 @@ def purge_notifications(self, session_key: str = "", owns_event=None) -> int:
         durable delegation results are claim-scoped drops, never successful deliveries.
         Only inspect the queue's starting size so new completions remain notifications.
         """
-        from tools.async_delegation import claim_event_delivery, drop_completion_delivery
+        from tools.async_delegation import (
+            claim_event_delivery, defer_completion_delivery, drop_completion_delivery,
+        )
 
         requeue = []
         purged = 0
@@ -1853,15 +1855,20 @@ def purge_notifications(self, session_key: str = "", owns_event=None) -> int:
                 if not self._owns_event(event, session_key, owns_event, is_delegation):
                     requeue.append(event)
                     continue
+                claim = None
                 try:
                     claim = claim_event_delivery(event, "notification-purge")
                     if claim is None:
                         requeue.append(event)
                         continue
                     if claim and not drop_completion_delivery(str(event["delegation_id"]), claim):
+                        defer_completion_delivery(str(event["delegation_id"]), claim)
                         requeue.append(event)
                         continue
                 except Exception:
+                    if claim:
+                        with suppress(Exception):
+                            defer_completion_delivery(str(event["delegation_id"]), claim)
                     requeue.append(event)
                     raise
                 if not is_delegation and event.get("type", "completion") == "completion":

From 09404606731dee87ab403d62e5c79be380737427 Mon Sep 17 00:00:00 2001
From: funky-xamarin <30426178+Wenfengcheng@users.noreply.github.com>
Date: Sat, 26 Sep 2026 06:16:33 +0800
Subject: [PATCH 3/4] fix(cli): purge legacy delegation notices without durable
 rows

---
 tests/hermes_cli/test_ctrl_c_notification_backlog.py | 9 +++++++++
 tools/process_registry.py                            | 4 +++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/tests/hermes_cli/test_ctrl_c_notification_backlog.py b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
index 4703a5403cd04..cb5f24e39a1dc 100644
--- a/tests/hermes_cli/test_ctrl_c_notification_backlog.py
+++ b/tests/hermes_cli/test_ctrl_c_notification_backlog.py
@@ -155,6 +155,15 @@ def test_empty_purge_still_interrupts(monkeypatch):
     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
diff --git a/tools/process_registry.py b/tools/process_registry.py
index cf4a8203572c7..bcc9b17b7b633 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -1841,6 +1841,7 @@ def purge_notifications(self, session_key: str = "", owns_event=None) -> int:
         """
         from tools.async_delegation import (
             claim_event_delivery, defer_completion_delivery, drop_completion_delivery,
+            get_durable_delegation,
         )
 
         requeue = []
@@ -1861,7 +1862,8 @@ def purge_notifications(self, session_key: str = "", owns_event=None) -> int:
                     if claim is None:
                         requeue.append(event)
                         continue
-                    if claim and not drop_completion_delivery(str(event["delegation_id"]), claim):
+                    if (claim and get_durable_delegation(str(event["delegation_id"])) is not None
+                            and not drop_completion_delivery(str(event["delegation_id"]), claim)):
                         defer_completion_delivery(str(event["delegation_id"]), claim)
                         requeue.append(event)
                         continue

From cbef31b6b4800942920bc8c303d1f001fdd9c5f8 Mon Sep 17 00:00:00 2001
From: funky-xamarin <30426178+Wenfengcheng@users.noreply.github.com>
Date: Sat, 26 Sep 2026 06:42:09 +0800
Subject: [PATCH 4/4] fix(cli): expose non-destructive backlog purge during
 active turns

---
 hermes_cli/cli_info_mixin.py                  |  4 +-
 hermes_cli/cli_process_notifications.py       | 18 +++-
 hermes_cli/commands.py                        |  2 +
 .../test_purge_notification_command.py        | 85 +++++++++++++++++++
 4 files changed, 105 insertions(+), 4 deletions(-)
 create mode 100644 tests/hermes_cli/test_purge_notification_command.py

diff --git a/hermes_cli/cli_info_mixin.py b/hermes_cli/cli_info_mixin.py
index 978a92077c129..93d175dfd8903 100644
--- a/hermes_cli/cli_info_mixin.py
+++ b/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.
diff --git a/hermes_cli/cli_process_notifications.py b/hermes_cli/cli_process_notifications.py
index 534bddf86e7a8..4ffbbdefe959b 100644
--- a/hermes_cli/cli_process_notifications.py
+++ b/hermes_cli/cli_process_notifications.py
@@ -22,10 +22,22 @@ def _owns_process_notification(self, event: dict) -> bool:
     def _purge_interrupt_notifications(self) -> None:
         """A first Ctrl+C explicitly discards this session's pending reports only."""
         from cli import CLI_CONFIG
-        from tools.process_registry import process_registry
 
         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 "",
@@ -35,8 +47,10 @@ def _purge_interrupt_notifications(self) -> None:
             # 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:
+        if count or report_empty:
             print(f"Discarded {count} pending background notification(s); output remains available.")
 
     def _drain_process_notifications(self, consumer: str) -> None:
diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py
index cfe1d07566ef2..27b47a9ef8b57 100644
--- a/hermes_cli/commands.py
+++ b/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",
diff --git a/tests/hermes_cli/test_purge_notification_command.py b/tests/hermes_cli/test_purge_notification_command.py
new file mode 100644
index 0000000000000..2ab891dfc4fa1
--- /dev/null
+++ b/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)