feat(gateway): per-platform admin/user split for slash commands (salvage of #4443) (#23373)

* feat(gateway): per-platform admin/user split for slash commands

Adds an opt-in two-list access control on top of the existing per-platform
`allow_from` allowlists, scoped to slash commands only:

  - allow_admin_from         — full slash command access
  - user_allowed_commands    — what non-admins may run
  - group_allow_admin_from   — same, group/channel scope
  - group_user_allowed_commands

When `allow_admin_from` is unset for a scope, gating is disabled and every
allowed user keeps full access (backward compat). Plain chat is unaffected.
`/help` and `/whoami` are always reachable so users can see what they
can run.

Gate runs at the slash command dispatch site in gateway/run.py and uses
`is_gateway_known_command()`, so it covers built-in AND plugin-registered
commands through the live registry without per-feature wiring.

Adds `/whoami` showing platform, scope, tier, and runnable commands.

Salvage of PR #4443's permission tier work, scoped down. The full tier
system, tool filtering, audit log, usage tracking, rate limiting,
`/promote` flow, and persistent SQLite stores are not included here —
those can be re-expanded later if needed.

Co-authored-by: ReqX <mike@grossmann.at>

* fix(gateway): close running-agent fast-path bypass + add coverage and central docs

The slash command access gate was only applied at the cold dispatch site
(line ~5921). When an agent was already running, the running-agent
fast-path block (line ~5574) dispatched /restart, /stop, /new, /steer,
/model, /approve, /deny, /agents, /background, /kanban, /goal, /yolo,
/verbose, /footer, /help, /commands, /profile, /update directly
without going through the gate — letting non-admins bypass gating just
because an agent happens to be busy.

Refactored the gate into _check_slash_access() and called from BOTH
paths. /status remains intentionally pre-gate so users can always see
session state.

Also added 18 more dispatch tests covering:
  - Running-agent fast-path: blocks non-admin, allows admin, /status
    always works
  - Alias canonicalization (gate uses canonical name, not user alias)
  - Unknown / unregistered commands pass through (don't false-positive)
  - DM admin scope-locked when group has its own admin list
  - Multi-platform isolation (Discord gated, Telegram unrestricted)

Docs: added Slash Command Access Control section to the central
messaging index page + /whoami row in the chat commands table.

Co-authored-by: ReqX <mike@grossmann.at>

---------

Co-authored-by: ReqX <mike@grossmann.at>

tests/gateway/test_slash_access.py

@@ -0,0 +1,289 @@
+"""Unit tests for gateway.slash_access — per-platform slash command access control.
+
+Tests the pure policy resolver (no gateway plumbing). Integration tests that
+exercise the dispatch site live in test_slash_access_dispatch.py.
+"""
+from __future__ import annotations
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.session import SessionSource
+from gateway.slash_access import (
+    SlashAccessPolicy,
+    policy_for_source,
+    policy_from_extra,
+)
+
+
+# ---------------------------------------------------------------------------
+# policy_from_extra — input normalization + scope resolution
+# ---------------------------------------------------------------------------
+
+
+class TestPolicyFromExtra:
+    def test_empty_extra_is_disabled(self):
+        p = policy_from_extra({}, "dm")
+        assert p.enabled is False
+        assert p.admin_user_ids == frozenset()
+        assert p.user_allowed_commands == frozenset()
+
+    def test_disabled_policy_treats_anyone_as_admin(self):
+        # When gating is off, downstream code uses is_admin/can_run uniformly.
+        # Both must short-circuit to True so existing behavior is preserved.
+        p = policy_from_extra({}, "dm")
+        assert p.is_admin("anyone") is True
+        assert p.can_run("anyone", "stop") is True
+
+    def test_dm_admin_list_only(self):
+        p = policy_from_extra({"allow_admin_from": ["111", "222"]}, "dm")
+        assert p.enabled is True
+        assert p.admin_user_ids == frozenset({"111", "222"})
+        assert p.user_allowed_commands == frozenset()
+
+    def test_admin_runs_anything(self):
+        p = policy_from_extra(
+            {"allow_admin_from": [111], "user_allowed_commands": ["help"]},
+            "dm",
+        )
+        assert p.is_admin("111") is True
+        assert p.can_run("111", "stop") is True
+        assert p.can_run("111", "kanban") is True
+
+    def test_non_admin_runs_only_listed_commands(self):
+        p = policy_from_extra(
+            {
+                "allow_admin_from": ["111"],
+                "user_allowed_commands": ["status", "model"],
+            },
+            "dm",
+        )
+        assert p.is_admin("999") is False
+        assert p.can_run("999", "status") is True
+        assert p.can_run("999", "model") is True
+        assert p.can_run("999", "stop") is False
+        assert p.can_run("999", "kanban") is False
+
+    def test_always_allowed_floor_for_non_admin(self):
+        # /help and /whoami always reachable so users can see what they can do.
+        p = policy_from_extra(
+            {"allow_admin_from": ["111"], "user_allowed_commands": []},
+            "dm",
+        )
+        assert p.can_run("999", "help") is True
+        assert p.can_run("999", "whoami") is True
+        assert p.can_run("999", "stop") is False
+
+    def test_unknown_user_id_blocked(self):
+        # Empty/None user_id → no admin status, no command access (except floor).
+        p = policy_from_extra(
+            {"allow_admin_from": ["111"], "user_allowed_commands": ["status"]},
+            "dm",
+        )
+        assert p.is_admin(None) is False
+        assert p.can_run(None, "status") is True  # listed command works
+        assert p.can_run(None, "stop") is False
+        assert p.can_run("", "stop") is False
+
+    def test_id_coercion_ints_become_strings(self):
+        # YAML often loads numeric IDs as ints; we stringify on ingest.
+        p = policy_from_extra({"allow_admin_from": [12345, 67890]}, "dm")
+        assert p.admin_user_ids == frozenset({"12345", "67890"})
+        assert p.is_admin("12345") is True
+        assert p.is_admin(12345) is True  # is_admin also stringifies
+
+    def test_id_coercion_csv_string(self):
+        p = policy_from_extra({"allow_admin_from": "111, 222 ,333"}, "dm")
+        assert p.admin_user_ids == frozenset({"111", "222", "333"})
+
+    def test_command_coercion_strips_leading_slash_and_lowercases(self):
+        p = policy_from_extra(
+            {
+                "allow_admin_from": ["111"],
+                "user_allowed_commands": ["/Status", "MODEL", "/help"],
+            },
+            "dm",
+        )
+        assert p.user_allowed_commands == frozenset({"status", "model", "help"})
+
+    def test_command_coercion_csv_string(self):
+        p = policy_from_extra(
+            {
+                "allow_admin_from": ["111"],
+                "user_allowed_commands": "status, model , /help",
+            },
+            "dm",
+        )
+        assert p.user_allowed_commands == frozenset({"status", "model", "help"})
+
+    def test_group_scope_uses_group_keys(self):
+        extra = {
+            "allow_admin_from": ["111"],          # DM admins
+            "user_allowed_commands": ["status"],  # DM commands
+            "group_allow_admin_from": ["222"],
+            "group_user_allowed_commands": ["help"],
+        }
+        dm = policy_from_extra(extra, "dm")
+        gp = policy_from_extra(extra, "group")
+        assert dm.admin_user_ids == frozenset({"111"})
+        assert gp.admin_user_ids == frozenset({"222"})
+        assert dm.user_allowed_commands == frozenset({"status"})
+        # group's user_allowed_commands does not leak into DM's allowed list
+        # except via the explicit fallback rule (only when DM list is unset).
+        assert "help" in gp.user_allowed_commands
+
+    def test_dm_falls_back_to_group_user_commands_when_dm_unset(self):
+        # Common case: operator wants the same command set DM and group;
+        # they should only have to list it once on the group keys.
+        extra = {
+            "allow_admin_from": ["111"],
+            "group_user_allowed_commands": ["status", "model"],
+        }
+        dm = policy_from_extra(extra, "dm")
+        assert dm.user_allowed_commands == frozenset({"status", "model"})
+
+    def test_dm_admin_does_not_imply_group_admin(self):
+        # Admin lists are scope-specific. DM admin must not auto-promote in groups.
+        extra = {"allow_admin_from": ["111"]}
+        dm = policy_from_extra(extra, "dm")
+        gp = policy_from_extra(extra, "group")
+        assert dm.is_admin("111") is True
+        # Group has no admin list set → gating disabled in groups → "111"
+        # gets unrestricted access, but that's the backward-compat fallback,
+        # not implicit admin promotion. The distinction matters when the
+        # group DOES have an admin list set:
+        extra2 = {
+            "allow_admin_from": ["111"],
+            "group_allow_admin_from": ["222"],
+        }
+        gp2 = policy_from_extra(extra2, "group")
+        assert gp2.is_admin("111") is False
+        assert gp2.is_admin("222") is True
+
+
+# ---------------------------------------------------------------------------
+# policy_for_source — wires GatewayConfig + SessionSource together
+# ---------------------------------------------------------------------------
+
+
+class TestPolicyForSource:
+    def test_no_config_returns_disabled(self):
+        p = policy_for_source(None, None)
+        assert p.enabled is False
+        assert p.is_admin("anyone") is True
+
+    def test_no_platform_config_returns_disabled(self):
+        cfg = GatewayConfig(platforms={})
+        src = SessionSource(
+            platform=Platform.DISCORD, chat_id="42", chat_type="dm", user_id="7"
+        )
+        p = policy_for_source(cfg, src)
+        assert p.enabled is False
+
+    def test_dm_chat_type_resolves_to_dm_scope(self):
+        cfg = GatewayConfig(
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True,
+                    extra={
+                        "allow_admin_from": ["111"],
+                        "user_allowed_commands": ["status"],
+                        "group_allow_admin_from": ["222"],
+                        "group_user_allowed_commands": ["help"],
+                    },
+                )
+            }
+        )
+        dm_src = SessionSource(
+            platform=Platform.DISCORD, chat_id="A", chat_type="dm", user_id="111"
+        )
+        p = policy_for_source(cfg, dm_src)
+        assert p.is_admin("111") is True
+        assert p.can_run("999", "status") is True
+        assert p.can_run("999", "help") is True  # always-allowed floor
+        assert p.can_run("999", "kanban") is False
+
+    def test_group_chat_type_resolves_to_group_scope(self):
+        cfg = GatewayConfig(
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True,
+                    extra={
+                        "allow_admin_from": ["111"],
+                        "user_allowed_commands": ["status"],
+                        "group_allow_admin_from": ["222"],
+                        "group_user_allowed_commands": ["help"],
+                    },
+                )
+            }
+        )
+        grp_src = SessionSource(
+            platform=Platform.DISCORD, chat_id="G", chat_type="group", user_id="222"
+        )
+        p = policy_for_source(cfg, grp_src)
+        assert p.is_admin("222") is True
+        assert p.is_admin("111") is False  # DM admin, not group admin
+        # In group scope, the only listed user command is "help"; "status"
+        # is not in the group list and should be denied for non-admins.
+        assert p.can_run("999", "help") is True
+        assert p.can_run("999", "status") is False
+
+    def test_channel_thread_chat_types_treated_as_group_scope(self):
+        # Discord channels and threads are group-scoped, not DM-scoped.
+        cfg = GatewayConfig(
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True,
+                    extra={
+                        "allow_admin_from": ["111"],
+                        "group_allow_admin_from": ["222"],
+                    },
+                )
+            }
+        )
+        for ct in ("group", "channel", "thread", "supergroup"):
+            src = SessionSource(
+                platform=Platform.DISCORD, chat_id="X", chat_type=ct, user_id="222"
+            )
+            p = policy_for_source(cfg, src)
+            assert p.is_admin("222") is True, f"chat_type={ct} should map to group scope"
+            assert p.is_admin("111") is False, f"chat_type={ct} should not see DM admins"
+
+    def test_no_admin_list_for_dm_means_unrestricted_in_dm(self):
+        # Group has admin list, DM does not → DM gating disabled, group active.
+        cfg = GatewayConfig(
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True,
+                    extra={"group_allow_admin_from": ["222"]},
+                )
+            }
+        )
+        dm_src = SessionSource(
+            platform=Platform.DISCORD, chat_id="A", chat_type="dm", user_id="999"
+        )
+        grp_src = SessionSource(
+            platform=Platform.DISCORD, chat_id="G", chat_type="group", user_id="999"
+        )
+        dm_p = policy_for_source(cfg, dm_src)
+        grp_p = policy_for_source(cfg, grp_src)
+        assert dm_p.enabled is False
+        assert dm_p.can_run("999", "stop") is True  # backward compat
+        assert grp_p.enabled is True
+        assert grp_p.can_run("999", "stop") is False  # gated
+
+    def test_per_platform_isolation(self):
+        # Discord has gating, Telegram doesn't → Telegram is unaffected.
+        cfg = GatewayConfig(
+            platforms={
+                Platform.DISCORD: PlatformConfig(
+                    enabled=True,
+                    extra={"allow_admin_from": ["111"]},
+                ),
+                Platform.TELEGRAM: PlatformConfig(enabled=True, extra={}),
+            }
+        )
+        tg_src = SessionSource(
+            platform=Platform.TELEGRAM, chat_id="T", chat_type="dm", user_id="999"
+        )
+        p = policy_for_source(cfg, tg_src)
+        assert p.enabled is False
+        assert p.can_run("999", "stop") is True

