fix(gateway): make WriteEngine + converter raise-safe against malformed write commands

ingestor/converters.py

@@ -25,6 +25,8 @@ def to_typed_value(raw, data_type: str):
     data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'.
     Raises ConversionError on any invalid input (never a bare TypeError).
     """
+    if not isinstance(data_type, str):
+        raise ConversionError(f'data_type must be a string, got {type(data_type).__name__}')
     dtype = data_type.strip().lower()
     try:
         if dtype in ('bool', 'boolean'):

ingestor/engines/write.py

@@ -88,6 +88,10 @@ class WriteEngine(SientiaMonitoring):
     async def write(self, cmd: dict) -> WriteResult:
         await self._rate_limit()
 
+        missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd]
+        if missing:
+            return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error')
+
         try:
             typed = to_typed_value(cmd['value'], cmd['data_type'])
         except ConversionError as e:
@@ -156,7 +160,7 @@ class WriteEngine(SientiaMonitoring):
         return result
 
     async def _emit_success(self, cmd: dict, start: float):
-        tags = {'pod_id': self.pod_id, 'server_name': cmd['server_id']}
+        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
         await self.emit_metric(
             metric_object=metrics.OPC_WRITE_TOTAL,
             tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'},