fix(cli): answer gateway-owned state from the gateway in classic CLI commands

The classic CLI drives a GatewayAgent while tools, checkpoints and the
environment live in the gateway, so several commands read empty CLI-process
state.

- CLI-7 /agents lists the gateway's processes and its running-subagent count.
- CLI-8 /diff session goes to the gateway's /diff handler via slash.exec
  (it crashed on the proxy's missing checkpoint manager).
- CLI-17 /handoff says it is not available on the shared gateway yet, as
  Ink and Desktop do, instead of silently doing nothing.
- CLI-21 /reload also reloads the gateway's .env (reload.env) and reports
  the gateway's count.
- CLI-23 /insights --source cli also matches `local` rows whose
  created_source is that surface (older databases fall back to the label).
- N4: /stop typed during a turn interrupts it like Ctrl+C (its declared
  interrupt_then_dispatch policy), then stops background processes as
  before. An interrupt before the first lease attempt no longer says
  "Stopped waiting for another Hermes process".
- W2: tools/bot_live_delivery.py skips chmod(0o700) on Windows, where it
  rewrote the NTFS ACL and the Telegram adapter crash-looped on its lock.

Refs #134889 (CLI-7, CLI-8, CLI-17, CLI-21, CLI-23, N4, W2)

agent/insights.py

@@ -77,9 +77,16 @@ def _day(ts: Any) -> str:
     return safe_strftime(datetime.fromtimestamp(ts), "%b %d") if ts and (ts := coerce_epoch(ts)) else "?"
 
 
-def _scoped(before: str, after: str = "", *, src: str = " AND s.source = ?") -> tuple[str, str]:
+# A source filter also matches the gateway's ``local`` rows that surface created: interactive local
+# sessions are stored as ``local`` with the surface kept in ``created_source``
+# (gateway/session_a2a.storage_source), so ``--source cli`` still finds the classic CLI's sessions.
+_LOCAL_SURFACE = " OR ({a}source = 'local' AND {a}created_source = ?)"
+
+
+def _scoped(before: str, after: str = "", *, alias: str = "s.") -> tuple[str, str]:
     """(unfiltered, source-filtered) query pair sharing one body. Built once at class definition,
     so no runtime value can alter query structure."""
+    src = f" AND ({alias}source = ?{_LOCAL_SURFACE.format(a=alias)})"
     return before + after, before + src + after
 
 
@@ -95,7 +102,7 @@ class InsightsEngine:
     _GET_SESSIONS_ALL, _GET_SESSIONS_WITH_SOURCE = _scoped(
         f"SELECT {_SESSION_COLS} FROM sessions WHERE started_at >= ?",
         " ORDER BY started_at DESC",
-        src=" AND source = ?",
+        alias="",
     )
 
     # ``INDEXED BY`` pins the partial index so the plan is deterministic on a
@@ -151,6 +158,8 @@ class InsightsEngine:
         " WHERE s.started_at >= ?",
     )
     _PINNED = ("_GET_TOOL_CALLS", "_GET_SKILL_CALLS")
+    _SOURCE_SCOPED = ("_GET_SESSIONS", "_GET_TOOL_CALLS", "_GET_SKILL_CALLS", "_GET_TOOL_NAMES",
+                      "_GET_MESSAGE_STATS", "_GET_MODEL_USAGE")
 
     def __init__(self, db):
         self.db = db
@@ -165,11 +174,24 @@ class InsightsEngine:
             for base in self._PINNED:
                 for suffix in ("_ALL", "_WITH_SOURCE"):
                     setattr(self, base + suffix, getattr(self, base + suffix).replace(strip, ""))
+        # Same read-only caveat: an older writer's DB has no ``created_source`` (and no ``local`` rows).
+        try:
+            self._local_surfaces = any(row[1] == "created_source" for row in self._conn.execute("PRAGMA table_info(sessions)"))
+        except sqlite3.Error:
+            self._local_surfaces = False
+        if not self._local_surfaces:
+            for base in self._SOURCE_SCOPED:
+                sql = getattr(self, base + "_WITH_SOURCE")
+                for alias in ("", "s."):
+                    sql = sql.replace(_LOCAL_SURFACE.format(a=alias), "")
+                setattr(self, base + "_WITH_SOURCE", sql)
 
     def _query(self, base: str, cutoff: float, source: Optional[str]) -> list:
         """Rows of ``<base>_WITH_SOURCE`` or ``<base>_ALL`` (instance attrs, so the unpinned fallback applies)."""
