fix: harden collector runtime sessions

Author: Maksym Shtembuliak

CHANGELOG.md

@@ -27,6 +27,12 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Changed
 
+- Marked the documented Anenji `0x8401` Protocol 4 control surface as tested for
+  this exact fingerprint after device-owner validation included working OP2
+  controls that are not exposed by the vendor apps. The irreversible
+  generation-data clear and user-parameter reset actions remain blocked by the
+  independent destructive-action safety policy even though the profile is
+  treated as compatible.
 - Unknown inverter models that report documented Anenji Communication Protocol
   3, 4, 5, or 6 now select the matching protocol-specific telemetry map
   automatically instead of requiring a new exact-model catalog entry. Their
@@ -55,6 +61,15 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Fixed
 
+- Fixed a framed-transport desynchronization where one unexpected unwrapped
+  Modbus RTU response could be combined with the next EyeBond header, consume
+  subsequent valid frames, and stall telemetry until the socket was replaced.
+  Malformed, oversized, or incomplete frames now close only the affected
+  session with a typed diagnostic reason, after which a fresh callback can
+  resume normally.
+- Preload the SmartESS semantic catalog with the other JSON metadata caches in
+  Home Assistant's executor, avoiding a first-use blocking file read from the
+  event loop when diagnostics or cloud evidence first resolves a field.
 - Restored exact Anenji ANJ-11KW-48V-WIFI-P identification for the captured
   Protocol 4 fingerprint `layout=4`, `model=0x8401`. Unlike the broad legacy
   heuristic, the new catalog entry affects only this proven fingerprint; it

catalog/inverter_models/models/anenji_anj_11kw_48v_wifi_p.json

@@ -40,19 +40,21 @@
     "notes": [
       "Device-specific runtime profile is user-confirmed.",
       "Communication Protocol No. 3-10 documentation defines the protocol-4 telemetry projection used by this model.",
-      "The captured 0x8401 hardware fingerprint uses the documented Protocol 4 telemetry surface; its write controls remain untested and require explicit Full Control.",
-      "Register 602 secondary-output priority is documented and hardware-read-correlated; local write behavior remains conditional pending a device retest."
+      "The captured 0x8401 hardware fingerprint uses the documented Protocol 4 telemetry and control surface.",
+      "The 0x8401 owner confirmed local writes and resulting device behavior for OP2 enable plus start-hour and end-hour timer controls, including controls absent from the vendor apps.",
+      "The documented Protocol 4 control surface is treated as tested for this exact fingerprint; irreversible generation-data clear and user-parameter reset actions remain blocked by the independent safety policy."
     ]
   },
   "known_limitations": [
-    "Secondary Output Priority writes are available only in Full Control until register 602 is write-tested on hardware."
+    "Irreversible generation-data clear and user-parameter reset actions remain blocked even though the ordinary Protocol 4 control surface is enabled."
   ],
   "knowledge_summary": "Anenji 11kW unit with a model-specific control profile over the shared protocol-4 register projection, backed by donor captures, user-confirmed controls, and the Protocol No. 3-10 documentation.",
   "source_keys": [
     "corpus_anenji_anj_11kw",
     "private_message_anenji_anj_11kw_validation",
     "documentation_anenji_protocol_3_10",
     "support_archive_anenji_anj_11kw_secondary_controls_20260707",
-    "github_issue_13_anenji_8401_beta4"
+    "github_issue_13_anenji_8401_beta4",
+    "github_issue_38_anenji_8401_op2_controls"
   ]
 }

catalog/inverter_models/sources/github_issue_38_anenji_8401_op2_controls.json

@@ -0,0 +1,15 @@
+{
+  "schema_version": 1,
+  "source_key": "github_issue_38_anenji_8401_op2_controls",
+  "kind": "github_issue",
+  "title": "Anenji 0x8401 OP2 hardware control validation",
+  "reference": "project-issue:38#issuecomment-5482296845",
+  "captured_at": "2026-08-31",
+  "visibility": "public",
+  "summary": "The owner of the captured layout 4, model 0x8401 Protocol 4 inverter confirmed successful local writes and resulting behavior for OP2 output enable, OP2 start hour, and OP2 end hour. The overload-alarm setting was visible but was not explicitly reported as write-tested.",
+  "assertions": [
+    "hardware_test",
+    "write_validation",
+    "control_behavior"
+  ]
+}

