GatewayConfig missing dict-like .get() method bug

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #1287 fix(gateway): avoid slash-command crash with GatewayConfig

  merged to main

  files: gateway/config.py, gateway/run.py, tests/gateway/test_config.py, tests/test_quick_commands.py

  Show the diff

  • gateway/config.py

    @@ -151,6 +151,9 @@ class GatewayConfig:
         
         # Reset trigger commands
         reset_triggers: List[str] = field(default_factory=lambda: ["/new", "/reset"])
    +
    +    # User-defined quick commands (slash commands that bypass the agent loop)
    +    quick_commands: Dict[str, Any] = field(default_factory=dict)
         
         # Storage paths
         sessions_dir: Path = field(default_factory=lambda: get_hermes_home() / "sessions")
    @@ -218,6 +221,7 @@ def to_dict(self) -> Dict[str, Any]:
                     p.value: v.to_dict() for p, v in self.reset_by_platform.items()
                 },
                 "reset_triggers": self.reset_triggers,
    +            "quick_commands": self.quick_commands,
                 "sessions_dir": str(self.sessions_dir),
                 "always_log_local": self.always_log_local,
             }
    @@ -252,12 +256,17 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
             if "sessions_dir" in data:
                 sessions_dir = Path(data["sessions_dir"])
             
    +        quick_commands = data.get("quick_commands", {})
    +        if not isinstance(quick_commands, dict):
    +            quick_commands = {}
    +
             return cls(
                 platforms=platforms,
                 default_reset_policy=default_policy,
                 reset_by_type=reset_by_type,
                 reset_by_platform=reset_by_platform,
                 reset_triggers=data.get("reset_triggers", ["/new", "/reset"]),
    +            quick_commands=quick_commands,
                 sessions_dir=sessions_dir,
                 always_log_local=data.get("always_log_local", True),
             )
    @@ -299,6 +308,16 @@ def load_gateway_config() -> GatewayConfig:
                 if sr and isinstance(sr, dict):
                     config.default_reset_policy = SessionResetPolicy.from_dict(sr)
     
    +            # Bridge quick commands from config.yaml into gateway runtime config.
    +            # config.yaml is the user-facing config source, so when present it
    +            # should override gateway.json for this setting.
    +            qc = yaml_cfg.get("quick_commands")
    +            if qc is not None:
    +                if isinstance(qc, dict):
    +                    config.quick_commands = qc
    +                else:
    +                    logger.warning("Ignoring invalid quick_commands in config.yaml (expected mapping, got %s)", type(qc).__name__)
    +
                 # Bridge discord settings from config.yaml to env vars
                 # (env vars take precedence — only set if not already defined)
                 discord_cfg = yaml_cfg.get("discord", {})

  • gateway/run.py

    @@ -1013,7 +1013,12 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
             
             # User-defined quick commands (bypass agent loop, no LLM call)
             if command:
    -            quick_commands = self.config.get("quick_commands", {})
    +            if isinstance(self.config, dict):
    +                quick_commands = self.config.get("quick_commands", {}) or {}
    +            else:
    +                quick_commands = getattr(self.config, "quick_commands", {}) or {}
    +            if not isinstance(quick_commands, dict):
    +                quick_commands = {}
                 if command in quick_commands:
                     qcmd = quick_commands[command]
                     if qcmd.get("type") == "exec":

  • tests/gateway/test_config.py

    @@ -6,6 +6,7 @@
         Platform,
         PlatformConfig,
         SessionResetPolicy,
    +    load_gateway_config,
     )
     
     
    @@ -89,15 +90,49 @@ def test_full_roundtrip(self):
                 platforms={
                     Platform.TELEGRAM: PlatformConfig(
                         enabled=True,
    -                    token="tok",
    +                    token="[redacted:secret-assignment]",
                         home_channel=HomeChannel(Platform.TELEGRAM, "123", "Home"),
                     ),
                 },
                 reset_triggers=["/new"],
    +            quick_commands={"limits": {"type": "exec", "command": "echo ok"}},
             )
             d = config.to_dict()
             restored = GatewayConfig.from_dict(d)
     
             assert Platform.TELEGRAM in restored.platforms
    -        assert restored.platforms[Platform.TELEGRAM].token == "tok"
    +        assert restored.platforms[Platform.TELEGRAM].token == "tok_123"
             assert restored.reset_triggers == ["/new"]
    +        assert restored.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
    +
    +
    +class TestLoadGatewayConfig:
    +    def test_bridges_quick_commands_from_config_yaml(self, tmp_path, monkeypatch):
    +        hermes_home = tmp_path / ".hermes"
    +        hermes_home.mkdir()
    +        config_path = hermes_home / "config.yaml"
    +        config_path.write_text(
    +            "quick_commands:\n"
    +            "  limits:\n"
    +            "    type: exec\n"
    +            "    command: echo ok\n",
    +            encoding="utf-8",
    +        )
    +
    +        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
    +
    +        config = load_gateway_config()
    +
    +        assert config.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
    +
    +    def test_invalid_quick_commands_in_config_yaml_are_ignored(self, tmp_path, monkeypatch):
    +        hermes_home = tmp_path / ".hermes"
    +        hermes_home.mkdir()
    +        config_path = hermes_home / "config.yaml"
    +        config_path.write_text("quick_commands: not-a-mapping\n", encoding="utf-8")
    +
    +        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
    +
    +        config = load_gateway_config()
    +
    +        assert config.quick_commands == {}

  • tests/test_quick_commands.py

    @@ -146,3 +146,20 @@ async def test_timeout_returns_error(self):
                 result = await runner._handle_message(event)
             assert result is not None
             assert "timed out" in result.lower()
    +
    +    @pytest.mark.asyncio
    +    async def test_gateway_config_object_supports_quick_commands(self):
    +        from gateway.config import GatewayConfig
    +        from gateway.run import GatewayRunner
    +
    +        runner = GatewayRunner.__new__(GatewayRunner)
    +        runner.config = GatewayConfig(
    +            quick_commands={"limits": {"type": "exec", "command": "echo ok"}}
    +        )
    +        runner._running_agents = {}
    +        runner._pending_messages = {}
    +        runner._is_user_authorized = MagicMock(return_value=True)
    +
    +        event = self._make_event("limits")
    +        result = await runner._handle_message(event)
    +        assert result == "ok"