fix: use dedicated AT collector reboot command

Author: Maksym Shtembuliak

CHANGELOG.md

@@ -55,6 +55,10 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Fixed
 
+- Fixed AT-text collector restart verification by separating the vendor's
+  dedicated soft-reset command from its staged-settings apply command. A command
+  acknowledgement still does not count as recovery until the old session drops
+  and the same collector reconnects on a new physical session.
 - Fixed canonical power-flow estimates that could count the same PV watts both
   as home consumption and battery charging when asynchronously sampled
   Protocol 3/4 registers disagreed. Instantaneous routes and their accumulated

custom_components/eybond_local/collector/management.py

@@ -7,9 +7,9 @@
 ONLY thing that selects an implementation here.
 
 Each adapter encapsulates its wire (framed FC=2/FC=3, or AT-text CLDSRVHOST1 /
-INTPARA) entirely; the parameter numbers and AT command strings live only inside
-the implementations. Every operation returns the SAME normalized model, and an
-operation a wire cannot confirm fails with a typed error under
+INTPARA / RESET) entirely; the parameter numbers and AT command strings live only
+inside the implementations. Every operation returns the SAME normalized model,
+and an operation a wire cannot confirm fails with a typed error under
 ``CollectorManagementError`` -- it never simulates success. Adapters resolve the
 live transport lazily through a provider callable, so a reconnect/handover never
 leaves them holding a stale socket.
