Recover kit checksums from a truncated Sox schema reply

The device builds the schema and version replies in a fixed 512-byte
packet and drops whatever does not fit. Read strings defensively, recover
a cut-off checksum from the one local manifest matching its prefix, and
fail with a clear message when the kit list cannot be recovered.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app_state.py

@@ -8,7 +8,7 @@ so later rebinding would only ever be seen by the module that did it.
 import configparser
 import os
 
-VERSION = "0.0.0.133"
+VERSION = "0.0.0.134"
 PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties")
 DEFAULT_GRID_SIZE = 10
 DEFAULT_ADD_TIMESTAMP = True

device_session.py

@@ -45,6 +45,7 @@ class DeviceSession:
         """DASP connect → read schema → load root. Called from background thread."""
         self.sox_client = SoxClient(host, int(port), user, password)
         self.sox_client.on_event = self._on_sox_event
+        self.sox_client.checksum_resolver = schema.resolve_truncated_checksum
         self.sox_client.connect()
 
         parts = self.sox_client.read_schema()

schema.py

@@ -224,9 +224,32 @@ def find_manifest_path(kit_name, checksum):
     return None
 
 
+def resolve_truncated_checksum(kit_name, prefix):
+    """The checksum of the one local manifest whose leading bytes match ``prefix``.
+
+    Used when a device's schema reply cut a kit's checksum short. Returns the
+    checksum as a signed 32-bit int (as read off the wire), or None when no
+    manifest, or more than one, matches.
+    """
+    manifest_dir = os.path.join(app_state.SEDONA_HOME, "manifests", kit_name)
+    if not os.path.isdir(manifest_dir):
+        return None
+    want = prefix.hex()
+    matches = set()
+    for f in os.listdir(manifest_dir):
+        if f.startswith(f"{kit_name}-") and f.endswith(".xml"):
+            hex_part = f[len(kit_name) + 1:-4].zfill(8).lower()
+            if hex_part.startswith(want):
+                matches.add(hex_part)
+    if len(matches) != 1:
+        return None
+    value = int(matches.pop(), 16)
+    return value - (1 << 32) if value & 0x80000000 else value
+
+
 def find_manifest(kit_name, checksum):
     """Returns the parsed ET root of a kit manifest, or None."""
-    checksum_hex = f"{checksum:08x}" if isinstance(checksum, int) else checksum
+    checksum_hex = f"{checksum & 0xFFFFFFFF:08x}" if isinstance(checksum, int) else checksum
     path = find_manifest_path(kit_name, checksum_hex)
     if path is None:
         return None

sox.py

@@ -95,6 +95,13 @@ class SoxMsg:
         self._pos += 1  # null terminator
         return s
 
+    def try_str(self):
+        """Like str(), but returns None instead of overrunning a truncated buffer."""
+        end = self._buf.find(0, self._pos, self._size)
+        if end < 0:
+            return None
+        return self.str()
+
     def read_bytes(self, n):
         data = bytes(self._buf[self._pos:self._pos + n])
         self._pos += n
@@ -363,6 +370,10 @@ class SoxClient:
         # each slot dict: {name, id, type_name, is_config, is_runtime, is_action, is_operator, is_as_str}
         self.slot_registry = {}
 
+        # fn(kit_name, prefix_bytes) -> signed i4 checksum or None; recovers a
+        # checksum the device's reply truncated. Set externally by device_session.
+        self.checksum_resolver = None
+
         # type name map — set externally by device_session
         # maps (kit_id, type_id) -> "kit::TypeName" for proper type resolution
         self.type_name_map = {}
@@ -413,10 +424,28 @@ class SoxClient:
         res.check_response('V')
         num = res.u1()
         parts = []
+        # The device builds this reply in a fixed 512-byte packet and silently
+        # drops whatever does not fit, so a device with many kits can cut the
+        # list short — sometimes mid-checksum of the last kit.
         for _ in range(num):
-            name = res.str()
+            name = res.try_str()
+            if name is None:
+                break
+            if res.remaining >= 4:
                 checksum = res.i4()
+            else:
+                prefix = res.read_bytes(res.remaining)
+                checksum = (self.checksum_resolver(name, prefix)
+                            if self.checksum_resolver else None)
+                if checksum is None:
+                    break
             parts.append(KitPart(name, checksum))
+        if len(parts) < num:
+            raise SoxException(
+                f"The device lists {num} kits but its reply was cut off after "
+                f"{len(parts)} — the list does not fit in the device's 512-byte "
+                f"Sox packet. Remove unused kits from the device's scode, or "
+                f"shorten kit names.")
         self.schema_parts = parts
         return parts
 
@@ -427,17 +456,21 @@ class SoxClient:
         req = _prepare_request(ord('y'))
         res = self.request(req)
         res.check_response('Y')
-        platform_id = res.str()
-        scode_flags = res.u1()
+        # Same 512-byte limit as read_schema: whatever is cut off is left
+        # blank rather than failing the connect.
+        platform_id = res.try_str() or ""
+        scode_flags = res.u1() if res.remaining > 0 else 0
         kit_versions = []
         for p in parts:
-            ver = res.str()
-            kit_versions.append((p.name, p.checksum, ver))
+            ver = res.try_str()
+            kit_versions.append((p.name, p.checksum, ver or ""))
         vi = VersionInfo(platform_id, scode_flags, kit_versions)
-        prop_num = res.u1()
+        prop_num = res.u1() if res.remaining > 0 else 0
         for _ in range(prop_num):
-            k = res.str()
-            v = res.str()
+            k = res.try_str()
+            v = res.try_str()
+            if k is None or v is None:
+                break
             vi.props[k] = v
         self.version_info = vi
         return vi