@@ -240,6 +240,7 @@ class GatewayMQTT(ComponentBase):
     _inverter_slot_serials = ()
     _slots_without_capacity = frozenset()
     _last_ratings = {}  # replaced, never changed in place
+    _assumed_ratings = frozenset()
     gateway_device_id = None
 
     def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, gateway_shared_ct=True, gateway_integrate_power=False, **kwargs):
@@ -347,6 +348,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt
         self._pv_only_serials = ()  # listed inverters with no battery: counted for PV, never written to
         self._slots_without_capacity = frozenset()  # slots bound with no battery size sensor, re-bound once the hub reports a size
         self._last_ratings = {}  # (entity suffix, rating name) -> the last real value the hub reported
+        self._assumed_ratings = frozenset()  # entity suffixes of inverters whose maximum rate has never been reported
         self._discovery = {}  # upper-case serial -> what the hub last reported it to be, for gateway_inverter_serial
         self._discovery_dirty = False
         self._discovery_save_retry_at = 0
@@ -1253,7 +1255,13 @@ def _inject_inverter_entities(self, inv, suffix, site_energy=None):
         # ratings unpopulated (seen for a few minutes after a hub restart) otherwise turned a 3kW
         # inverter into a 6kW one with no export limit, and those rates were written to it (#5478).
         record = self._discovery.get(inv.serial.upper(), {})