custom_components/eybond_local/collector/transport/common.py

@@ -162,21 +162,7 @@ def _looks_like_eybond_frame_start(chunk: bytes) -> bool:
         header = decode_header(chunk)
     except Exception:
         return False
-    if header.total_len < HEADER_SIZE:
-        return False
-    if header.total_len > 4096:
-        return False
-    if header.fcode not in {
-        FC_HEARTBEAT,
-        FC_QUERY_COLLECTOR,
-        FC_SET_COLLECTOR,
-        FC_FORWARD_TO_DEVICE,
-        FC_SET_DEVICE_REG,
-        FC_TRIGGER_QUERY_REAL_TIME,
-        FC_TRIGGER_QUERY_HISTORY,
-    }:
-        return False
-    return True
+    return _runtime_eybond_header_error(header) == ""
 
 
 def _classify_initial_protocol_shape(chunk: bytes) -> str:
@@ -242,6 +228,26 @@ def _short_ascii(value: bytes, *, limit: int = 160) -> str:
 }
 
 
+def _runtime_eybond_header_error(header: EybondHeader) -> str:
+    """Return a typed reason when a runtime EyeBond header is implausible.
+
+    ``decode_header`` deliberately stays a mechanical wire decoder because the
+    transparent cloud tooling understands more function codes than the local
+    runtime.  The runtime reader, protocol sniffing, and the mixed AT reader do
+    need one shared fail-closed definition, however: accepting an arbitrary
+    eight-byte sequence as a header can turn one raw RTU response into a large
+    pseudo-frame and consume every valid frame that follows it.
+    """
+
+    if header.payload_len < 0:
+        return "collector_frame_length_invalid"
+    if header.payload_len > _AT_TEXT_MAX_MIXED_FRAME_PAYLOAD_LEN:
+        return "collector_frame_payload_too_large"
+    if header.fcode not in _AT_TEXT_MIXED_FRAME_FCODES:
+        return "collector_frame_function_invalid"
+    return ""
+
+
 def _mask_identity_token(value: str) -> str:
     token = str(value or "").strip()
     if len(token) <= 6:

custom_components/eybond_local/collector/transport/connections.py

@@ -27,8 +27,6 @@
     parse_heartbeat_pn,
 )
 from .common import (
-    _AT_TEXT_MAX_MIXED_FRAME_PAYLOAD_LEN,
-    _AT_TEXT_MIXED_FRAME_FCODES,
     _AT_TEXT_MIXED_FRAME_READ_TIMEOUT,
     _PrefixedAsyncReader,
     _cancel_and_join_task,
@@ -38,12 +36,21 @@
     _looks_like_plain_raw_response_start,
     _looks_like_uart_passthrough_value,
     _parse_fc2_collector_pn,
+    _runtime_eybond_header_error,
     _short_ascii,
 )
 
 logger = logging.getLogger(__name__)
 
 
