fix: validate gateway server config block at load time (#3050)

Move the typed-validation fix from the example into the wrapper where
the bug actually lives. load_gateway_config now validates the optional
'gateway:' server block field-by-field against the SDK GatewayConfig
dataclass (single source of truth), so misspelled keys, wrong types and
negative values are rejected with clear errors instead of silently
falling back to defaults.

Revert the example to use the typed SDK GatewayConfig instead of a local
shadowing class that bypassed validation.

Co-authored-by: Mervin Praison <[email protected]>

examples/python/gateway_example.py

@@ -7,16 +7,17 @@ multiple agents and managing real-time communication.
 
 from praisonaiagents import Agent, GatewayConfig, SessionConfig
 
-# Configure the gateway
+# Configure the gateway using the typed SDK GatewayConfig.
+# Invalid or misspelled fields are caught by the dataclass signature.
 gateway_config = GatewayConfig(
     host="127.0.0.1",
     port=8765,
     max_connections=100,
     heartbeat_interval=30,
     session_config=SessionConfig(
-        timeout=3600,  # 1 hour session timeout
+        timeout=3600,
         max_messages=500,
-    )
+    ),
 )
 
 # Create specialized agents
@@ -40,7 +41,7 @@ if __name__ == "__main__":
     print(f"  WebSocket URL: {gateway_config.ws_url}")
     print(f"  Max Connections: {gateway_config.max_connections}")
     print()
-    
+
     # Test agent
     response = researcher.start("What are the key benefits of multi-agent systems?")
     print("Researcher Response:")

src/praisonai-bot/praisonai_bot/gateway/server.py

@@ -54,6 +54,42 @@ from .unicode_utils import safe_error_message, safe_log_message, extract_root_ca
 from .supervisor import ChannelState, ChannelSupervisor
 
 
+def _closest_match(key: str, candidates) -> Optional[str]:
+    """Return the closest candidate name to ``key`` for typo suggestions."""
+    import difflib
+
+    matches = difflib.get_close_matches(key, list(candidates), n=1, cutoff=0.6)
+    return matches[0] if matches else None
+
+
+def _value_matches_field(value: Any, field_type: Any) -> bool:
+    """Best-effort check that ``value`` is compatible with a dataclass field type.
+
+    Handles the plain and string ("from __future__ import annotations") forms of
+    the primitive/container types used by ``GatewayConfig``. bools are rejected
+    where an int is expected. Unknown/complex types are treated as valid so we
+    never reject something we cannot confidently validate.
+    """
+    type_name = field_type if isinstance(field_type, str) else getattr(field_type, "__name__", str(field_type))
+    type_name = type_name.replace("typing.", "")
+    if type_name.startswith("Optional[") and type_name.endswith("]"):
+        type_name = type_name[len("Optional["):-1]
+
+    if type_name.startswith("int"):
+        return isinstance(value, int) and not isinstance(value, bool)
+    if type_name.startswith("float"):
+        return isinstance(value, (int, float)) and not isinstance(value, bool)
+    if type_name.startswith("bool"):
+        return isinstance(value, bool)
+    if type_name.startswith("str"):
+        return isinstance(value, str)
+    if type_name.startswith("List") or type_name.startswith("list"):
+        return isinstance(value, list)
+    if type_name.startswith("Dict") or type_name.startswith("dict"):
+        return isinstance(value, dict)
+    return True
+
+
 # Per-platform token env-var fallbacks used by the generic channel-launch path
 # (Issue #3578). Channels whose credentials live in the environment rather than
 # gateway.yaml (email/AgentMail/Linear) resolve their token from the first env
@@ -5554,6 +5590,46 @@ class WebSocketGateway:
                         "(use ${ENV_VAR} syntax for env vars)"
                     )
 
+        # Validate the optional 'gateway:' server block field-by-field,
+        # reusing the typed GatewayConfig dataclass as the single source of
+        # truth so misspelled or wrongly-typed server settings are caught at
+        # load time instead of being silently ignored at runtime (see #3050).
+        gw_block = migrated.get("gateway")
+        if gw_block is not None:
+            if not isinstance(gw_block, dict):
+                errors.append("'gateway' must be a dictionary")
+            else:
+                import dataclasses
+
+                allowed_fields = {f.name: f for f in dataclasses.fields(GatewayConfig)}
+                # session_config is a nested dataclass configured elsewhere.
+                allowed_fields.pop("session_config", None)
+                for key, value in gw_block.items():
+                    if key not in allowed_fields:
+                        suggestion = _closest_match(key, allowed_fields.keys())
+                        hint = f" (did you mean '{suggestion}'?)" if suggestion else ""
+                        errors.append(
+                            f"Unknown gateway setting '{key}'{hint}. "
+                            f"Allowed: {', '.join(sorted(allowed_fields))}"
+                        )
+                        continue
+                    if value is None:
+                        continue
+                    field_type = allowed_fields[key].type
+                    if not _value_matches_field(value, field_type):
+                        errors.append(
+                            f"Gateway setting '{key}' has invalid type "
+                            f"'{type(value).__name__}' (expected {field_type})"
+                        )
+                    elif (
+                        isinstance(value, (int, float))
+                        and not isinstance(value, bool)
+                        and value < 0
+                    ):
+                        errors.append(
+                            f"Gateway setting '{key}' must not be negative (got {value})"
+                        )
+
         if errors:
             raise ValueError(
                 f"Gateway config validation failed ({config_path}):\n"

src/praisonai/tests/integration/test_websocket_integration.py

@@ -134,6 +134,78 @@ class TestWebSocketIntegration:
         assert "channels" in cfg
         assert cfg["channels"]["telegram"]["token"] == "test123"
 
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_unknown_key(self, tmp_path):
+        """Misspelled gateway server settings are rejected with a suggestion."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  reconnect_timout: 60\n"
+        )
+
+        with pytest.raises(ValueError, match="reconnect_timeout"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_wrong_type(self, tmp_path):
+        """A string where an int is expected is rejected at load time."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  port: notaport\n"
+        )
+
+        with pytest.raises(ValueError, match="invalid type"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_negative(self, tmp_path):
+        """Negative numeric server settings are rejected."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  max_connections: -5\n"
+        )
+
+        with pytest.raises(ValueError, match="must not be negative"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_valid(self, tmp_path):
+        """A valid gateway server block loads successfully."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        good_config = tmp_path / "good.yaml"
+        good_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  host: 127.0.0.1\n  port: 8765\n  max_connections: 100\n"
+        )
+
+        cfg = WebSocketGateway.load_gateway_config(str(good_config))
+        assert cfg["gateway"]["port"] == 8765
+
     @pytest.mark.asyncio
     async def test_websocket_connect_and_message(self):
         """Test actual WebSocket connection to gateway server.