-        battery_rate_max = self._last_rating(suffix, "battery_rate_max", bat.rate_max_w) or record.get("rate_max_w") or 6000
+        battery_rate_max = self._last_rating(suffix, "battery_rate_max", bat.rate_max_w) or record.get("rate_max_w")
+        # With no rating ever reported the plan still needs a figure, but it is only an assumption:
+        # rate writes for the inverter are held until the real one arrives (see _rate_write_held)
+        assumed = not battery_rate_max and inv.type not in GATEWAY_COORDINATOR_TYPES
+        if assumed != (suffix in self._assumed_ratings):
+            self._assumed_ratings = self._assumed_ratings | {suffix} if assumed else self._assumed_ratings - {suffix}
+        battery_rate_max = battery_rate_max or 6000
         self.dashboard_item(f"sensor.{pfx}_battery_rate_max", battery_rate_max, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_rate_max", {}), app="gateway")
 
         self.dashboard_item(f"sensor.{pfx}_pv_power", inv.pv.power_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("pv_power", {}), app="gateway")
@@ -2412,6 +2420,9 @@ async def _check_rate_caps(self):
         for serial, caps in self._inverter_rate_caps().items():
             if self._rate_caps_sent.get(serial) == caps:
                 continue
+            if _serial_suffix(serial) in self._assumed_ratings:
+                # The cap comes from a rating the inverter has not reported yet (see _rate_write_held)
+                continue
             charge_cap_w, discharge_cap_w = caps
             await self.publish_command("set_rate_cap", serial=serial, charge_cap_w=charge_cap_w, discharge_cap_w=discharge_cap_w)
             self._rate_caps_sent[serial] = caps
@@ -2643,6 +2654,8 @@ async def number_event(self, entity_id, value):
         if serial is None:
             self.log(f"Warn: GatewayMQTT: number_event: cannot resolve serial for entity '{entity_id}' — command not sent")
             return
+        if ("_discharge_rate" in entity_id or "_charge_rate" in entity_id) and self._rate_write_held(serial, val):
+            return
         if "_discharge_rate" in entity_id:
             await self._send_control(entity_id, "set_discharge_rate", power_w=val, serial=serial)
         elif "_charge_rate" in entity_id:
@@ -2652,6 +2665,27 @@ async def number_event(self, entity_id, value):
         elif "_target_soc" in entity_id:
             await self._send_control(entity_id, "set_target_soc", target_soc=val, serial=serial)
 
+    def _rate_write_held(self, serial, power_w):
+        """Whether a charge or discharge rate write must wait for the inverter's rating.
+
+        Until an inverter has reported its maximum rate, the rate Predbat works to is the 6000W
+        default, and writing that to a 3kW inverter is the fault in #5478. Such a write is not
+        sent. Predbat reads the unchanged rate back as a write that did not complete and tries
+        again, so it goes through once the rating is known. A rate of zero is safe on any
+        inverter and is always sent, so a charge or discharge can still be stopped.
+
+        Args:
+            serial: The inverter the write is for.
+            power_w: The rate being written.
+
+        Returns:
+            bool: True if the write is held.
+        """
+        if power_w <= 0 or _serial_suffix(serial) not in self._assumed_ratings:
+            return False
+        self.log(f"Warn: GatewayMQTT: inverter {serial} has not reported its maximum rate — rate write of {power_w}W held until it does")
+        return True
+
     async def switch_event(self, entity_id, service):
         """Handle switch entity service calls (charge/discharge enable).
 
 
@@ -4042,6 +4042,80 @@ def test_ratings_never_reported_keep_the_existing_defaults(self):
         assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 6000
         assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 99999
 
+    def _writable(self, gw):
+        """Let *gw* take control writes, recording the commands it publishes."""
+        from unittest.mock import MagicMock
+
+        gw._mqtt_connected = True
+        gw._mqtt_client = MagicMock()
+        gw._gateway_online = True
+        gw._rate_caps_sent = {}
+        gw.is_alive = lambda: True
+        gw.get_arg = lambda name, default=None, **kwargs: default
+        gw._published = []
+
+        async def fake_publish_command(command, **kwargs):
+            gw._published.append((command, kwargs))
+
+        gw.publish_command = fake_publish_command
+        return gw
+
+    def test_rate_writes_are_held_until_the_inverter_reports_its_rating(self):
+        """With no rating ever reported the 6000W default is only an assumption, so no rate above zero is written on it."""
+        gw = self._writable(self._listed_gateway([]))
+        gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0).SerializeToString())
+        assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 6000
+
+        asyncio.run(gw.number_event("number.predbat_gateway_25g400_discharge_rate", 6000))
+        asyncio.run(gw.number_event("number.predbat_gateway_25g400_charge_rate", 6000))
+        assert gw._published == []
+        assert self._logged(gw, "Warn", "CE2225G400", "has not reported its maximum rate")
+
+        # Stopping a charge or discharge is safe at any rating, as are the other controls and the other inverter
+        asyncio.run(gw.number_event("number.predbat_gateway_25g400_charge_rate", 0))
+        asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 4))
+        asyncio.run(gw.number_event("number.predbat_gateway_23g800_discharge_rate", 3000))
+        assert gw._published == [
+            ("set_charge_rate", {"power_w": 0, "serial": "CE2225G400"}),
+            ("set_reserve", {"target_soc": 4, "serial": "CE2225G400"}),
+            ("set_discharge_rate", {"power_w": 3000, "serial": "CE2223G800"}),
+        ]
+
+        # Once the rating arrives the write goes through, and a later gap in the rating does not hold it again
+        for rate_max_w in (3000, 0):
+            gw._published.clear()
+            gw._process_telemetry(self._rated_status(rate_max_w=rate_max_w).SerializeToString())
+            asyncio.run(gw.number_event("number.predbat_gateway_25g400_discharge_rate", 3000))
+            assert gw._published == [("set_discharge_rate", {"power_w": 3000, "serial": "CE2225G400"})], rate_max_w
+
+    def test_rate_cap_is_not_sent_on_an_assumed_rating(self):
+        """The cap sent to the hub comes from the same rating, so it waits for the real one too."""
+        from types import SimpleNamespace
+
+        gw = self._writable(self._listed_gateway([]))
+        gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0).SerializeToString())
+        gw.base.inverters = [SimpleNamespace(inverter_type="GWMQTT", id=slot, battery_rate_max_charge=0.05, battery_rate_max_discharge=0.05) for slot in (0, 1)]
+        gw._inverter_rate_caps = lambda: {"CE2223G800": (3000, 3000), "CE2225G400": (6000, 6000)}
+
+        asyncio.run(gw._check_rate_caps())
+        assert gw._published == [("set_rate_cap", {"serial": "CE2223G800", "charge_cap_w": 3000, "discharge_cap_w": 3000})]
+
+        gw._process_telemetry(self._rated_status().SerializeToString())
+        gw._inverter_rate_caps = lambda: {"CE2223G800": (3000, 3000), "CE2225G400": (3000, 3000)}
+        asyncio.run(gw._check_rate_caps())
+        assert gw._published[1:] == [("set_rate_cap", {"serial": "CE2225G400", "charge_cap_w": 3000, "discharge_cap_w": 3000})]
+
+    def test_coordinator_rate_writes_are_not_held(self):
+        """A Gateway or Plant EMS is not judged on a battery rating of its own."""
+        gw = self._writable(self._listed_gateway([]))
+        status = self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"])
+        status.inverters[0].battery.rate_max_w = 0
+        gw._process_telemetry(status.SerializeToString())
+        assert gw._inverter_slot_serials == ["GW2347G077"]
+
+        asyncio.run(gw.number_event("number.predbat_gateway_47g077_charge_rate", 6000))
+        assert gw._published == [("set_charge_rate", {"power_w": 6000, "serial": "GW2347G077"})]
+
     def test_stored_ratings_are_used_after_a_restart(self):
         """With a serial list the ratings are in the discovery record, so a restart does not fall back to 6000W."""
         first = self._listed_gateway(["CE2223G800", "CE2225G400"])