@@ -26,7 +26,7 @@
 
 import asyncio
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass, field
 
 from ..connection.session_handle import (
@@ -51,6 +51,8 @@
 _AT_ENDPOINT_COMMAND = "CLDSRVHOST1"
 _AT_APPLY_COMMAND = "INTPARA"
 _AT_APPLY_VALUE = "29,1"
+_AT_REBOOT_COMMAND = "RESET"
+_AT_REBOOT_VALUE = "S"
 # The project's confirmed AT success status (see smartess_ble.extract_at_status_code).
 _AT_SUCCESS_STATUS = "W000"
 
@@ -566,12 +568,13 @@ async def async_set_uart_baudrate(self, baudrate: str) -> str:
 
 
 class AtTextCollectorManagementAdapter(CollectorManagementAdapter):
-    """Collector management over the AT-text wire (CLDSRVHOST1 / INTPARA).
+    """Collector management over the AT-text wire.
 
-    The AT command strings live only inside this class. Endpoint read/write and
-    apply/reboot are real and confirmed by the command response. The vendor's
-    collector-restart flow uses ``AT+INTPARA=29,1``; a recovery transaction
-    separately proves the resulting disconnect and same-identity reconnect.
+    The AT command strings live only inside this class. ``AT+INTPARA=29,1``
+    applies staged settings; ``AT+RESET=S`` requests a soft reboot. A ``W000``
+    response confirms only that the collector accepted the selected command.
+    Recovery transactions separately prove the physical disconnect and a new
+    same-identity session before treating a reboot as completed.
     """
 
     adapter_id = ADAPTER_COLLECTOR_AT_COMMANDS
@@ -672,6 +675,22 @@ async def _apply(self, transport: object) -> str:
             )
         return status
 
+    async def _reboot(self, transport: object) -> str:
+        """Send RESET S and require a confirmed ``W000`` acceptance status."""
+
+        status = await self._send_write(
+            transport, _AT_REBOOT_COMMAND, _AT_REBOOT_VALUE
+        )
+        if not status:
+            raise CollectorManagementConfirmationError(
+                f"collector_at_reboot_unconfirmed:command={_AT_REBOOT_COMMAND}"
+            )
+        if status != _AT_SUCCESS_STATUS:
+            raise CollectorManagementCommandError(
+                f"collector_at_reboot_rejected:command={_AT_REBOOT_COMMAND}:status={status}"
+            )
+        return status
+
     async def async_read_endpoint_state(self) -> CollectorEndpointState:
         transport = self._resolve_transport()
         current = await self._query(transport, _AT_ENDPOINT_COMMAND)
@@ -716,28 +735,38 @@ async def async_write_endpoint(
             extra=extra,
         )
 
-    async def _system_action(self, action: str) -> CollectorSystemActionResult:
+    async def _system_action(
+        self,
+        *,
+        action: str,
+        operation: Callable[[object], Awaitable[str]],
+        warning: str,
+    ) -> CollectorSystemActionResult:
         transport = self._resolve_transport(needs_write=True)
         current = await self._query(transport, _AT_ENDPOINT_COMMAND)
-        await self._apply(transport)
+        await operation(transport)
         return CollectorSystemActionResult(
             action=action,
             current_endpoint=current,
             reboot_required_before="",
             performed=True,
             adapter_id=self.adapter_id,
-            warnings=(
-                self._REBOOT_ACTION_WARNING
-                if action == "reboot"
-                else self._APPLY_ACTION_WARNING,
-            ),
+            warnings=(warning,),
         )
 
     async def async_apply_changes(self) -> CollectorSystemActionResult:
-        return await self._system_action("apply")
+        return await self._system_action(
+            action="apply",
+            operation=self._apply,
+            warning=self._APPLY_ACTION_WARNING,
+        )
 
     async def async_reboot(self) -> CollectorSystemActionResult:
-        return await self._system_action("reboot")
+        return await self._system_action(
+            action="reboot",
+            operation=self._reboot,
+            warning=self._REBOOT_ACTION_WARNING,
+        )
 
     async def async_query_parameters(
         self,

docs/architecture/CONNECTION_ARCHITECTURE.md

@@ -165,6 +165,12 @@ session-pinned and its teardown preserves that socket; it cannot switch to a
 same-PN sibling and cannot create reconnect evidence by closing the socket
 itself.
 
+The negotiated wire adapter owns the distinction between applying staged
+collector settings and requesting a reboot. On AT-text sessions these are
+separate vendor operations; an acknowledgement proves only command acceptance.
+The recovery authority still requires observable reset activity and a new
+same-PN session before it confirms a completed restart.
+
 Reset activity is observed against the complete pre-action session cohort, not
 against one selected socket's EOF. Some collectors briefly keep several
 same-PN sockets alive: a baseline sibling may close, or a new socket may open,

tests/test_collector_management.py

@@ -106,6 +106,8 @@ async def async_write(self, command: str, value: str) -> CollectorAtResponse:
             if self._reject_intpara:
                 return CollectorAtResponse(command="ERR", value="", raw="")
             return CollectorAtResponse(command="INTPARA", value="W000", raw="")
+        if command == "RESET":
+            return CollectorAtResponse(command="RESET", value="W000", raw="")
         return CollectorAtResponse(command=command, value="", raw="")
 
 
@@ -268,13 +270,13 @@ async def test_apply_rejected_raises_typed_and_never_applied(self) -> None:
         with self.assertRaises(CollectorManagementConfirmationError):
             await self._adapter(transport).async_apply_changes()
 
-    async def test_reboot_uses_vendor_intpara_restart_command(self) -> None:
+    async def test_reboot_uses_dedicated_vendor_soft_reset_command(self) -> None:
         transport = _FakeAtTransport()
 
         result = await self._adapter(transport).async_reboot()
 
         self.assertEqual(transport.queries, ["CLDSRVHOST1"])
-        self.assertEqual(transport.writes, [("INTPARA", "29,1")])
+        self.assertEqual(transport.writes, [("RESET", "S")])
         self.assertEqual(result.action, "reboot")
         self.assertTrue(result.performed)
         self.assertEqual(result.current_endpoint, "iot.eybond.com,18899,TCP")
@@ -440,6 +442,26 @@ async def async_write(self, command, value):
 
         return AtTextCollectorManagementAdapter(lambda: _At())
 
+    def _reboot_adapter(self, reset_command: str, reset_value: str):
+        class _At:
+            async def async_query(self, command):
+                return CollectorAtResponse(
+                    command="CLDSRVHOST1",
+                    value="1.2.3.4,18899,TCP",
+                    raw="",
+                )
+
+            async def async_write(self, command, value):
+                if command == "RESET":
+                    return CollectorAtResponse(
+                        command=reset_command,
+                        value=reset_value,
+                        raw="",
+                    )
+                return CollectorAtResponse(command=command, value=value, raw="")
+
+        return AtTextCollectorManagementAdapter(lambda: _At())
+
     async def test_intpara_w000_apply_succeeds(self) -> None:
         result = await self._apply_adapter("INTPARA", "W000").async_apply_changes()
         self.assertTrue(result.performed)
@@ -456,6 +478,22 @@ async def test_intpara_foreign_command_is_confirmation_error(self) -> None:
         with self.assertRaises(CollectorManagementConfirmationError):
             await self._apply_adapter("SOMETHINGELSE", "W000").async_apply_changes()
 
+    async def test_reset_w000_reboot_succeeds(self) -> None:
+        result = await self._reboot_adapter("RESET", "W000").async_reboot()
+        self.assertTrue(result.performed)
+
+    async def test_reset_w001_reboot_rejected_is_command_error(self) -> None:
+        with self.assertRaises(CollectorManagementCommandError):
+            await self._reboot_adapter("RESET", "W001").async_reboot()
+
+    async def test_reset_empty_status_is_confirmation_error(self) -> None:
+        with self.assertRaises(CollectorManagementConfirmationError):
+            await self._reboot_adapter("RESET", "").async_reboot()
+
+    async def test_reset_foreign_command_is_confirmation_error(self) -> None:
+        with self.assertRaises(CollectorManagementConfirmationError):
+            await self._reboot_adapter("SOMETHINGELSE", "W000").async_reboot()
+
     async def test_cldsrvhost1_w001_write_and_empty_readback_rejected(self) -> None:
         class _At:
             async def async_query(self, command):

tests/test_hub.py

@@ -4621,7 +4621,7 @@ async def _run() -> None:
 
         asyncio.run(_run())
 
-    def test_at_reboot_uses_intpara_and_returns_confirmed_result(self) -> None:
+    def test_at_reboot_uses_soft_reset_and_returns_confirmed_result(self) -> None:
         async def _run() -> None:
             hub, link = self._at_hub()
 
@@ -4630,9 +4630,7 @@ async def _run() -> None:
             self.assertEqual(result["status"], "reboot_triggered")
             self.assertEqual(result["action"], "reboot")
             self.assertIn("CLDSRVHOST1", link.collector_at_transport.queries)
-            self.assertIn(
-                ("INTPARA", "29,1"), link.collector_at_transport.writes
-            )
+            self.assertIn(("RESET", "S"), link.collector_at_transport.writes)
 
         asyncio.run(_run())
 

tests/test_strategy_verification.py

@@ -1205,7 +1205,7 @@ async def _exact_transport():
         )
         self.assertIn((FC_SET_COLLECTOR, bytes((29,)) + b"1"), wire.calls)
 
-    def test_at_adapter_reboot_uses_exact_session_vendor_command(self) -> None:
+    def test_at_adapter_reboot_uses_exact_session_soft_reset_command(self) -> None:
         from custom_components.eybond_local.collector.at import CollectorAtResponse
 
         class _AtWire:
@@ -1241,7 +1241,7 @@ async def _exact_transport():
         receipt = asyncio.run(channel.async_send_restart())
 
         self.assertEqual(wire.queries, ["CLDSRVHOST1"])
-        self.assertEqual(wire.writes, [("INTPARA", "29,1")])
+        self.assertEqual(wire.writes, [("RESET", "S")])
         self.assertEqual(receipt["status"], "reboot_requested")
         self.assertEqual(receipt["action"], "reboot")
         self.assertTrue(receipt["performed"])