-        sql, params = (getattr(self, base + "_WITH_SOURCE"), (cutoff, source)) if source else (getattr(self, base + "_ALL"), (cutoff,))
-        return self._conn.execute(sql, params).fetchall()
+        if not source:
+            return self._conn.execute(getattr(self, base + "_ALL"), (cutoff,)).fetchall()
+        params = (cutoff, source, source) if self._local_surfaces else (cutoff, source)
+        return self._conn.execute(getattr(self, base + "_WITH_SOURCE"), params).fetchall()
 
     def generate(self, days: int = 30, source: str | None = None) -> dict[str, Any]:
         """Generate a complete insights report for the last ``days`` days, optionally filtered by source platform."""

agent/turn_facade_lease.py

@@ -347,7 +347,8 @@ def admit_durable_turn_lease(
         on_wait=_on_wait, on_contended=_on_contended,
         should_abort=lambda: getattr(agent, "_interrupt_requested", False),
     ):
-        admission.early_result = _lease_not_acquired_result(agent, session_id, conversation_history)
+        admission.early_result = _lease_not_acquired_result(agent, session_id, conversation_history,
+                                                            waited=announced)
         return admission
 
     # Assign only after admission so the finally cannot release a holder that never owned the
@@ -437,7 +438,9 @@ def carry_unadmitted_user_message(
     append_message(early_result["messages"], deferred_user, timestamp=timestamp)
 
 
-def _lease_not_acquired_result(agent, session_id: str, conversation_history) -> dict[str, Any]:
+def _lease_not_acquired_result(agent, session_id: str, conversation_history, *, waited: bool) -> dict[str, Any]:
+    """``waited``: another holder was seen (the user was told so), not merely an interrupt that
+    landed before the first acquisition attempt, which no other process was involved in."""
     base = {"messages": list(conversation_history or []), "api_calls": 0, "completed": False}
     if getattr(agent, "_interrupt_requested", False):
         logger.info("session turn lease wait aborted by interrupt: %s", session_id)
@@ -449,6 +452,8 @@ def _lease_not_acquired_result(agent, session_id: str, conversation_history) ->
             "final_response": (
                 "Stopped waiting for another Hermes process on this session. "
                 "Your message was not processed."
+                if waited else
+                "Stopped before this turn started. Your message was not processed."
             ),
             **base,
             "interrupted": True,

cli.py

@@ -1194,6 +1194,9 @@ class HermesCLI(CLIInitMixin, CLITuiRuntimeMixin, CLIProcessNotificationsMixin,
         from hermes_cli.gateway_setting_commands import run as run_gateway_setting
         if run_gateway_setting(self, canonical, cmd_original):
             return True  # a per-session setting the gateway's handler applied
+        from hermes_cli.gateway_cli_commands import run as run_gateway_command
+        if run_gateway_command(self, canonical, cmd_original):
+            return True  # state the gateway session owns (checkpoints, .env, handoff)
         method_name, pass_arg = entry
         handler = getattr(self, method_name)
         result = handler(cmd_original) if pass_arg else handler()

hermes_cli/cli_commands_session_tools.py

@@ -97,11 +97,16 @@ class CLICommandsSessionToolsMixin:
     # ---- /stop, /agents -------------------------------------------------------------------
     def _handle_stop_command(self):
         """Handle /stop — kill all running background processes and background (async) delegations.
-        Separate from interrupt (stop the current turn), as in Codex.
+        Typed while a turn runs it stops that turn first, as Ctrl+C does (the registry's
+        ``busy_policy="interrupt_then_dispatch"``; the input loop dispatches it inline).
 
         See #14602.
         """
         from hermes_cli.gateway_tools import process_registry as _registry, process_view
+        if getattr(self, "_agent_running", False) and getattr(self, "agent", None):
+            from agent.interrupt_compat import request_hard_interrupt
+            print(f"  {t('cli.tui.interrupting')}")
+            request_hard_interrupt(self.agent)
         process_registry, view = _registry(), process_view()
         if view is not None:
             view.refresh(force=True)
@@ -122,8 +127,13 @@ class CLICommandsSessionToolsMixin:
             print(f"  {_t('stop.interrupted_delegations', count=interrupted)}")
 
     def _handle_agents_command(self):
-        """Handle /agents — show background processes and agent status."""
-        from tools.process_registry import format_uptime_short, process_registry
+        """Handle /agents — show background processes and agent status. On the gateway both run there:
+        the session's ``process.list`` and its usage's running-subagent count (no per-delegation rows)."""
+        from hermes_cli.gateway_tools import process_registry as _registry, process_view
+        from tools.process_registry import format_uptime_short
+        process_registry, view = _registry(), process_view()
+        if view is not None:
+            view.refresh(force=True)
         processes = process_registry.list_sessions()
         running = [p for p in processes if p.get("status") == "running"]
         finished = [p for p in processes if p.get("status") != "running"]
@@ -133,36 +143,10 @@ class CLICommandsSessionToolsMixin:
             _cp(f"    {p.get('session_id', '?')} · {up} · {p.get('command', '')[:80]}")
         if finished:
             _cp(f"  {_t('agents.recently_finished', count=len(finished))}")
-        # Background (async) delegations — delegate_task(background=true)
-        delegations = _probe("tools.async_delegation", "list_async_delegations", [])
-        if delegations:
-            running_d = [d for d in delegations if d.get("status") in ("running", "stalling")]
-            _cp(f"  {_t('agents.background_delegations', count=len(running_d))}")
-            for d in delegations:
-                status = d.get("status", "?")
-                line = f"    {d.get('delegation_id', '?')} · {status} · {(d.get('goal') or '')[:60]}"
-                # Live-status detail for in-flight delegations.
-                # See #51690.
-                if status == "stalling":
-                    quiet = d.get("stalled_after_quiet_seconds")
-                    if quiet is not None:
-                        line += _t("agents.no_progress", seconds=f"{quiet:.0f}")
-                elif status == "running":
-                    quiet = d.get("seconds_since_progress")
-                    if quiet is not None and quiet >= 60:
-                        line += _t("agents.quiet", seconds=f"{quiet:.0f}")
-                _cp(line)
-                for i, child in enumerate(d.get("children_activity") or []):
-                    if not isinstance(child, dict):
-                        continue
-                    tool = child.get("current_tool")
-                    doing = _t("agents.in_tool", tool=tool) if tool else _t("agents.between_turns")
-                    part = "      " + _t("agents.child_line", index=i + 1,
-                                         api_calls=child.get("api_calls", "?"), doing=doing)
-                    idle = child.get("seconds_since_activity")
-                    if idle is not None:
-                        part += _t("agents.last_activity", seconds=f"{idle:.0f}")
-                    _cp(part)
+        if view is None:
+            _print_local_delegations()
+        elif n_async := view.active_delegations():
+            _cp(f"  {_t('agents.background_delegations', count=n_async)}")
         agent_running = getattr(self, "_agent_running", False)
         _cp(f"  {_t('agents.agent_running') if agent_running else _t('agents.agent_idle')}")
 
@@ -307,6 +291,40 @@ class CLICommandsSessionToolsMixin:
                    f"  {_t('profile.home', home=reply.data['home'])}")
 
 
+def _print_local_delegations() -> None:
+    """This process's background (async) delegations — delegate_task(background=true)."""
+    delegations = _probe("tools.async_delegation", "list_async_delegations", [])
+    if not delegations:
+        return
+    running_d = [d for d in delegations if d.get("status") in ("running", "stalling")]
+    _cp(f"  {_t('agents.background_delegations', count=len(running_d))}")
+    for d in delegations:
+        status = d.get("status", "?")
+        line = f"    {d.get('delegation_id', '?')} · {status} · {(d.get('goal') or '')[:60]}"
+        # Live-status detail for in-flight delegations.
+        # See #51690.
+        if status == "stalling":
+            quiet = d.get("stalled_after_quiet_seconds")
+            if quiet is not None:
+                line += _t("agents.no_progress", seconds=f"{quiet:.0f}")
+        elif status == "running":
+            quiet = d.get("seconds_since_progress")
+            if quiet is not None and quiet >= 60:
+                line += _t("agents.quiet", seconds=f"{quiet:.0f}")
+        _cp(line)
+        for i, child in enumerate(d.get("children_activity") or []):
+            if not isinstance(child, dict):
+                continue
+            tool = child.get("current_tool")
+            doing = _t("agents.in_tool", tool=tool) if tool else _t("agents.between_turns")
+            part = "      " + _t("agents.child_line", index=i + 1,
+                                 api_calls=child.get("api_calls", "?"), doing=doing)
+            idle = child.get("seconds_since_activity")
+            if idle is not None:
+                part += _t("agents.last_activity", seconds=f"{idle:.0f}")
+            _cp(part)
+
+
 class _TTYBuf(StringIO):
     """StringIO that claims to be a TTY so ``hermes_cli.colors.color()`` still emits ANSI escapes."""
     def isatty(self) -> bool:

hermes_cli/cli_info_mixin.py

@@ -493,6 +493,11 @@ class CLIInfoMixin:
         ``CommandDef`` entries declare ``busy_policy="dispatch"``; the classic CLI honours it here)."""
         return self._busy_inline_command(text, has_images, ("bg", "btw"))
 
+    def _should_handle_stop_command_inline(self, text: str, has_images: bool = False) -> bool:
+        """Return True when /stop should be dispatched while the agent runs: its ``CommandDef`` declares
+        ``busy_policy="interrupt_then_dispatch"``, and queued it would only run after the turn it stops."""
+        return self._busy_inline_command(text, has_images, ("stop",))
+
     def handle_bang_shell(self, text: str) -> bool:
         """Run a ``!<command>`` submission. Returns True when it was handled.
 

