DIAGNOSTIC: log VCR body mismatches + per-episode body hashes

Temporary observability boost so we can root-cause why
``test_image_edits.py`` async parametrizes still record fresh
episodes on every CI run even though the multipart boundary is now
pinned (sync parametrizes cache cleanly as VCR HIT). The matcher
currently raises ``AssertionError("request bodies differ")`` with
zero context, so we cannot tell whether the live body genuinely
varies, the matcher is comparing a bytes object to a stream object,
or the normalizer is silently skipping the body because it is not
bytes/str.

Three logs added; the first two are worth keeping permanently, the
third is intended to be reverted after the diagnosis lands:

1. ``_safe_body_matcher`` now emits a structured stderr block on
   mismatch (type of each side, length, SHA-256, first divergent
   byte offset, ±100-byte window). Always-on -- mismatches are
   signal, not noise, and the existing per-test verdict already
   logs once per test. PERMANENT.

2. ``_normalize_multipart_boundary`` now logs to stderr when the
   body type is not bytes/bytearray/str -- the silent ``else:
   return`` branch was masking exactly the case we suspect is
   firing on async (httpx ``MultipartStream`` handed to vcrpy
   before the body is read). PERMANENT.

3. ``_RedisPersister.save_cassette`` now logs every episode's body
   SHA-256, length, and 120-byte preview at save time. This lets
   two consecutive CI runs be diffed: if the same test records a
   different hash run-to-run, the live body genuinely varies; if
   both runs record the same hash but the matcher still misses, the
   bug is in the matcher itself. TEMPORARY -- revert once the
   async variance is identified and fixed.

Once a single ``image_gen_testing`` CI run produces these logs,
revert this commit (or just the persister hash block) with a force
push so the cassette save path is not noisy in steady-state.

tests/_vcr_conftest_common.py

@@ -243,6 +243,13 @@ def _safe_body_matcher(r1, r2) -> None:
     (e.g. the Bedrock batch S3 PUT) before it can return "no match".
     This matcher is strictly more conservative — the only equivalence
     it gives up vs. the default is "JSON key order doesn't matter".
