fix: drain proxy capture safely and recognize raw PI30 activity

Author: Maksym Shtembuliak

CHANGELOG.md

@@ -100,6 +100,12 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Fixed
 
+- Proxy Capture recognizes CRC-checked raw PI30 commands when checking cloud
+  session activity. Stopping or cancelling capture drains relay tasks and pending
+  trace writes before closing the recording file. Startup diagnostics report
+  connection, identity and activity separately; the existing timeout and live
+  endpoint-restoration checks remain unchanged.
+
 - An initially silent collector that reveals a short framed heartbeat can
   confirm its full PN through FC2 on the same socket instead of being closed
   and retried with another dialect (#23). A short PN alone still cannot claim

custom_components/eybond_local/runtime/coordinator/cloud_tools.py

@@ -1871,15 +1871,35 @@ async def _async_validate_proxy_capture_target(self, *, target_host: str, target
             )
 
     async def _async_wait_for_proxy_capture_reconnect(self, trace_path: Path) -> None:
-        deadline = asyncio.get_running_loop().time() + 75.0
+        loop = asyncio.get_running_loop()
+        started = loop.time()
+        deadline = started + 75.0
         identity_verified = False
-        while asyncio.get_running_loop().time() < deadline:
+        last_progress = None
+        next_progress_at = started
+        status: dict[str, object] = {}
+        while loop.time() < deadline:
             if not self._proxy_capture_process_running():
                 raise RuntimeError("proxy_capture_route_stopped")
             status = await self.hass.async_add_executor_job(
                 lambda: inspect_proxy_capture_start_status(trace_path)
             )
             upstream_error = str(status.get("upstream_error") or "")
+            progress = (
+                bool(status.get("connected")),
+                bool(status.get("identity_verified")),
+                bool(status.get("operational_activity")),
+                bool(upstream_error),
+            )
+            now = loop.time()
+            if progress != last_progress or now >= next_progress_at:
+                logger.debug(
+                    "Proxy capture startup entry=%s elapsed=%.1fs connected=%s "
+                    "identity_verified=%s operational_activity=%s upstream_error=%s",
+                    self.config_entry.entry_id, now - started, *progress,
+                )
+                last_progress = progress
+                next_progress_at = now + 15.0
             if upstream_error:
                 raise RuntimeError(f"proxy_capture_upstream_connect_failed:{upstream_error}")
             if status.get("identity_mismatch"):
@@ -1892,6 +1912,14 @@ async def _async_wait_for_proxy_capture_reconnect(self, trace_path: Path) -> Non
             ):
                 return
             await asyncio.sleep(1.0)
+        logger.warning(
+            "Proxy capture startup timed out entry=%s elapsed=%.1fs connected=%s "
+            "identity_verified=%s operational_activity=%s upstream_error=%s",
+            self.config_entry.entry_id, loop.time() - started,
+            bool(status.get("connected")), identity_verified,
+            bool(status.get("operational_activity")),
+            bool(status.get("upstream_error")),
+        )
         if identity_verified:
             raise TimeoutError("proxy_capture_cloud_session_not_ready")
         raise TimeoutError("proxy_capture_collector_identity_timeout")

custom_components/eybond_local/support/cloud_session_wire.py

@@ -14,6 +14,7 @@
     decode_header,
 )
 from ..payload.modbus import crc16_modbus
+from ..payload.pi30 import build_request as build_pi30_request
 
 _MODBUS_FIXED_LEN_FCODES = frozenset({3, 4, 6})
 _MODBUS_WRITE_MULTIPLE_FCODE = 16
@@ -30,10 +31,20 @@
 )
 _MODBUS_INCOMPLETE = object()
 _ASCII_INCOMPLETE = object()
+# One bounded raw PI30 line, including its two escaped CRC bytes and CR.
+# Recognition is opt-in for transparent capture, not a new shadow carrier.
+_PI30_MAX_FRAME_LENGTH = 512
 
 
-def consume_cloud_message(buffer: bytearray) -> tuple[str, bytes] | None:
-    """Consume one complete AT, inverter, or EyeBond message from ``buffer``."""
+def consume_cloud_message(
+    buffer: bytearray, *, allow_pi30: bool = False,
+) -> tuple[str, bytes] | None:
+    """Consume one complete AT, inverter, or EyeBond message from ``buffer``.
+
+    Transparent observation may opt into CRC-verified raw PI30 (kind ``pi30``).
+    The default framing and shadow allowlists are unchanged. Legacy no-CRC
+    ASCII remains distinct: printable text is not proof of a valid PI30 CRC.
+    """
 
     if not buffer:
         return None
@@ -54,6 +65,11 @@ def consume_cloud_message(buffer: bytearray) -> tuple[str, bytes] | None:
     if modbus is not None:
         return "modbus", modbus
 
+    if allow_pi30:
+        pi30_frame = _consume_pi30_frame(buffer)
+        if pi30_frame is not None:
+            return "pi30", pi30_frame
+
     ascii_frame = _consume_g_ascii_frame(buffer)
     if ascii_frame is _ASCII_INCOMPLETE:
         frame = _consume_eybond_frame_if_complete(buffer)
@@ -120,6 +136,31 @@ def _modbus_crc_is_valid(frame: bytes) -> bool:
     return crc16_modbus(frame[:-2]) == int.from_bytes(frame[-2:], "little")
 
 
+def _consume_pi30_frame(buffer: bytearray) -> bytes | None:
+    """Consume a bounded printable body with an exact PI30 CRC and terminator.
+
+    CRC bytes are binary, not part of the printable body. Rebuilding through
+    the payload helper also validates PI30's special CRC-byte escaping, without
+    duplicating it. No scanning past a bad line, stripping, or buffer mutation
+    on incomplete/invalid data. Legacy no-CRC G-ASCII is handled separately.
+    """
+
+    first = buffer[0]
+    if first != 0x28 and not 0x41 <= first <= 0x5A:
+        return None
+    carriage = buffer.find(b"\r", 0, _PI30_MAX_FRAME_LENGTH)
+    if carriage < 3:
+        return None
+    frame = bytes(buffer[:carriage + 1])
+    body = frame[:-3]
+    if body.startswith(b"AT+") or any(byte < 0x20 or byte > 0x7E for byte in body):
+        return None
+    if build_pi30_request(body.decode("ascii")) != frame:
+        return None
+    del buffer[:carriage + 1]
+    return frame
+
+
 def _consume_g_ascii_frame(buffer: bytearray) -> bytes | object | None:
     if not buffer:
         return None