cold_clone: skip-not-crash on hash mismatch; utf-8 for str affinity

Some chunks in real corpora have content stored as decoded text (str)
not bytes, with Unicode codepoints beyond latin-1. Encode as utf-8 so
the conversion always succeeds; if the resulting bytes do not hash to
the row's leaf_hash, skip that chunk (no blob uploaded, content kept
local) rather than aborting the whole shard's snapshot.

arborist/cold_clone.py

@@ -166,13 +166,14 @@ def _strip_chunk_content_and_dump_blobs(
         ]
         BATCH = 1000
 
-        def _upload(triple: tuple[int, str, bytes]) -> int:
+        def _upload(triple: tuple[int, str, bytes]):
             chunk_id, leaf_hash, body = triple
             if hash_leaf(body).hex() != leaf_hash:
-                raise ValueError(
-                    f"chunk {chunk_id}: leaf_hash {leaf_hash[:12]} "
-                    f"does not match content"
-                )
+                # Inconsistent row (likely str-affinity content that
+                # wasn't the original zstd bytes). Skip rather than
+                # crash the whole shard; the local row keeps its
+                # content so the producer copy stays self-consistent.
+                return None
             key = f"{BLOB_PREFIX}/{leaf_hash[:2]}/{leaf_hash[2:]}"
             backend.put(key, body, content_type="application/zstd")
             return chunk_id
@@ -187,22 +188,23 @@ def _strip_chunk_content_and_dump_blobs(
             ).fetchall()
             # SQLite type affinity: a `BLOB` column can come back as
             # bytes / bytearray / memoryview, OR as `str` if the value
-            # was originally INSERTed as text. latin-1 preserves
-            # arbitrary byte values 1:1 (codepoints 0..255) so the
-            # hash check below still matches.
+            # was originally INSERTed as text. UTF-8 covers all Unicode;
+            # if the resulting bytes don't hash-check the row had its
+            # affinity messed up at ingest and we skip it (leave content
+            # intact, no blob uploaded — better than corrupting the
+            # commitment).
             def _as_bytes(v):
                 if isinstance(v, (bytes, bytearray)):
                     return bytes(v)
                 if isinstance(v, memoryview):
                     return v.tobytes()
-                return v.encode("latin-1")
+                return v.encode("utf-8")
             triples = [
                 (cid, lh, _as_bytes(c)) for cid, lh, c in rows if c is not None
             ]
             with ThreadPoolExecutor(max_workers=upload_workers) as ex:
-                for cid in ex.map(_upload, triples):
-                    pass
-            uploaded_ids = [t[0] for t in triples]
+                results = list(ex.map(_upload, triples))
+            uploaded_ids = [r for r in results if r is not None]
             with conn:
                 conn.execute(
                     f"UPDATE chunks SET content = NULL "