+
+    On mismatch, emits a structured diagnostic to stderr (type of each
+    body, length, SHA-256, first divergent offset, ±100-byte window).
+    Without this, vcrpy returns "request bodies differ" with zero
+    context, and bugs where the live request body is an unbytes-like
+    object (e.g. an httpx ``MultipartStream`` for async requests) look
+    indistinguishable from genuine content drift.
     """
     body1 = getattr(r1, "body", None)
     body2 = getattr(r2, "body", None)
@@ -262,9 +269,64 @@ def _safe_body_matcher(r1, r2) -> None:
     n2 = _to_bytes(body2)
     if n1 is not None and n2 is not None and n1 == n2:
         return
+    _emit_body_mismatch_diagnostic(r1, r2, body1, body2, n1, n2)
     raise AssertionError("request bodies differ")
 
 
+def _emit_body_mismatch_diagnostic(r1, r2, body1, body2, n1, n2) -> None:
+    """Dump enough info to a single stderr block to root-cause why two
+    requests that look semantically identical failed the body matcher.
+
+    Always-on (matcher mismatches are signal, not noise): the volume
+    is bounded by the number of stored episodes a live request is
+    compared against, and we already log a per-test verdict line for
+    every test.
+    """
+
+    def _describe(label, raw, asbytes):
+        t = type(raw).__name__
+        if asbytes is None:
+            return (
+                f"  {label}: type={t!r} length=unknown sha256=N/A "
+                f"(body could not be coerced to bytes)"
+            )
+        length = len(asbytes)
+        digest = hashlib.sha256(asbytes).hexdigest()
+        preview = asbytes[:120]
+        return (
+            f"  {label}: type={t!r} length={length} sha256={digest} "
+            f"preview={preview!r}"
+        )
+
+    method_a = getattr(r1, "method", "?")
+    method_b = getattr(r2, "method", "?")
+    url_a = getattr(r1, "uri", getattr(r1, "url", "?"))
+    url_b = getattr(r2, "uri", getattr(r2, "url", "?"))
+    lines = [
+        "[vcr-safe-body-matcher] request body mismatch",
+        f"  request[a]: {method_a} {url_a}",
+        f"  request[b]: {method_b} {url_b}",
+        _describe("body[a]", body1, n1),
+        _describe("body[b]", body2, n2),
+    ]
+    if n1 is not None and n2 is not None and n1 != n2:
+        # Find the first divergent byte offset and dump a ±100 window
+        # around it so the human reading the CI log can see at a glance
+        # whether the variance is a UUID, a timestamp, a random multipart
+        # boundary, or something else.
+        offset = next(
+            (i for i in range(min(len(n1), len(n2))) if n1[i] != n2[i]),
+            min(len(n1), len(n2)),
+        )
+        start = max(0, offset - 100)
+        end_a = min(len(n1), offset + 100)
+        end_b = min(len(n2), offset + 100)
+        lines.append(f"  first divergent byte offset: {offset}")
+        lines.append(f"  window[a] @ {start}..{end_a}: {n1[start:end_a]!r}")
+        lines.append(f"  window[b] @ {start}..{end_b}: {n2[start:end_b]!r}")
+    sys.stderr.write("\n".join(lines) + "\n")
+
+
 def _iter_header_values(headers, name: str):
     if headers is None:
         return
@@ -409,6 +471,20 @@ def _normalize_multipart_boundary(request) -> None:
     elif isinstance(body, str):
         new_body = body.replace(current_boundary, VCR_FIXED_MULTIPART_BOUNDARY)
     else:
+        # The body is something other than bytes/bytearray/str -- most
+        # likely an httpx ``MultipartStream`` or an aiter chunked stream
+        # we cannot rewrite in place. Log it so a body-matcher miss on a
+        # multipart request can be correlated with "normalizer skipped
+        # because body type was X". The header was still rewritten
+        # above, so the recorded Content-Type stays stable; only the
+        # body bytes carry the random boundary verbatim.
+        sys.stderr.write(
+            f"[vcr-multipart-normalize] body normalization SKIPPED: "
+            f"body type {type(body).__name__!r} is not bytes/bytearray/str. "
+            f"content-type={content_type_value!r}. "
+            f"Recorded body will retain the random boundary substring "
+            f"and the safe_body matcher will miss on the next run.\n"
+        )
         return
 
     try:

tests/_vcr_redis_persister.py

@@ -168,6 +168,7 @@ def make_redis_persister(
             key = redis_key_for(cassette_path)
             passed = _passed_by_cassette_key.pop(key, True)
             episode_count = len(cassette_dict.get("requests", []) or [])
+            _maybe_log_episode_body_hashes(key, cassette_dict)
             if episode_count > MAX_EPISODES_PER_CASSETTE:
                 _log.warning(
                     "VCR redis save refused for %s; cassette has %d episodes "
@@ -210,6 +211,54 @@ def make_redis_persister(
     return _RedisPersister
 
 
+# TEMP DIAGNOSTIC -- intended to be reverted once the async image-edit
+# cassette variance is root-caused. Logs a per-episode body SHA-256
+# at save time so two consecutive CI runs can be diffed: if the same
+# test records ``sha=abc`` on run 1 and ``sha=def`` on run 2, the live
+# request body genuinely varies; if both runs record the same hash
+# but the matcher still misses, the bug is in the matcher (e.g. it is
+# comparing a bytes object to a stream object). Always-on for any
+# session that loads this persister -- ungated because we are
+# specifically trying to capture data from CI right now.
+def _maybe_log_episode_body_hashes(key: str, cassette_dict) -> None:
+    import hashlib
+
+    requests = cassette_dict.get("requests", []) or []
+    if not requests:
+        return
+    for i, req in enumerate(requests):
+        body = getattr(req, "body", None)
+        if body is None:
+            body_bytes = b""
+        elif isinstance(body, (bytes, bytearray)):
+            body_bytes = bytes(body)
+        elif isinstance(body, str):
+            body_bytes = body.encode("utf-8")
+        else:
+            _log.warning(
+                "[vcr-episode-body-hash] %s episode[%d]: body type=%r is "
+                "not bytes/bytearray/str -- cannot hash. This is the "
+                "smoking gun for matcher-side bugs on async multipart.",
+                key,
+                i,
+                type(body).__name__,
+            )
+            continue
+        method = getattr(req, "method", "?")
+        uri = getattr(req, "uri", getattr(req, "url", "?"))
+        _log.warning(
+            "[vcr-episode-body-hash] %s episode[%d] %s %s body sha256=%s "
+            "len=%d preview=%r",
+            key,
+            i,
+            method,
+            uri,
+            hashlib.sha256(body_bytes).hexdigest(),
+            len(body_bytes),
+            body_bytes[:120],
+        )
+
+
 def filter_non_2xx_response(response):
     if not isinstance(response, dict):
         return response