tests/gateway/test_slash_access_dispatch.py

@@ -0,0 +1,558 @@
+"""Integration tests for slash command access control gating in gateway/run.py.
+
+Drives the real ``GatewayRunner._handle_message`` path with a stub session
+store so we exercise the actual gate inserted at the dispatch site (not a
+re-implementation in the test). Uses the same ``object.__new__`` runner
+construction pattern as test_status_command.py.
+
+Coverage targets:
+  - Backward compat: no ``allow_admin_from`` set → behaves exactly as before
+    (no denial messages, dispatch reaches the real handler).
+  - Admin path: user in ``allow_admin_from`` runs anything.
+  - User path: user not in admin list, but command in
+    ``user_allowed_commands`` → allowed.
+  - User denied: command not in either list → returns the ⛔ denial.
+  - Always-allowed floor: /help and /whoami reachable for non-admins
+    even with empty user_allowed_commands.
+  - DM vs group scope isolation.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionEntry, SessionSource, build_session_key
+
+
+def _make_source(
+    *,
+    platform: Platform = Platform.DISCORD,
+    user_id: str = "user1",
+    chat_type: str = "dm",
+    chat_id: str = "c1",
+) -> SessionSource:
+    return SessionSource(
+        platform=platform,
+        user_id=user_id,
+        chat_id=chat_id,
+        user_name=f"name-{user_id}",
+        chat_type=chat_type,
+    )
+
+
+def _make_event(text: str, source: SessionSource) -> MessageEvent:
+    return MessageEvent(text=text, source=source, message_id="m1")
+
+
+def _make_runner(*, platform_extra: dict | None = None,
+                 platform: Platform = Platform.DISCORD):
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(
+        platforms={
+            platform: PlatformConfig(
+                enabled=True,
+                token="***",
+                extra=platform_extra or {},
+            )
+        }
+    )
+    adapter = MagicMock()
+    adapter.send = AsyncMock()
+    runner.adapters = {platform: adapter}
+    runner._voice_mode = {}
+    runner.hooks = SimpleNamespace(
+        emit=AsyncMock(),
+        emit_collect=AsyncMock(return_value=[]),
+        loaded_hooks=False,
+    )
+    runner.session_store = MagicMock()
+    session_entry = SessionEntry(
+        session_key="agent:main:discord:dm:c1",
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=platform,
+        chat_type="dm",
+        total_tokens=0,
+    )
+    runner.session_store.get_or_create_session.return_value = session_entry
+    runner.session_store.load_transcript.return_value = []
+    runner.session_store.has_any_sessions.return_value = True
+    runner.session_store.append_to_transcript = MagicMock()
+    runner.session_store.rewrite_transcript = MagicMock()
+    runner.session_store.update_session = MagicMock()
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._session_run_generation = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._session_sources = {}
+    runner._session_db = MagicMock()
+    runner._session_db.get_session_title.return_value = None
+    runner._session_db.get_session.return_value = None
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._show_reasoning = False
+    runner._is_user_authorized = lambda _source: True
+    runner._set_session_env = lambda _context: None
+    runner._should_send_voice_reply = lambda *_args, **_kwargs: False
+    runner._send_voice_reply = AsyncMock()
+    runner._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None
+    runner._emit_gateway_run_progress = AsyncMock()
+    return runner
+
+
+# ---------------------------------------------------------------------------
+# /whoami response shape — proves the handler is reachable AND uses the
+# resolver. We use /whoami because it's deterministic and short-circuits
+# before any session/agent setup.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_whoami_unrestricted_when_no_admin_list():
+    runner = _make_runner(platform_extra={})  # no admin list
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="999")))
+    assert "Tier: unrestricted" in result
+    assert "no admin list configured" in result
+
+
+@pytest.mark.asyncio
+async def test_whoami_admin_user():
+    runner = _make_runner(platform_extra={"allow_admin_from": ["111"]})
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="111")))
+    assert "**admin**" in result
+
+
+@pytest.mark.asyncio
+async def test_whoami_non_admin_lists_runnable_commands():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": ["status", "model"],
+        }
+    )
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="999")))
+    assert "Tier: user" in result
+    assert "/help" in result      # always-allowed floor
+    assert "/whoami" in result    # always-allowed floor
+    assert "/status" in result
+    assert "/model" in result
+
+
+# ---------------------------------------------------------------------------
+# Gate denial — admin-only command attempted by non-admin
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_non_admin_denied_for_unlisted_command():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": ["status"],
+        }
+    )
+    # /stop is NOT in user_allowed_commands and not in the always-allowed floor.
+    result = await runner._handle_message(_make_event("/stop", _make_source(user_id="999")))
+    assert result is not None
+    assert "⛔" in result
+    assert "/stop is admin-only here" in result
+    assert "/status" in result  # denial preview shows what they CAN run
+
+
+@pytest.mark.asyncio
+async def test_non_admin_with_empty_user_commands_gets_floor_only():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],  # explicitly empty
+        }
+    )
+    # /stop denied
+    result = await runner._handle_message(_make_event("/stop", _make_source(user_id="999")))
+    assert "⛔" in result
+    assert "No slash commands are enabled" in result
+    # /whoami still works (always-allowed floor)
+    whoami_result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="999")))
+    assert "Tier: user" in whoami_result
+
+
+# ---------------------------------------------------------------------------
+# Gate ALLOW — admin and listed user
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_admin_runs_unlisted_command():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],  # users can run nothing
+        }
+    )
+    # Admin runs /whoami (proxy for "any command works"); the gate must NOT
+    # return the ⛔ denial. The /whoami handler is deterministic and doesn't
+    # need a real agent, so we can assert against its content.
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="111")))
+    assert "⛔" not in result
+    assert "**admin**" in result
+
+
+@pytest.mark.asyncio
+async def test_user_runs_listed_command():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": ["whoami"],  # explicit
+        }
+    )
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="999")))
+    assert "⛔" not in result
+    assert "Tier: user" in result
+
+
+# ---------------------------------------------------------------------------
+# Backward compatibility — no admin list set means no gating at all
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backward_compat_no_admin_list_means_no_gate():
+    runner = _make_runner(platform_extra={})  # nothing configured
+    # Random non-listed user runs /whoami; should return unrestricted profile,
+    # never a denial.
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="anyone")))
+    assert "⛔" not in result
+    assert "Tier: unrestricted" in result
+
+
+# ---------------------------------------------------------------------------
+# Scope isolation — DM vs group
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_dm_admin_is_not_group_admin():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "group_allow_admin_from": ["222"],
+            "group_user_allowed_commands": [],
+        }
+    )
+    # User 111 is DM admin. In group context they're a non-admin with no
+    # listed commands → /stop denied.
+    result = await runner._handle_message(
+        _make_event("/stop", _make_source(user_id="111", chat_type="group"))
+    )
+    assert "⛔" in result
+
+
+@pytest.mark.asyncio
+async def test_group_only_gating_leaves_dm_unrestricted():
+    runner = _make_runner(
+        platform_extra={
+            # Only group has an admin list → DM scope stays in backward-compat mode
+            "group_allow_admin_from": ["222"],
+        }
+    )
+    result = await runner._handle_message(_make_event("/whoami", _make_source(user_id="anyone", chat_type="dm")))
+    assert "Tier: unrestricted" in result
+
+
+# ---------------------------------------------------------------------------
+# Plugin-registered slash commands are gated through the same path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_plugin_registered_command_is_gated(monkeypatch):
+    """The gate must recognize plugin-registered slash commands, not just
+    built-in COMMAND_REGISTRY entries. We verify by stubbing
+    is_gateway_known_command and resolve_command so a fictitious /myplugin
+    command is treated as a known plugin command.
+    """
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],
+        }
+    )
+
+    from hermes_cli import commands as cmd_mod
+
+    real_resolve = cmd_mod.resolve_command
+    real_is_known = cmd_mod.is_gateway_known_command
+
+    def fake_resolve(name):
+        if name == "myplugin":
+            # Return a CommandDef-like duck so canonical resolution succeeds
+            return SimpleNamespace(name="myplugin")
+        return real_resolve(name)
+
+    def fake_is_known(name):
+        if name == "myplugin":
+            return True
+        return real_is_known(name)
+
+    monkeypatch.setattr(cmd_mod, "resolve_command", fake_resolve)
+    monkeypatch.setattr(cmd_mod, "is_gateway_known_command", fake_is_known)
+
+    # Non-admin tries to run the plugin command → must be denied by the gate.
+    result = await runner._handle_message(
+        _make_event("/myplugin foo bar", _make_source(user_id="999"))
+    )
+    assert "⛔" in result
+    assert "/myplugin is admin-only here" in result
+
+
+# ---------------------------------------------------------------------------
+# Running-agent fast-path gating — admin/user split must hold even when an
+# agent is already running. The fast-path block in _handle_message dispatches
+# /stop, /restart, /new, /steer, /model, /approve, /deny, /agents,
+# /background, /kanban, /goal, /yolo, /verbose, /footer, /help, /commands,
+# /profile, /update directly without going through the cold dispatch site.
+# We must apply the gate there too — otherwise non-admins could bypass
+# gating just because an agent happens to be busy.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_running_agent_fastpath_blocks_non_admin_command():
+    """When an agent is running, /restart from a non-admin must be denied."""
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],
+        }
+    )
+    src = _make_source(user_id="999")
+    # Mark the session as having an in-flight agent so the fast-path runs.
+    from gateway.session import build_session_key
+    sk = build_session_key(src)
+    runner._running_agents[sk] = MagicMock()
+    runner._running_agents_ts[sk] = 0  # not stale (epoch + small delta on this machine)
+
+    result = await runner._handle_message(_make_event("/restart", src))
+    assert result is not None
+    assert "⛔" in result
+    assert "/restart is admin-only here" in result
+
+
+@pytest.mark.asyncio
+async def test_running_agent_fastpath_allows_admin_command():
+    """Admins must still be able to run privileged commands like /restart
+    through the running-agent fast-path. We check that we don't get the
+    denial message; the actual /restart handler is mocked out via the
+    runner's MagicMock."""
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],
+        }
+    )
+    src = _make_source(user_id="111")  # admin
+    from gateway.session import build_session_key
+    sk = build_session_key(src)
+    runner._running_agents[sk] = MagicMock()
+    runner._running_agents_ts[sk] = 0
+    # Mock the restart handler so it doesn't actually try to restart anything.
+    runner._handle_restart_command = AsyncMock(return_value="restart-handled")
+
+    result = await runner._handle_message(_make_event("/restart", src))
+    assert result == "restart-handled"
+    assert "⛔" not in (result or "")
+
+
+@pytest.mark.asyncio
+async def test_running_agent_fastpath_status_always_works():
+    """/status is intentionally pre-gate on the fast-path so users can
+    always see session state, even non-admins."""
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],
+        }
+    )
+    src = _make_source(user_id="999")  # non-admin
+    from gateway.session import build_session_key
+    sk = build_session_key(src)
+    runner._running_agents[sk] = MagicMock()
+    runner._running_agents_ts[sk] = 0
+    runner._handle_status_command = AsyncMock(return_value="status-handled")
+
+    result = await runner._handle_message(_make_event("/status", src))
+    assert result == "status-handled"
+    assert "⛔" not in (result or "")
+
+
+# ---------------------------------------------------------------------------
+# Alias resolution — /h aliases to /help; the gate must canonicalize before
+# checking access. /hist (history alias) is a real one to exercise.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_gate_uses_canonical_name_not_alias():
+    """If /hist resolves to canonical 'history' and history is in
+    user_allowed_commands, the alias must be allowed too."""
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": ["history"],
+        }
+    )
+    # Find a real alias in the registry to use.
+    from hermes_cli.commands import COMMAND_REGISTRY
+    history_def = next(c for c in COMMAND_REGISTRY if c.name == "history")
+    # If /history has aliases, use one. Otherwise just use /history.
+    alias = history_def.aliases[0] if history_def.aliases else "history"
+    # Mock the history handler so we don't need real session state.
+    runner._handle_history_command = AsyncMock(return_value="history-handled")
+    result = await runner._handle_message(_make_event(f"/{alias}", _make_source(user_id="999")))
+    assert "⛔" not in (result or "")
+
+
+# ---------------------------------------------------------------------------
+# Unknown / unregistered command — gate must NOT intercept (let the existing
+# unknown-command path handle it normally).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_gate_does_not_intercept_unknown_command():
+    """Random non-command text like /xyzzy is not in the registry. The gate
+    must not produce a denial message — the existing unknown-command path
+    will handle it (or the agent will see it as plain text)."""
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],
+            "user_allowed_commands": [],
+        }
+    )
+    # /xyzzy is not in COMMAND_REGISTRY and not a plugin command.
+    # The gate should pass through (no ⛔) since canonical resolution
+    # returns the raw command and is_gateway_known_command returns False.
+    # We can only verify the gate didn't fire — downstream behavior may
+    # vary (returns None, agent processes it, etc.). What matters: no denial.
+    runner._handle_unknown_command = AsyncMock(return_value=None)
+    # Stub out the rest of the cold path to short-circuit
+    runner.session_store.get_or_create_session.side_effect = RuntimeError("would have proceeded past gate")
+    try:
+        await runner._handle_message(_make_event("/xyzzy", _make_source(user_id="999")))
+    except RuntimeError as e:
+        # Reaching session creation means we got past the gate without a denial.
+        assert "would have proceeded past gate" in str(e)
+
+
+# ---------------------------------------------------------------------------
+# Scope independence — admin in DM scope is NOT auto-admin in group when
+# group has its own admin list (regression guard for the "admin lists are
+# scope-specific" rule).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_dm_admin_blocked_in_group_with_separate_admin_list():
+    runner = _make_runner(
+        platform_extra={
+            "allow_admin_from": ["111"],          # DM admin
+            "group_allow_admin_from": ["222"],    # group admin
+            "group_user_allowed_commands": ["status"],
+        }
+    )
+    # User 111 is DM admin. In a group, they're a non-admin and can only
+    # run group_user_allowed_commands. /restart is not in that list → denied.
+    grp_src = _make_source(user_id="111", chat_type="group", chat_id="g1")
+    result = await runner._handle_message(_make_event("/restart", grp_src))
+    assert "⛔" in result
+    assert "/restart is admin-only here" in result
+
+
+# ---------------------------------------------------------------------------
+# Multi-platform isolation — gating on Discord doesn't leak to Telegram.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_gating_isolated_per_platform():
+    """When Discord is gated and Telegram isn't, the same user_id on
+    Telegram must be unrestricted."""
+    from gateway.run import GatewayRunner
+    from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(
+        platforms={
+            Platform.DISCORD: PlatformConfig(
+                enabled=True,
+                token="***",
+                extra={
+                    "allow_admin_from": ["111"],
+                    "user_allowed_commands": [],
+                },
+            ),
+            Platform.TELEGRAM: PlatformConfig(
+                enabled=True, token="***", extra={}
+            ),
+        }
+    )
+    runner.adapters = {
+        Platform.DISCORD: MagicMock(send=AsyncMock()),
+        Platform.TELEGRAM: MagicMock(send=AsyncMock()),
+    }
+    runner._voice_mode = {}
+    runner.hooks = SimpleNamespace(
+        emit=AsyncMock(),
+        emit_collect=AsyncMock(return_value=[]),
+        loaded_hooks=False,
+    )
+    runner.session_store = MagicMock()
+    session_entry = SessionEntry(
+        session_key="agent:main:telegram:dm:c1",
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        total_tokens=0,
+    )
+    runner.session_store.get_or_create_session.return_value = session_entry
+    runner.session_store.load_transcript.return_value = []
+    runner.session_store.has_any_sessions.return_value = True
+    runner.session_store.append_to_transcript = MagicMock()
+    runner.session_store.rewrite_transcript = MagicMock()
+    runner.session_store.update_session = MagicMock()
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._session_run_generation = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._session_sources = {}
+    runner._session_db = MagicMock()
+    runner._session_db.get_session_title.return_value = None
+    runner._session_db.get_session.return_value = None
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._show_reasoning = False
+    runner._is_user_authorized = lambda _source: True
+    runner._set_session_env = lambda _context: None
+    runner._should_send_voice_reply = lambda *_args, **_kwargs: False
+    runner._send_voice_reply = AsyncMock()
+    runner._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None
+    runner._emit_gateway_run_progress = AsyncMock()
+
+    # Same user_id on Telegram → must be unrestricted (Telegram has no admin list).
+    tg_src = _make_source(platform=Platform.TELEGRAM, user_id="999", chat_id="t1")
+    result = await runner._handle_message(_make_event("/whoami", tg_src))
+    assert "Tier: unrestricted" in result