hermes_cli/cli_tui_mixin.py

@@ -1491,9 +1491,9 @@ class CLITuiMixin:
                 self._should_exit = True
                 if event.app.is_running:
                     event.app.exit()
-        elif (
-            self._should_handle_steer_command_inline(text, has_images=has_images)
-            or self._should_handle_background_command_inline(text, has_images=has_images)):
+        elif (self._should_handle_steer_command_inline(text, has_images=has_images)
+              or self._should_handle_background_command_inline(text, has_images=has_images)
+              or self._should_handle_stop_command_inline(text, has_images=has_images)):
             self.process_command(text)
         else:
             return False

hermes_cli/gateway_cli_commands.py

@@ -0,0 +1,65 @@
+"""Classic CLI commands whose state lives in the gateway session, answered there.
+
+Rewired, the session's tools, checkpoints, environment and handoff watcher run in the gateway, so
+main's handlers (which read this process's own agent and environment) answer nothing or the wrong
+thing:
+
+- ``/diff session``: the checkpoint baseline is the gateway tools'; its ``/diff`` handler (a reviewed
+  ``slash.exec`` read, run in this session's cwd) renders it. Working-tree modes stay main's local
+  git read of the same checkout.
+- ``/reload``: ``reload.env`` re-reads ``.env`` where turns and tools run (this process re-reads too).
+- ``/handoff``: no gateway verb carries a handoff yet; refused as Ink and Desktop refuse it, instead
+  of the store's ``request_handoff`` refusal dying silently in the process loop.
+"""
+from __future__ import annotations
+
+from agent.i18n import t
+from hermes_cli.gateway_client import GatewayClientError
+
+
+def _diff(cli, connection, cmd):
+    args = cmd.partition(" ")[2].strip()
+    if "session" not in args.lower().split():
+        return False
+    from cli import _cprint
+    reply = connection.call("slash.exec", session_id=cli.session_id, command=f"diff {args}")
+    # The messaging handler fences its blocks for chat; a terminal shows the text itself.
+    for line in str((reply or {}).get("output") or "").splitlines():
+        if not line.startswith("```"):
+            _cprint(f"  {line}" if line else "")
+    return True
+
+
+def _reload(cli, connection, cmd):
+    from hermes_cli.config import reload_env
+    reload_env()
+    updated = (connection.call("reload.env", session_id=cli.session_id) or {}).get("updated", 0)
+    print(f"  {t('cli.reload.env_reloaded', count=updated)}")
+    return True
+
+
+def _handoff(cli, connection, cmd):
+    from cli import _cprint
+    from hermes_cli.gateway_chat_commands import refusal
+    _cprint(f"  {refusal('handoff')}")
+    return True
+
+
+_ROUTES = {"diff": _diff, "reload": _reload, "handoff": _handoff}
+
+
+def run(cli, canonical, cmd):
+    """True when the gateway answered the command (the local handler must not run)."""
+    route = _ROUTES.get(canonical)
+    if route is None:
+        return False
+    from hermes_cli.gateway_side import connection_for
+    connection = connection_for(cli)
+    if connection is None:
+        return False
+    try:
+        return route(cli, connection, (cmd or "").strip())
+    except GatewayClientError as exc:
+        from cli import _cprint
+        _cprint(f"  /{canonical}: {exc}")
+        return True

