Native capture decoder parity, and the tests that hid the gaps

native/csrc/catalog/reader.cpp

@@ -372,10 +372,28 @@ std::vector<std::string> parse_tsv_tuple(const std::string& text) {
         // turned `packs/a\b.dmi-pack` into a backspace. Grouped columns
         // are the opposite case -- plain TSV fields, unescaped where the
         // row is flattened, never here.
+        //
+        // The cases below are the escapes ClickHouse actually EMITS when it
+        // renders a string inside a tuple, measured with
+        // `SELECT tuple(concat('block', char(N), 'resid')) FORMAT TSV` over
+        // the control range: \0 \b \t \n \f \r \' \\ and nothing else. In
+        // particular 0x07 and 0x0B travel RAW here, so they need no case --
+        // this set is deliberately NOT sql_quote's, which also writes \a and
+        // \v, and the footer decoder (unquote_sql) covers that one.
+        //
+        // Both sets have to DECODE to the same bytes even though they differ,
+        // because hydrate binds the two field by field. \b and \f were
+        // missing and fell to the default, so they decoded to the letters "b"
+        // and "f" while the footer side decoded them to 0x08 and 0x0C: a
+        // hook_name carrying either (legal -- _validate_text asks only for
+        // non-empty UTF-8 under 512 bytes) staged, uploaded and indexed and
+        // then failed with "does not match the pack footer: field 12".
         switch (next) {
           case 'n': current.push_back('\n'); break;
           case 't': current.push_back('\t'); break;
           case 'r': current.push_back('\r'); break;
+          case 'b': current.push_back('\b'); break;
+          case 'f': current.push_back('\f'); break;
           case '0': current.push_back('\0'); break;
           default: current.push_back(next); break;  // \' and \\ included
         }

native/csrc/common/json.cpp

@@ -2,8 +2,19 @@
 
 namespace dmi_common {
 
-std::string Unescape(const std::string& text, size_t& q) {
+std::string Unescape(const std::string& text, size_t& q, bool* ok) {
   std::string raw;
+  // An escape with no code point behind it. The decoder stays TOTAL -- it has
+  // to, since most callers read fields whose contents nothing validates -- so
+  // it emits U+FFFD and lets `ok` carry the refusal to the callers that have
+  // an error channel. U+FFFD matters on its own: whatever a caller does next,
+  // no sequence that is not valid UTF-8 leaves this function, which is what
+  // put CESU-8 in a pack. `ok` is only ever cleared, never set, so one flag
+  // can latch a whole object's worth of fields.
+  const auto reject = [&ok, &raw] {
+    if (ok != nullptr) *ok = false;
+    raw.append("\xEF\xBF\xBD");
+  };
   while (q < text.size() && text[q] != '"') {
     if (text[q] == '\\' && q + 1 < text.size()) {
       ++q;
@@ -15,36 +26,80 @@ std::string Unescape(const std::string& text, size_t& q) {
         case 'f': raw.push_back('\f'); ++q; break;
         case 'u': {
           // Four hex digits at text[at+1..at+4], or -1 when they are not
-          // there (a truncated escape at the end of the text).
+          // there: a truncated escape at the end of the text, or a character
+          // that is not a hex digit at all. The digits were NOT checked, so
+          // `h <= '9' ? h - '0' : (h | 0x20) - 'a' + 10` turned \uZZZZ into
+          // U+25553 and the field decoded to F0 A5 95 93, where json.loads
+          // raises "Invalid \uXXXX escape" and the record never lands.
           const auto hex4 = [&text](size_t at) -> long {
             if (at + 4 >= text.size()) return -1;
             unsigned v = 0;
             for (int k = 1; k <= 4; ++k) {
               const char h = text[at + k];
-              v = v * 16 + (h <= '9' ? h - '0' : (h | 0x20) - 'a' + 10);
+              unsigned digit = 0;
+              if (h >= '0' && h <= '9') {
+                digit = static_cast<unsigned>(h - '0');
+              } else if ((h | 0x20) >= 'a' && (h | 0x20) <= 'f') {
+                digit = static_cast<unsigned>((h | 0x20) - 'a' + 10);
+              } else {
+                return -1;
+              }
+              v = v * 16 + digit;
             }
             return static_cast<long>(v);
           };
           const long first = hex4(q);
-          unsigned v = first < 0 ? 0u : static_cast<unsigned>(first);
+          // Advance past the four digits only when they ARE four hex
+          // digits. On failure the "digits" can include the field's real
+          // closing quote -- `"x\u12"` -- and stepping over them stepped
+          // over that quote too, so the loop kept copying the FOLLOWING
+          // JSON text (comma, separator, the next field's key) into this
+          // field's value until the next quote; a truncated escape at the
+          // end of the buffer left q strictly past text.size(). Consuming
+          // only the `u` keeps the decoder on the string: whatever failed
+          // hex4 is copied as literal text, the real closing quote
+          // terminates the value, and the latch still carries the refusal.
+          if (first < 0) {
+            ++q;
+            reject();
+            break;
+          }
           q += 5;
+          unsigned v = static_cast<unsigned>(first);
           // json.dumps (ensure_ascii, the default) writes a code point
           // outside the BMP as a UTF-16 surrogate PAIR: U+1F600 is
           // \ud83d\ude00. The two halves are one character, so they have to
           // be combined into one code point BEFORE encoding: encoding each
           // half on its own produced six bytes (ED A0 BD ED B8 80) that are
           // not UTF-8 at all, and the identifier reached the catalog
           // corrupted (reproduced with hook_name "block" + U+1F600: Python's reader
-          // then failed to decode byte 0xED). A lone surrogate keeps the
-          // old three-byte form -- there is nothing to combine it with.
-          if (v >= 0xD800 && v <= 0xDBFF && q + 1 < text.size() &&
-              text[q] == '\\' && text[q + 1] == 'u') {
-            const long second = hex4(q + 1);
-            if (second >= 0xDC00 && second <= 0xDFFF) {
-              v = 0x10000 + ((v - 0xD800) << 10) +
-                  (static_cast<unsigned>(second) - 0xDC00);
-              q += 6;
+          // then failed to decode byte 0xED).
+          //
+          // A surrogate with no partner has NOTHING to combine with, and
+          // keeping its three-byte form is the same corruption in the small:
+          // \ud83d decoded to ED A0 BD, which ValidText admits (it checks
+          // lead/continuation byte SHAPE, not the surrogate range) and the
+          // record was packed and inserted. The oracle refuses it --
+          // CaptureMetadata's _validate_text calls .encode("utf-8"), which
+          // raises UnicodeEncodeError on a lone surrogate -- so it is
+          // refused here rather than encoded.
+          if (v >= 0xD800 && v <= 0xDBFF) {
+            const long second =
+                (q + 1 < text.size() && text[q] == '\\' && text[q + 1] == 'u')
+                    ? hex4(q + 1)
+                    : -1;
+            if (second < 0xDC00 || second > 0xDFFF) {
+              reject();
+              break;
             }
+            v = 0x10000 + ((v - 0xD800) << 10) +
+                (static_cast<unsigned>(second) - 0xDC00);
+            q += 6;
+          } else if (v >= 0xDC00 && v <= 0xDFFF) {
+            // A trailing surrogate reached on its own: the leading half it
+            // belongs to is not there, so this one has no partner either.
+            reject();
+            break;
           }
           if (v < 0x80) {
             raw.push_back(static_cast<char>(v));
@@ -73,16 +128,19 @@ std::string Unescape(const std::string& text, size_t& q) {
 }
 
 std::string FindString(const std::string& text, const std::string& key,
-                       size_t from) {
+                       size_t from, bool* ok) {
   for (const char* sep : {": \"", ":\""}) {
     // NOTE: sep includes the opening quote; search for key+sep.
     const std::string needle = "\"" + key + "\"" + sep;
     const size_t at = text.find(needle, from);
     if (at == std::string::npos) continue;
     size_t q = at + needle.size() - 1;  // back up onto the opening quote
     ++q;
-    return Unescape(text, q);
+    return Unescape(text, q, ok);
   }
+  // An ABSENT key is not a decode failure, so `ok` is left alone: a caller
+  // latching one flag over a whole object must not be told the field it did
+  // not send was malformed.
   return "";
 }
 

native/csrc/common/json.h

@@ -16,12 +16,24 @@ namespace dmi_common {
 // JSON string escape sequences back to raw bytes; \uXXXX to UTF-8, with a
 // UTF-16 surrogate pair combined into its one non-BMP code point.
 // `q` must be positioned just after the opening quote; returns with `q` on
-// the closing quote.
-std::string Unescape(const std::string& text, size_t& q);
+// the closing quote (at text.size() for an unterminated text, never past
+// it -- a malformed escape does not advance over the quote or the end).
+//
+// `ok`, when given, is CLEARED (never set) for an escape that has no code
+// point behind it: a \uXXXX whose digits are not hex, and a surrogate with
+// no partner. Both are things the oracle refuses -- json.loads raises
+// "Invalid \uXXXX escape" for the first, and _validate_text's
+// .encode("utf-8") raises UnicodeEncodeError for the second -- and this
+// decoder has no way to refuse on its own. The returned text stays valid
+// UTF-8 either way (U+FFFD stands in), so a caller that cannot check still
+// never receives the CESU-8 that reached a pack. Initialise it to true and
+// pass the same flag to every field of one object to latch the whole parse.
+std::string Unescape(const std::string& text, size_t& q, bool* ok = nullptr);
 
 // First string value for `key` anywhere in `text` (either separator style).
+// `ok` is as for Unescape; an absent key leaves it alone.
 std::string FindString(const std::string& text, const std::string& key,
-                       size_t from = 0);
+                       size_t from = 0, bool* ok = nullptr);
 
 // Outcome of a bounded integer scan.
 enum class IntFind {

native/csrc/sink/conformance_sink.cpp

@@ -13,11 +13,14 @@
 //   {"op":"snapshot"} -> {"ok":true,"snapshot":{...}}
 //   {"op":"object_key","tenant_id":"...","session_id":"...","producer_rank":N,
 //    "captured_at_ns":N,"pack_id":"..."} -> {"ok":true,"object_key":"..."}
+//   {"op":"parse_metadata","metadata_json":"..."}
+//     -> {"ok":true,"hook_name_hex":"..."}
 // Errors: {"ok":false,"what":"..."}.
 
 #include "pack_sink.h"
 
 #include <cstdint>
+#include <cstdio>
 #include <iostream>
 #include <memory>
 #include <string>
@@ -44,6 +47,14 @@ namespace {
 // are the driver's remaining decodes.
 std::string g_out_of_range;
 
+// Latched the same way, for a metadata string the JSON decoder could not turn
+// into code points: non-hex \uXXXX digits, or a surrogate with no partner.
+// The oracle refuses both (json.loads raises "Invalid \uXXXX escape";
+// _validate_text's .encode("utf-8") raises UnicodeEncodeError), and this op is
+// the native side of CaptureMetadata.from_mapping, so it has to as well --
+// SubmitRow's own parse already does (record_row.cpp).
+bool g_bad_text = false;
+
 // A field the oracle types as SIGNED must pass IntDomain::kSigned: the
 // union's two's-complement bit pattern is a legal-looking negative to such a
 // field, and 18446744073709551615 arrived at layer_number as -1 -- its own
@@ -62,20 +73,25 @@ int64_t Integer(const std::string& text, const char* key,
 
 dmi_pack::RecordMetadata ParseMetadata(const std::string& obj) {
   dmi_pack::RecordMetadata m;
-  m.capture_id = jc::FindString(obj, "capture_id");
-  m.tenant_id = jc::FindString(obj, "tenant_id");
-  m.experiment_id = jc::FindString(obj, "experiment_id");
-  m.run_id = jc::FindString(obj, "run_id");
-  m.session_id = jc::FindString(obj, "session_id");
-  m.request_id = jc::FindString(obj, "request_id");
-  m.sequence_id = jc::FindString(obj, "sequence_id");
-  m.model_id = jc::FindString(obj, "model_id");
-  m.model_revision = jc::FindString(obj, "model_revision");
+  bool text_ok = true;
+  const auto text_field = [&obj, &text_ok](const char* name) {
+    return jc::FindString(obj, name, 0, &text_ok);
+  };
+  m.capture_id = text_field("capture_id");
+  m.tenant_id = text_field("tenant_id");
+  m.experiment_id = text_field("experiment_id");
+  m.run_id = text_field("run_id");
+  m.session_id = text_field("session_id");
+  m.request_id = text_field("request_id");
+  m.sequence_id = text_field("sequence_id");
+  m.model_id = text_field("model_id");
+  m.model_revision = text_field("model_revision");
   if (!jc::FindNull(obj, "adapter_revision")) {
-    m.adapter_revision = jc::FindString(obj, "adapter_revision");
+    m.adapter_revision = text_field("adapter_revision");
   }
-  m.capture_policy_version = jc::FindString(obj, "capture_policy_version");
-  m.hook_name = jc::FindString(obj, "hook_name");
+  m.capture_policy_version = text_field("capture_policy_version");
+  m.hook_name = text_field("hook_name");
+  if (!text_ok) g_bad_text = true;
   m.layer_number = Integer(obj, "layer_number", jc::IntDomain::kSigned);
   m.producer_rank = static_cast<uint64_t>(Integer(obj, "producer_rank"));
   m.step_number = static_cast<uint64_t>(Integer(obj, "step_number"));
@@ -161,8 +177,18 @@ int main() {
                    &out);
     std::cout << out << "}\n";
   };
+  const auto refuse_bad_text = [] {
+    std::string out = "{\"ok\":false,\"what\":";
+    // Same wording as ParseMetadataJson's refusal: the decoder replaces a
+    // malformed \uXXXX or an unpaired surrogate with U+FFFD (valid UTF-8),
+    // so the failure is the escape, not the encoding.
+    jc::EscapeJson("capture metadata text has an invalid or unpaired Unicode escape",
+                   &out);
+    std::cout << out << "}\n";
+  };
   while (std::getline(std::cin, line)) {
     g_out_of_range.clear();
+    g_bad_text = false;
     const std::string op = jc::FindString(line, "op");
     if (op == "open") {
       dmi_sink::SinkConfig config;
@@ -233,6 +259,38 @@ int main() {
       std::cout << out << "}\n";
       continue;
     }
+    if (op == "parse_metadata") {
+      // Sessionless like `object_key`: the row path's metadata decoder alone,
+      // handing hook_name back as RAW BYTES in hex.
+      //
+      // Why the bytes, and why not read them off a staged pack: the footer
+      // writer decodes each UTF-8 sequence and re-emits \uXXXX
+      // (pack_builder.cpp, EncodeJsonString), which is an exact inverse of
+      // this decoder. A broken CESU-8 surrogate pair written by the decoder
+      // comes back out of the footer as a correct 😀 that
+      // json.loads recombines, so every assertion made on a hook_name read
+      // back through the pack holds whether or not the decoder combines --
+      // the surrogate tests passed with the combine branch disabled. These
+      // bytes never meet that serializer, so they can tell the two apart.
+      dmi_pack::RecordMetadata metadata;
+      std::string error;
+      if (!dmi_sink::ParseMetadataJson(jc::FindString(line, "metadata_json"),
+                                       &metadata, &error)) {
+        std::string out = "{\"ok\":false,\"what\":";
+        jc::EscapeJson(error, &out);
+        std::cout << out << "}\n";
+        continue;
+      }
+      std::string hex;
+      for (const char byte : metadata.hook_name) {
+        char buf[3];
+        std::snprintf(buf, sizeof(buf), "%02x",
+                      static_cast<unsigned>(static_cast<unsigned char>(byte)));
+        hex.append(buf, 2);
+      }
+      std::cout << "{\"ok\":true,\"hook_name_hex\":\"" << hex << "\"}\n";
+      continue;
+    }
     if (!sink) {
       std::cout << "{\"ok\":false,\"what\":\"sink is not open\"}\n";
       continue;
@@ -247,6 +305,10 @@ int main() {
         refuse_out_of_range();
         continue;
       }
+      if (g_bad_text) {
+        refuse_bad_text();
+        continue;
+      }
       std::vector<uint8_t> payload;
       jc::DecodeBase64(jc::FindString(line, "payload_b64"), &payload);
       const dmi_sink::Admission admission =

native/csrc/sink/record_row.cpp

@@ -37,22 +37,42 @@ bool ParseMetadataJson(const std::string& text,
     return false;
   };
   // The object may carry surrounding whitespace; Find* scans for keys.
-  out->capture_id = jc::FindString(text, "capture_id");
-  out->tenant_id = jc::FindString(text, "tenant_id");
-  out->experiment_id = jc::FindString(text, "experiment_id");
-  out->run_id = jc::FindString(text, "run_id");
-  out->session_id = jc::FindString(text, "session_id");
-  out->request_id = jc::FindString(text, "request_id");
-  out->sequence_id = jc::FindString(text, "sequence_id");
-  out->model_id = jc::FindString(text, "model_id");
-  out->model_revision = jc::FindString(text, "model_revision");
+  //
+  // One latch across every identifier: an escape the decoder cannot turn into
+  // a code point (non-hex \uXXXX digits, or a surrogate with no partner) is a
+  // refusal, not a value. Both are things the oracle refuses -- json.loads
+  // raises "Invalid \uXXXX escape" for the first and _validate_text's
+  // .encode("utf-8") raises UnicodeEncodeError for the second -- while native
+  // encoded them as U+25553 and as three-byte CESU-8, which ValidText admits
+  // (it checks byte SHAPE, not the surrogate range) and the record was packed
+  // and inserted.
+  bool text_ok = true;
+  const auto text_field = [&text, &text_ok](const char* name) {
+    return jc::FindString(text, name, 0, &text_ok);
+  };
+  out->capture_id = text_field("capture_id");
+  out->tenant_id = text_field("tenant_id");
+  out->experiment_id = text_field("experiment_id");
+  out->run_id = text_field("run_id");
+  out->session_id = text_field("session_id");
+  out->request_id = text_field("request_id");
+  out->sequence_id = text_field("sequence_id");
+  out->model_id = text_field("model_id");
+  out->model_revision = text_field("model_revision");
   if (!jc::FindNull(text, "adapter_revision")) {
-    out->adapter_revision = jc::FindString(text, "adapter_revision");
+    out->adapter_revision = text_field("adapter_revision");
   } else {
     out->adapter_revision.reset();
   }
-  out->capture_policy_version = jc::FindString(text, "capture_policy_version");
-  out->hook_name = jc::FindString(text, "hook_name");
+  out->capture_policy_version = text_field("capture_policy_version");
+  out->hook_name = text_field("hook_name");
+  if (!text_ok) {
+    // The decoder replaces a malformed \uXXXX (non-hex or truncated digits)
+    // and an unpaired surrogate with U+FFFD, which IS valid UTF-8, so "not
+    // encodable UTF-8" named the wrong failure. These are the only two ways
+    // `ok` is cleared; name them.
+    return fail("capture metadata text has an invalid or unpaired Unicode escape");
+  }
   // Presence first: layer_number == -1 is legal (logits-style captures),
   // so FindInt's missing sentinel cannot stand in for absence.
   for (const char* name :