+# An idle connection may remain silent indefinitely. Once the collector has
+# started a frame, however, every remaining structural part is bounded. This
+# prevents a truncated header or a false payload length from pinning the one
+# reader task and swallowing later valid responses on the same TCP socket.
+_FRAMED_HEADER_COMPLETION_TIMEOUT = 2.0
+_FRAMED_PAYLOAD_COMPLETION_TIMEOUT = 5.0
+
+
 def _modbus_rtu_response_length(prefix: bytes) -> int | None:
     """Return the complete RTU response length for a three-byte prefix.
 
@@ -422,17 +429,82 @@ async def _read_loop(
             reader = _PrefixedAsyncReader(reader)
         try:
             while True:
-                prefix = await reader.readexactly(3)
+                first = await reader.readexactly(1)
+                try:
+                    prefix = first + await asyncio.wait_for(
+                        reader.readexactly(2),
+                        timeout=_FRAMED_HEADER_COMPLETION_TIMEOUT,
+                    )
+                except asyncio.TimeoutError:
+                    self._collector.last_disconnect_reason = (
+                        "collector_frame_header_timeout"
+                    )
+                    logger.warning(
+                        "Closing collector session after partial frame prefix "
+                        "remote=%s prefix=%s",
+                        self._collector.remote_ip,
+                        first.hex(),
+                    )
+                    return
                 if prefix == b"AT+":
                     line = prefix + await reader.read_at_response()
                     self._handle_at_response(line)
                     continue
 
-                header_bytes = prefix + await reader.readexactly(HEADER_SIZE - len(prefix))
+                try:
+                    header_bytes = prefix + await asyncio.wait_for(
+                        reader.readexactly(HEADER_SIZE - len(prefix)),
+                        timeout=_FRAMED_HEADER_COMPLETION_TIMEOUT,
+                    )
+                except asyncio.TimeoutError:
+                    self._collector.last_disconnect_reason = (
+                        "collector_frame_header_timeout"
+                    )
+                    logger.warning(
+                        "Closing collector session after incomplete frame header "
+                        "remote=%s prefix=%s",
+                        self._collector.remote_ip,
+                        prefix.hex(),
+                    )
+                    return
                 header = decode_header(header_bytes)
+                header_error = _runtime_eybond_header_error(header)
+                if header_error:
+                    self._collector.last_disconnect_reason = header_error
+                    logger.warning(
+                        "Closing collector session after malformed frame header "
+                        "remote=%s reason=%s header=%s tid=%d devcode=0x%04X "
+                        "devaddr=0x%02X fc=%d payload=%d",
+                        self._collector.remote_ip,
+                        header_error,
+                        header_bytes.hex(),
+                        header.tid,
+                        header.devcode,
+                        header.devaddr,
+                        header.fcode,
+                        header.payload_len,
+                    )
+                    return
                 payload = b""
                 if header.payload_len > 0:
-                    payload = await reader.readexactly(header.payload_len)
+                    try:
+                        payload = await asyncio.wait_for(
+                            reader.readexactly(header.payload_len),
+                            timeout=_FRAMED_PAYLOAD_COMPLETION_TIMEOUT,
+                        )
+                    except asyncio.TimeoutError:
+                        self._collector.last_disconnect_reason = (
+                            "collector_frame_payload_timeout"
+                        )
+                        logger.warning(
+                            "Closing collector session after incomplete frame payload "
+                            "remote=%s tid=%d fc=%d expected=%d",
+                            self._collector.remote_ip,
+                            header.tid,
+                            header.fcode,
+                            header.payload_len,
+                        )
+                        return
 
                 self._collector.last_devcode = header.devcode
                 logger.debug(
@@ -1218,13 +1290,7 @@ async def _read_mixed_frame_tail(
             return None
 
     def _looks_like_mixed_frame_header(self, header: EybondHeader) -> bool:
-        if header.payload_len < 0:
-            return False
-        if header.payload_len > _AT_TEXT_MAX_MIXED_FRAME_PAYLOAD_LEN:
-            return False
-        if header.fcode not in _AT_TEXT_MIXED_FRAME_FCODES:
-            return False
-        return True
+        return _runtime_eybond_header_error(header) == ""
 
     def _record_unhandled_raw_fragment(self, payload: bytes, *, parser: str) -> None:
         self._collector.raw_unhandled_line_count += 1

custom_components/eybond_local/collector/transport/listener.py

@@ -81,6 +81,7 @@ class _CollectorSessionInventoryEntry:
     collector_identity_source: str = ""
     collector_identity_sources: set[str] = field(default_factory=set)
     observed_protocol_shapes: set[str] = field(default_factory=set)
+    close_reason: str = ""
 
     def diagnostics(self) -> dict[str, object]:
         result: dict[str, object] = {
@@ -106,6 +107,8 @@ def diagnostics(self) -> dict[str, object]:
             result["observed_protocol_shapes"] = sorted(
                 self.observed_protocol_shapes
             )
+        if self.close_reason:
+            result["close_reason"] = self.close_reason
         return result
 
 
@@ -922,6 +925,15 @@ def _mark_socket_session_closed(self, session_id: str, connection: object) -> No
         ):
             if mapping.get(normalized) is connection:
                 mapping.pop(normalized, None)
+        entry = self._session_inventory.get(normalized)
+        if entry is not None:
+            try:
+                collector = connection.collector_info  # type: ignore[attr-defined]
+            except Exception:
+                collector = None
+            entry.close_reason = str(
+                getattr(collector, "last_disconnect_reason", "") or ""
+            ).strip()
         self._mark_session_state(normalized, "closed_disconnected")
         logger.info(
             "Collector socket session closed listener_port=%s session=%s",
@@ -1091,6 +1103,8 @@ async def pop_pending_socket_for_route(
                 pending = self._select_pending_socket_by_session_id(normalized_session_id)
                 if pending is None:
                     return None
+                if self._reserved_for_transparent_route(pending):
+                    return None
                 await self._pause_pending_sniff(pending)
                 if not self._pending_socket_still_registered(pending):
                     return None
@@ -1101,13 +1115,17 @@ async def pop_pending_socket_for_route(
                 pending = self._select_pending_socket(collector_ip)
                 if pending is None:
                     return None
+                if self._reserved_for_transparent_route(pending):
+                    return None
                 await self._pause_pending_sniff(pending)
                 if not self._pending_socket_still_registered(pending):
                     return None
                 return self._claim_pending_socket(pending)
 
             matched = self._select_pending_socket_by_collector_pn(normalized_pn)
             if matched is not None:
+                if self._reserved_for_transparent_route(matched):
+                    return None
                 return self._claim_pending_socket(matched)
 
             candidates = self._route_identity_candidates(collector_ip)
@@ -1117,6 +1135,14 @@ async def pop_pending_socket_for_route(
             for pending in candidates:
                 if not self._pending_socket_still_registered(pending):
                     continue
+                # A causally-new socket reserved by a byte-transparent cloud
+                # route must never be identity-probed or claimed by the normal
+                # runtime.  The transparent route owns the first read so the
+                # real collector/cloud handshake remains unmodified.  Baseline
+                # sessions and fresh sockets whose observed wire cannot belong
+                # to that route are intentionally unaffected.
+                if self._reserved_for_transparent_route(pending):
+                    continue
                 # Ownership safety: when a PN is present, never probe or claim a
                 # socket already identified as a DIFFERENT collector (e.g. a
                 # second collector behind the same NAT/public IP). Only our-PN or

custom_components/eybond_local/drivers/registry.py

@@ -21,6 +21,7 @@
 from ..metadata.profile_loader import load_driver_profile
 from ..metadata.register_schema_loader import load_register_schema
 from ..metadata.smartess_protocol_catalog_loader import load_smartess_protocol_catalog
+from ..metadata.smartess_semantic_catalog_loader import load_smartess_semantic_catalog
 from ..models import (
     BinarySensorDescription,
     CapabilityGroup,
@@ -393,6 +394,7 @@ def _prime_catalog_driven_metadata() -> None:
     load_device_catalog()
     compiled_catalog = load_compiled_detection_catalog()
     load_collector_cloud_profile_catalog()
+    load_smartess_semantic_catalog()
 
     for surface in compiled_catalog.surfaces.values():
         if surface.profile_name:

custom_components/eybond_local/metadata/profile_loader.py

@@ -780,10 +780,6 @@ def _validate_profile(profile: DriverProfileMetadata) -> None:
                 f"profile:{profile.key}:unsupported_support_tier:"
                 f"{capability.key}:{capability.support_tier}"
             )
-        if capability.support_tier == "blocked" and capability.tested:
-            raise ValueError(
-                f"profile:{profile.key}:blocked_capability_cannot_be_tested:{capability.key}"
-            )
         _unique_or_raise(
             items=(str(choice.value) for choice in capability.enum_choices),
             kind=f"choice_value:{capability.key}",

custom_components/eybond_local/protocol_catalogs/inverter_catalog.json

@@ -1367,6 +1367,17 @@
    "read_only": false,
    "support_capture_policy": "smg_anenji_11k"
   },
+  {
+   "key": "smg_anenji_11kw_8401_full",
+   "protocol_key": "modbus_smg",
+   "driver_key": "modbus_smg",
+   "variant_key": "anenji_anj_11kw_48v_wifi_p_8401",
+   "profile_name": "modbus_smg/models/anenji_anj_11kw_48v_wifi_p_8401.json",
+   "register_schema_name": "modbus_smg/protocols/communication_protocol_4.json",
+   "tier": "full",
+   "read_only": false,
+   "support_capture_policy": "smg_anenji_11k"
+  },
   {
    "key": "smg_protocol_5_family_fallback",
    "protocol_key": "modbus_smg",
@@ -1935,7 +1946,7 @@
     "serial_ascii_plausible"
    ],
    "model_name": "Anenji ANJ-11KW-48V-WIFI-P",
-   "surface_key": "smg_protocol_4_family_fallback",
+   "surface_key": "smg_anenji_11kw_8401_full",
    "runtime_probe": {
     "validation": [
      {