tests/agent/test_insights.py

@@ -328,9 +328,9 @@ class TestInsightsPopulated:
     _INDEX = "idx_messages_assistant_calls_by_session"
     _PINNED_QUERIES = (
         ("_GET_TOOL_CALLS_ALL", (0.0,)),
-        ("_GET_TOOL_CALLS_WITH_SOURCE", (0.0, "cli")),
+        ("_GET_TOOL_CALLS_WITH_SOURCE", (0.0, "cli", "cli")),  # label, then created_source of a local row
         ("_GET_SKILL_CALLS_ALL", (0.0,)),
-        ("_GET_SKILL_CALLS_WITH_SOURCE", (0.0, "cli")),
+        ("_GET_SKILL_CALLS_WITH_SOURCE", (0.0, "cli", "cli")),
     )
 
     def test_assistant_call_queries_use_partial_index_without_analyze(
@@ -589,3 +589,31 @@ class TestEdgeCases:
         assert "~$0.0046" in terminal_text
         assert "~$0.00 estimated" not in gateway_text
         assert "~$0.0046 estimated" in gateway_text
+
+
+def _label(db, session_id, source, created_source):
+    db.create_session(session_id=session_id, source=source, model="model-a")
+    db.append_message(session_id, role="user", content="hi")
+    db._conn.execute("UPDATE sessions SET created_source = ? WHERE id = ?", (created_source, session_id))
+    db._conn.commit()
+
+
+def test_source_filter_includes_gateway_local_rows_that_surface_created(db):
+    """The gateway stores interactive CLI/TUI rows as ``local`` (``created_source`` keeps the
+    surface), so ``/insights --source cli`` reads both labels (tracker #134889 CLI-23)."""
+    _label(db, "legacy-cli", "cli", "cli")
+    _label(db, "local-cli", "local", "cli")
+    _label(db, "local-tui", "local", "tui")
+    _label(db, "tg", "telegram", "telegram")
+    engine = InsightsEngine(db)
+    assert engine.generate(days=30, source="cli")["overview"]["total_sessions"] == 2
+    assert engine.generate(days=30, source="tui")["overview"]["total_sessions"] == 1
+    assert engine.generate(days=30, source="telegram")["overview"]["total_sessions"] == 1
+
+
+def test_source_filter_on_a_db_without_created_source(db):
+    """A DB an older writer left behind (read-only opens skip migrations) still filters by label."""
+    _label(db, "legacy-cli", "cli", "cli")
+    db._conn.execute("ALTER TABLE sessions DROP COLUMN created_source")
+    db._conn.commit()
+    assert InsightsEngine(db).generate(days=30, source="cli")["overview"]["total_sessions"] == 1

tests/agent/test_turn_facade_lease.py

@@ -121,6 +121,24 @@ def test_timeout_and_interrupt_early_results():
     assert admission.early_result["interrupt_message"] == "stop"
 
 
+def test_interrupt_before_any_wait_does_not_blame_another_process():
+    """Ctrl+C right after submit lands before the first lease attempt (tracker #134889 N4): nothing
+    else held the session, so the reply must not say it stopped waiting for another process."""
+    agent = _agent(_Db(acquired=False), _interrupt_requested=True)
+    agent.clear_interrupt = lambda: None
+    reply = _admit(agent).early_result["final_response"]
+    assert "another Hermes process" not in reply and "not processed" in reply
+
+    class _HeldDb(_Db):
+        def acquire_session_turn_lease(self, session_id, holder, **kwargs):
+            kwargs["on_wait"](0.5)  # another holder seen, then the user interrupted the wait
+            return False
+
+    agent = _agent(_HeldDb(), _interrupt_requested=True)
+    agent.clear_interrupt = lambda: None
+    assert "another Hermes process" in _admit(agent).early_result["final_response"]
+
+
 def _active_lease(db):
     agent = _agent(db)
     calls = []

tests/hermes_cli/test_gateway_cli_commands.py

@@ -0,0 +1,66 @@
+"""Classic CLI commands whose state lives in the gateway session are answered there (tracker #134889
+CLI-8, CLI-17, CLI-21): ``/diff session`` (the gateway tools' checkpoint baseline), ``/reload``
+(the gateway's ``.env``) and ``/handoff`` (no gateway verb yet: refused, never silent). Driven
+through ``HermesCLI.process_command`` with a scripted gateway connection."""
+from types import SimpleNamespace
+
+import pytest
+
+from cli import HermesCLI
+
+
+class _Conn:
+    closed = False
+
+    def __init__(self, replies):
+        self.replies = replies
+        self.calls = []
+
+    def call(self, method, **params):
+        self.calls.append((method, params))
+        return self.replies.get(method, {})
+
+
+@pytest.fixture
+def gateway_cli(monkeypatch):
+    def make(**replies):
+        conn = _Conn(replies)
+        monkeypatch.setattr("hermes_cli.gateway_side.connection_for", lambda cli: conn)
+        cli = HermesCLI.__new__(HermesCLI)
+        cli.session_id = "local-s"
+        # The gateway-backed agent has no checkpoint manager of its own (GatewayAgent._checkpoint_mgr).
+        cli.agent = SimpleNamespace(_checkpoint_mgr=None)
+        return cli, conn
+    return make
+
+
+def test_diff_session_renders_the_gateway_sessions_checkpoint_diff(gateway_cli, capsys):
+    cli, conn = gateway_cli(**{"slash.exec": {"type": "exec", "output": "```\n a.py | 2 +-\n```"}})
+    assert cli.process_command("/diff session --stat") is True
+    assert conn.calls == [("slash.exec", {"session_id": "local-s", "command": "diff session --stat"})]
+    out = capsys.readouterr().out
+    assert "a.py | 2 +-" in out and "```" not in out
+
+
+def test_diff_of_the_working_tree_stays_a_local_git_read(gateway_cli, monkeypatch):
+    cli, conn = gateway_cli()
+    seen = []
+    monkeypatch.setattr("tools.working_diff.collect_working_diff",
+                        lambda cwd, mode, paths=None: seen.append(mode) or {"success": True, "empty": True})
+    cli.process_command("/diff staged")
+    assert seen == ["staged"] and conn.calls == []
+
+
+def test_reload_rereads_env_in_the_gateway_and_reports_its_count(gateway_cli, monkeypatch, capsys):
+    cli, conn = gateway_cli(**{"reload.env": {"updated": 3}})
+    monkeypatch.setattr("hermes_cli.config.reload_env", lambda: 0)
+    assert cli.process_command("/reload") is True
+    assert conn.calls == [("reload.env", {"session_id": "local-s"})]
+    assert "3" in capsys.readouterr().out
+
+
+def test_handoff_is_refused_with_a_message_on_the_gateway(gateway_cli, capsys):
+    cli, conn = gateway_cli()
+    assert cli.process_command("/handoff telegram") is True  # the session stays
+    assert "/handoff is not available on the shared gateway yet" in capsys.readouterr().out
+    assert conn.calls == []

tests/hermes_cli/test_gateway_tools.py

@@ -97,6 +97,41 @@ def test_stop_kills_the_session_processes_through_the_gateway(gateway, capsys):
     assert any(method == "process.stop" for method, _ in client.calls)
 
 
+def test_agents_lists_the_gateway_sessions_processes_and_subagents(gateway, capsys):
+    """/agents reads where the work runs (tracker #134889 CLI-7): the session's processes and its
+    running background subagents, not this process's empty registries."""
+    from hermes_cli.cli_commands_session_tools import CLICommandsSessionToolsMixin
+    agent, client = gateway
+    agent.connection.state(SID).usage = {"active_subagents": 2}
+    CLICommandsSessionToolsMixin._handle_agents_command(object())
+    out = capsys.readouterr().out
+    assert "proc_1" in out and "sleep 300" in out
+    assert "Background delegations: 2" in out
+    assert ("process.list", {"session_id": SID}) in client.calls
+
+
+def test_stop_typed_mid_turn_interrupts_the_turn_then_stops_processes(gateway, capsys):
+    """/stop typed while a turn runs is dispatched inline and interrupts it like Ctrl+C (tracker
+    #134889 N4), instead of being queued until the turn it should stop has finished."""
+    from types import SimpleNamespace
+    from cli import HermesCLI
+    _agent, client = gateway
+    stops = []
+
+    class _Agent:
+        def hard_interrupt(self, message=None):
+            stops.append(message)
+
+    cli = HermesCLI.__new__(HermesCLI)
+    cli._agent_running, cli.agent = True, _Agent()
+    buffer = SimpleNamespace(reset=lambda **_k: None)
+    event = SimpleNamespace(app=SimpleNamespace(current_buffer=buffer, invalidate=lambda: None))
+    assert cli._tui_enter_inline_command(event, "/stop", False) is True
+    assert stops == [None]
+    assert any(method == "process.stop" for method, _ in client.calls)
+    assert "Interrupting agent" in capsys.readouterr().out
+
+
 def test_dock_stop_kills_one_process_through_the_gateway(gateway):
     from hermes_cli import cli_process_dock as dock
     _agent, client = gateway

tests/tools/test_bot_live_delivery_mailbox_dir.py

@@ -0,0 +1,29 @@
+"""The live-delivery mailbox lock dir (``tools.bot_live_delivery._locked``).
+
+POSIX tightens the directory to 0700 on every entry. On Windows a chmod rewrites the NTFS ACL and
+the owner then gets Permission denied opening ``.lock`` (the Telegram adapter crash-looped on it,
+tracker #134889 W2), so the lock must stay re-enterable there.
+"""
+import stat
+
+import pytest
+
+from tools.bot_live_delivery import _locked, _root
+
+
+@pytest.mark.platforms("posix")
+def test_mailbox_dir_is_tightened_to_owner_only(tmp_path):
+    root = _root(tmp_path)
+    root.mkdir(parents=True, mode=0o755)
+    root.chmod(0o755)
+    with _locked(tmp_path):
+        pass
+    assert stat.S_IMODE(root.stat().st_mode) == 0o700
+
+
+@pytest.mark.platforms("windows")
+def test_mailbox_lock_is_reenterable_on_windows(tmp_path):
+    for _ in range(3):  # the live poller re-enters it twice a second
+        with _locked(tmp_path):
+            pass
+    (_root(tmp_path) / ".lock").open("a").close()

tools/bot_live_delivery.py

@@ -121,7 +121,10 @@ def _locked(home: Path | str):
     created = not root.is_dir()
     root.parent.mkdir(parents=True, exist_ok=True)
     root.mkdir(mode=0o700, exist_ok=True)
-    root.chmod(0o700)
+    if os.name != "nt":
+        # NTFS has no POSIX mode: chmod rewrites the directory ACL, and the owner then gets
+        # Permission denied opening ``.lock`` below (the Telegram adapter crash-looped on it).
+        root.chmod(0o700)
     if created:
         # Only a fresh mailbox dir needs its parents durably linked; the live
         # poller re-enters this lock twice a second per profile, and two